final CommandInfo oldCommand = CommandInfo.parseFrom(taskBuilder.getData());
final Environment oldEnvironment = oldCommand.getEnvironment();
final Map<String, String> newEnvMap = fromEnvironmentToMap(oldEnvironment);
newEnvMap.put("KAFKA_HEAP_OPTS", getKafkaHeapOpts(brokerConfig.getHeap()));
final CommandInfo.Builder newCommandBuilder = CommandInfo.newBuilder(oldCommand);
<BUG>newCommandBuilder.clearEnvironment();
final List<Variable> newEnvironmentVariables = EnvironmentBuilder.createEnvironment(newEnvMap);
newCommandBuilder.setEnvironment(Environment.newBuilder().addAllVariables(newEnvironmentVariables));</BUG>
taskBuilder.clearData();
taskBuilder.setData(newCommandBuilder.build().toByteString());

final CommandInfo oldCommand = CommandInfo.parseFrom(taskBuilder.getData());
final Environment oldEnvironment = oldCommand.getEnvironment();
final Map<String, String> newEnvMap = fromEnvironmentToMap(oldEnvironment);
newEnvMap.put("KAFKA_HEAP_OPTS", getKafkaHeapOpts(brokerConfig.getHeap()));
final CommandInfo.Builder newCommandBuilder = CommandInfo.newBuilder(oldCommand);
newCommandBuilder.clearEnvironment();
newCommandBuilder.setEnvironment(environment(newEnvMap));
taskBuilder.clearData();
taskBuilder.setData(newCommandBuilder.build().toByteString());
