[FLINK-13986][core][config] Remove legacy config option TaskManagerOp…

…tions#TASK_MANAGER_HEAP_MEMORY.
spotify · Dec 3, 2019 · ba1d711 · ba1d711
1 parent d24c324
commit ba1d711
Show file tree

Hide file tree

Showing 18 changed files with 21 additions and 32 deletions.
diff --git a/docs/ops/config.md b/docs/ops/config.md
@@ -92,7 +92,7 @@ Configuration options to control Flink's restart behaviour in case of job failur
 
 {% include generated/task_manager_configuration.html %}
 
-For *batch* jobs Flink allocates a fraction of 0.7 of the free memory (total memory configured via taskmanager.heap.size minus memory used for network buffers) for its managed memory. Managed memory helps Flink to run the batch operators efficiently. It prevents OutOfMemoryExceptions because Flink knows how much memory it can use to execute operations. If Flink runs out of managed memory, it utilizes disk space. Using managed memory, some operations can be performed directly on the raw data without having to deserialize the data to convert it into Java objects. All in all, managed memory improves the robustness and speed of the system.
+For *batch* jobs Flink allocates a fraction of 0.7 of the free memory (total memory configured via taskmanager.memory.total-process.size minus memory used for network buffers) for its managed memory. Managed memory helps Flink to run the batch operators efficiently. It prevents OutOfMemoryExceptions because Flink knows how much memory it can use to execute operations. If Flink runs out of managed memory, it utilizes disk space. Using managed memory, some operations can be performed directly on the raw data without having to deserialize the data to convert it into Java objects. All in all, managed memory improves the robustness and speed of the system.
 
 The default fraction for managed memory can be adjusted using the taskmanager.memory.fraction parameter. An absolute value may be set using taskmanager.memory.size (overrides the fraction parameter). If desired, the managed memory may be allocated outside the JVM heap. This may improve performance in setups with large memory sizes.
 

diff --git a/docs/ops/config.zh.md b/docs/ops/config.zh.md
@@ -91,7 +91,7 @@ Configuration options to control Flink's restart behaviour in case of job failur
 
 {% include generated/task_manager_configuration.html %}
 
-For *batch* jobs Flink allocates a fraction of 0.7 of the free memory (total memory configured via taskmanager.heap.size minus memory used for network buffers) for its managed memory. Managed memory helps Flink to run the batch operators efficiently. It prevents OutOfMemoryExceptions because Flink knows how much memory it can use to execute operations. If Flink runs out of managed memory, it utilizes disk space. Using managed memory, some operations can be performed directly on the raw data without having to deserialize the data to convert it into Java objects. All in all, managed memory improves the robustness and speed of the system.
+For *batch* jobs Flink allocates a fraction of 0.7 of the free memory (total memory configured via taskmanager.memory.total-process.size minus memory used for network buffers) for its managed memory. Managed memory helps Flink to run the batch operators efficiently. It prevents OutOfMemoryExceptions because Flink knows how much memory it can use to execute operations. If Flink runs out of managed memory, it utilizes disk space. Using managed memory, some operations can be performed directly on the raw data without having to deserialize the data to convert it into Java objects. All in all, managed memory improves the robustness and speed of the system.
 
 The default fraction for managed memory can be adjusted using the taskmanager.memory.fraction parameter. An absolute value may be set using taskmanager.memory.size (overrides the fraction parameter). If desired, the managed memory may be allocated outside the JVM heap. This may improve performance in setups with large memory sizes.
 

diff --git a/docs/ops/deployment/cluster_setup.md b/docs/ops/deployment/cluster_setup.md
@@ -107,7 +107,7 @@ Please see the [configuration page](../config.html) for details and additional c
 In particular,
 
  * the amount of available memory per JobManager (`jobmanager.heap.size`),
- * the amount of available memory per TaskManager (`taskmanager.heap.size`),
+ * the amount of available memory per TaskManager (`taskmanager.memory.total-process.size`),
  * the number of available CPUs per machine (`taskmanager.numberOfTaskSlots`),
  * the total number of CPUs in the cluster (`parallelism.default`) and
  * the temporary directories (`io.tmp.dirs`)

diff --git a/docs/ops/deployment/cluster_setup.zh.md b/docs/ops/deployment/cluster_setup.zh.md
@@ -107,7 +107,7 @@ Please see the [configuration page](../config.html) for details and additional c
 In particular,
 
  * the amount of available memory per JobManager (`jobmanager.heap.size`),
- * the amount of available memory per TaskManager (`taskmanager.heap.size`),
+ * the amount of available memory per TaskManager (`taskmanager.memory.total-process.size`),
  * the number of available CPUs per machine (`taskmanager.numberOfTaskSlots`),
  * the total number of CPUs in the cluster (`parallelism.default`) and
  * the temporary directories (`io.tmp.dirs`)

diff --git a/docs/ops/deployment/kubernetes.md b/docs/ops/deployment/kubernetes.md
@@ -129,7 +129,7 @@ data:
  jobmanager.rpc.port: 6123
  taskmanager.rpc.port: 6122
  jobmanager.heap.size: 1024m
- taskmanager.heap.size: 1024m
+ taskmanager.memory.total-process.size: 1024m
  log4j.properties: |+
  log4j.rootLogger=INFO, file
  log4j.logger.akka=INFO

diff --git a/docs/ops/deployment/kubernetes.zh.md b/docs/ops/deployment/kubernetes.zh.md
@@ -129,7 +129,7 @@ data:
  jobmanager.rpc.port: 6123
  taskmanager.rpc.port: 6122
  jobmanager.heap.size: 1024m
- taskmanager.heap.size: 1024m
+ taskmanager.memory.total-process.size: 1024m
  log4j.properties: |+
  log4j.rootLogger=INFO, file
  log4j.logger.akka=INFO

diff --git a/docs/ops/deployment/mesos.md b/docs/ops/deployment/mesos.md
@@ -219,7 +219,7 @@ For example:
   -Djobmanager.rpc.port=6123 \
   -Drest.port=8081 \
   -Dmesos.resourcemanager.tasks.mem=4096 \
-  -Dtaskmanager.heap.size=3500m \
+  -Dtaskmanager.memory.total-process.size=3500m \
   -Dtaskmanager.numberOfTaskSlots=2 \
   -Dparallelism.default=10
 
@@ -237,7 +237,7 @@ Here is an example configuration for Marathon:
 
  {
   "id": "flink",
-  "cmd": "$FLINK_HOME/bin/mesos-appmaster.sh -Djobmanager.heap.size=1024m -Djobmanager.rpc.port=6123 -Drest.port=8081 -Dmesos.resourcemanager.tasks.mem=1024 -Dtaskmanager.heap.size=1024m -Dtaskmanager.numberOfTaskSlots=2 -Dparallelism.default=2 -Dmesos.resourcemanager.tasks.cpus=1",
+  "cmd": "$FLINK_HOME/bin/mesos-appmaster.sh -Djobmanager.heap.size=1024m -Djobmanager.rpc.port=6123 -Drest.port=8081 -Dmesos.resourcemanager.tasks.mem=1024 -Dtaskmanager.memory.total-process.size=1024m -Dtaskmanager.numberOfTaskSlots=2 -Dparallelism.default=2 -Dmesos.resourcemanager.tasks.cpus=1",
   "cpus": 1.0,
   "mem": 1024
  }

diff --git a/docs/ops/deployment/mesos.zh.md b/docs/ops/deployment/mesos.zh.md
@@ -219,7 +219,7 @@ For example:
   -Djobmanager.rpc.port=6123 \
   -Drest.port=8081 \
   -Dmesos.resourcemanager.tasks.mem=4096 \
-  -Dtaskmanager.heap.size=3500m \
+  -Dtaskmanager.memory.total-process.size=3500m \
   -Dtaskmanager.numberOfTaskSlots=2 \
   -Dparallelism.default=10
 

diff --git a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
@@ -39,21 +39,10 @@ public class TaskManagerOptions {
  // General TaskManager Options
  // ------------------------------------------------------------------------
 
- /**
- * JVM heap size for the TaskManagers with memory size.
- */
- @Deprecated
- public static final ConfigOption<String> TASK_MANAGER_HEAP_MEMORY =
- key("taskmanager.heap.size")
- .defaultValue("1024m")
- .withDescription("JVM heap size for the TaskManagers, which are the parallel workers of" +
- " the system. On YARN setups, this value is automatically configured to the size of the TaskManager's" +
- " YARN container, minus a certain tolerance value.");
-
  /**
  * JVM heap size (in megabytes) for the TaskManagers.
  *
- * @deprecated use {@link #TASK_MANAGER_HEAP_MEMORY}
+ * @deprecated use {@link #TOTAL_PROCESS_MEMORY}
  */
  @Deprecated
  public static final ConfigOption<Integer> TASK_MANAGER_HEAP_MEMORY_MB =
@@ -255,7 +244,7 @@ public class TaskManagerOptions {
  public static final ConfigOption<String> TOTAL_PROCESS_MEMORY =
  key("taskmanager.memory.total-process.size")
  .noDefaultValue()
- .withDeprecatedKeys(TASK_MANAGER_HEAP_MEMORY.key())
+ .withDeprecatedKeys("taskmanager.heap.size")
  .withDescription("Total Process Memory size for the TaskExecutors. This includes all the memory that a"
  + " TaskExecutor consumes, consisting of Total Flink Memory, JVM Metaspace, and JVM Overhead. On"
  + " containerized setups, this should be set to the container memory.");

diff --git a/flink-dist/src/main/resources/flink-conf.yaml b/flink-dist/src/main/resources/flink-conf.yaml
@@ -44,7 +44,7 @@ jobmanager.heap.size: 1024m
 
 # The heap size for the TaskManager JVM
 
-taskmanager.heap.size: 1024m
+taskmanager.memory.total-process.size: 1024m
 
 
 # The number of task slots that each TaskManager offers. Each slot runs one parallel pipeline.

diff --git a/flink-end-to-end-tests/test-scripts/common.sh b/flink-end-to-end-tests/test-scripts/common.sh
@@ -154,7 +154,7 @@ function create_ha_config() {
  jobmanager.rpc.address: localhost
  jobmanager.rpc.port: 6123
  jobmanager.heap.size: 1024m
- taskmanager.heap.size: 1024m
+ taskmanager.memory.total-process.size: 1024m
  taskmanager.numberOfTaskSlots: ${TASK_SLOTS_PER_TM_HA}
 
  #==============================================================================

diff --git a/flink-end-to-end-tests/test-scripts/test_heavy_deployment.sh b/flink-end-to-end-tests/test-scripts/test_heavy_deployment.sh
@@ -28,7 +28,7 @@ TEST_PROGRAM_JAR=${END_TO_END_DIR}/$TEST/target/$TEST_PROGRAM_NAME.jar
 set_config_key "akka.ask.timeout" "60 s"
 set_config_key "web.timeout" "60000"
 
-set_config_key "taskmanager.heap.size" "1024m" # 1024Mb x 5TMs = 5Gb total heap
+set_config_key "taskmanager.memory.total-process.size" "1024m" # 1024Mb x 5TMs = 5Gb total heap
 
 set_config_key "taskmanager.memory.size" "8" # 8Mb
 set_config_key "taskmanager.network.memory.min" "128mb"

diff --git a/flink-end-to-end-tests/test-scripts/test_tpcds.sh b/flink-end-to-end-tests/test-scripts/test_tpcds.sh
@@ -54,7 +54,7 @@ cd "$END_TO_END_DIR"
 
 echo "[INFO]Preparing Flink cluster..."
 
-set_config_key "taskmanager.heap.size" "4096m"
+set_config_key "taskmanager.memory.total-process.size" "4096m"
 set_config_key "taskmanager.numberOfTaskSlots" "4"
 set_config_key "parallelism.default" "4"
 start_cluster

diff --git a/flink-jepsen/src/jepsen/flink/db.clj b/flink-jepsen/src/jepsen/flink/db.clj
@@ -304,7 +304,7 @@
  "-Djobmanager.heap.size=2048m"
  "-Djobmanager.rpc.port=6123"
  "-Dmesos.resourcemanager.tasks.mem=2048"
- "-Dtaskmanager.heap.size=2048m"
+ "-Dtaskmanager.memory.total-process.size=2048m"
  "-Dmesos.resourcemanager.tasks.cpus=1"
  "-Drest.bind-address=$(hostname -f)"))
 

diff --git a/...me/src/main/java/org/apache/flink/runtime/clusterframework/TaskExecutorResourceUtils.java b/...me/src/main/java/org/apache/flink/runtime/clusterframework/TaskExecutorResourceUtils.java
@@ -506,11 +506,11 @@ private static boolean isTotalFlinkMemorySizeExplicitlyConfigured(final Configur
  }
 
  private static boolean isTotalProcessMemorySizeExplicitlyConfigured(final Configuration config) {
- // backward compatible with the deprecated config options TASK_MANAGER_HEAP_MEMORY and TASK_MANAGER_HEAP_MEMORY_MB
+ // backward compatible with the deprecated config options TASK_MANAGER_HEAP_MEMORY_MB
  // only when they are explicitly configured by the user
  @SuppressWarnings("deprecation")
  final boolean legacyConfigured =
- config.contains(TaskManagerOptions.TASK_MANAGER_HEAP_MEMORY_MB) || config.contains(TaskManagerOptions.TASK_MANAGER_HEAP_MEMORY);
+ config.contains(TaskManagerOptions.TASK_MANAGER_HEAP_MEMORY_MB);
  final boolean configuredWithEnv = System.getenv("FLINK_TM_HEAP") != null;
  return config.contains(TaskManagerOptions.TOTAL_PROCESS_MEMORY) || legacyConfigured || configuredWithEnv;
  }

diff --git a/...rc/test/java/org/apache/flink/runtime/clusterframework/TaskExecutorResourceUtilsTest.java b/...rc/test/java/org/apache/flink/runtime/clusterframework/TaskExecutorResourceUtilsTest.java
@@ -543,7 +543,7 @@ public void testConfigTotalProcessMemoryLegacySize() {
  final MemorySize totalProcessMemorySize = MemorySize.parse("2g");
 
  @SuppressWarnings("deprecation")
- final ConfigOption<String> legacyOption = TaskManagerOptions.TASK_MANAGER_HEAP_MEMORY;
+ final ConfigOption<String> legacyOption = TaskManagerOptions.TOTAL_PROCESS_MEMORY;
 
  Configuration conf = new Configuration();
  conf.setString(legacyOption, totalProcessMemorySize.getMebiBytes() + "m");

diff --git a/...t/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerFactoryTest.java b/...t/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerFactoryTest.java
@@ -47,7 +47,7 @@ public void createResourceManager_WithLessMemoryThanContainerizedHeapCutoffMin_S
  final TestingRpcService rpcService = new TestingRpcService();
  try {
  final Configuration configuration = new Configuration();
- configuration.setString(TaskManagerOptions.TASK_MANAGER_HEAP_MEMORY, new MemorySize(128 * 1024 * 1024).toString());
+ configuration.setString(TaskManagerOptions.TOTAL_PROCESS_MEMORY, new MemorySize(128 * 1024 * 1024).toString());
  configuration.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 600);
 
  final ResourceManager<ResourceID> ignored = resourceManagerFactory.createResourceManager(

diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
@@ -374,7 +374,7 @@ public Configuration applyCommandLineOptionsToConfiguration(CommandLine commandL
  if (!MemorySize.MemoryUnit.hasUnit(tmMemoryVal)) {
  tmMemoryVal += "m";
  }
- effectiveConfiguration.setString(TaskManagerOptions.TASK_MANAGER_HEAP_MEMORY, tmMemoryVal);
+ effectiveConfiguration.setString(TaskManagerOptions.TOTAL_PROCESS_MEMORY, tmMemoryVal);
  }
 
  if (commandLine.hasOption(slots.getOpt())) {