KH-560: Allow setting parallelism and number of task slots via envs (#29

)
ozone-his · Jun 24, 2024 · 814169e · 814169e
1 parent e3123c5
commit 814169e
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/src/main/java/com/ozonehis/data/pipelines/utils/Environment.java b/src/main/java/com/ozonehis/data/pipelines/utils/Environment.java
@@ -35,7 +35,7 @@ public static MiniCluster initMiniClusterWithEnv(Boolean isStreaming) throws Exc
         Configuration flinkConfig = new Configuration();
         String port = System.getProperty(Constants.PROP_FLINK_REST_PORT);
         if (StringUtils.isBlank(port)) {
-            port = Environment.getEnv("FLINK_REST_PORT", null);
+            port = Environment.getEnv("FLINK_REST_PORT", "8081");
         }
 
         if (StringUtils.isNotBlank(port)) {
@@ -79,7 +79,7 @@ public static MiniCluster initMiniClusterWithEnv(Boolean isStreaming) throws Exc
         flinkConfig.setString("execution.checkpointing.unaligned.enabled", "true");
         flinkConfig.setString("execution.checkpointing.tolerable-failed-checkpoints", "50");
         flinkConfig.setString("table.dynamic-table-options.enabled", "true");
-        flinkConfig.setString("table.exec.resource.default-parallelism", "1");
+        flinkConfig.setString("table.exec.resource.default-parallelism", System.getenv().getOrDefault("TASK_PARALLELISM", "1"));
         flinkConfig.setString("state.backend.type", "rocksdb");
         flinkConfig.setString("state.backend.incremental", "true");
         flinkConfig.setString("state.checkpoints.dir", "file:///tmp/flink/checkpoints/");
@@ -101,7 +101,7 @@ public static MiniCluster initMiniClusterWithEnv(Boolean isStreaming) throws Exc
                 .setString("port", "9250");
         MiniClusterConfiguration clusterConfig = new MiniClusterConfiguration.Builder()
                 .setNumTaskManagers(1)
-                .setNumSlotsPerTaskManager(20)
+                .setNumSlotsPerTaskManager(Integer.parseInt(System.getenv().getOrDefault("TASK_MANAGER_SLOTS", "20")))
                 .setConfiguration(flinkConfig)
                 .build();
         MiniCluster cluster = new MiniCluster(clusterConfig);