[FLINK-21433][runtime] Pass number of slots as dynamic properties to …

…TMs. This closes apache#15128
weichunshen89 · Mar 10, 2021 · fa02b09 · fa02b09
1 parent e8daea9
commit fa02b09
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 0 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java
@@ -221,6 +221,7 @@ public static Map<String, String> parseTmResourceDynamicConfigs(String dynamicCo
         checkConfigContains(configs, TaskManagerOptions.JVM_METASPACE.key());
         checkConfigContains(configs, TaskManagerOptions.JVM_OVERHEAD_MIN.key());
         checkConfigContains(configs, TaskManagerOptions.JVM_OVERHEAD_MAX.key());
+        checkConfigContains(configs, TaskManagerOptions.NUM_TASK_SLOTS.key());
 
         return configs;
     }

diff --git a/...ime/src/main/java/org/apache/flink/runtime/clusterframework/TaskExecutorProcessUtils.java b/...ime/src/main/java/org/apache/flink/runtime/clusterframework/TaskExecutorProcessUtils.java
@@ -120,6 +120,9 @@ public static String generateDynamicConfigsStr(
                 TaskManagerOptions.JVM_OVERHEAD_MAX.key(),
                 taskExecutorProcessSpec.getJvmMetaspaceAndOverhead().getOverhead().getBytes()
                         + "b");
+        configs.put(
+                TaskManagerOptions.NUM_TASK_SLOTS.key(),
+                String.valueOf(taskExecutorProcessSpec.getNumSlots()));
         return assembleDynamicConfigsStr(configs);
     }
 

diff --git a/...src/test/java/org/apache/flink/runtime/clusterframework/TaskExecutorProcessUtilsTest.java b/...src/test/java/org/apache/flink/runtime/clusterframework/TaskExecutorProcessUtilsTest.java
@@ -112,6 +112,9 @@ public void testGenerateDynamicConfigurations() {
         assertThat(
                 MemorySize.parse(configs.get(TaskManagerOptions.JVM_OVERHEAD_MAX.key())),
                 is(TM_RESOURCE_SPEC.getJvmMetaspaceAndOverhead().getOverhead()));
+        assertThat(
+                Integer.valueOf(configs.get(TaskManagerOptions.NUM_TASK_SLOTS.key())),
+                is(TM_RESOURCE_SPEC.getNumSlots()));
     }
 
     @Test