[fix] Fix insufficient task slots

apache · Apr 10, 2024 · fff3342 · fff3342
1 parent 1847a5e
commit fff3342
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 14 deletions.
diff --git a/...on-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java b/...on-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
@@ -107,10 +107,7 @@ public void testFullCompactionTriggerInterval() throws Exception {
     @Timeout(1200)
     public void testFullCompactionWithLongCheckpointInterval() throws Exception {
         // create table
-        TableEnvironment bEnv = tableEnvironmentBuilder().batchMode().build();
-        bEnv.getConfig()
-                .getConfiguration()
-                .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 1);
+        TableEnvironment bEnv = tableEnvironmentBuilder().batchMode().parallelism(1).build();
         bEnv.executeSql(createCatalogSql("testCatalog", path));
         bEnv.executeSql("USE CATALOG testCatalog");
         bEnv.executeSql(
@@ -126,10 +123,11 @@ public void testFullCompactionWithLongCheckpointInterval() throws Exception {
 
         // run select job
         TableEnvironment sEnv =
-                tableEnvironmentBuilder().streamingMode().checkpointIntervalMs(100).build();
-        sEnv.getConfig()
-                .getConfiguration()
-                .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 1);
+                tableEnvironmentBuilder()
+                        .streamingMode()
+                        .checkpointIntervalMs(100)
+                        .parallelism(1)
+                        .build();
         sEnv.executeSql(createCatalogSql("testCatalog", path));
         sEnv.executeSql("USE CATALOG testCatalog");
         CloseableIterator<Row> it = sEnv.executeSql("SELECT * FROM T").collect();
@@ -435,9 +433,9 @@ private void testStandAloneFullCompactJobRandom(
                     streamExecutionEnvironmentBuilder()
                             .streamingMode()
                             .checkpointIntervalMs(random.nextInt(1900) + 100)
+                            .parallelism(2)
                             .allowRestart()
                             .build();
-            env.setParallelism(2);
             new CompactAction(path, "default", "T").withStreamExecutionEnvironment(env).build();
             env.executeAsync();
         }
@@ -491,10 +489,11 @@ private void testStandAloneLookupJobRandom(
 
     private void checkChangelogTestResult(int numProducers) throws Exception {
         TableEnvironment sEnv =
-                tableEnvironmentBuilder().streamingMode().checkpointIntervalMs(100).build();
-        sEnv.getConfig()
-                .getConfiguration()
-                .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 1);
+                tableEnvironmentBuilder()
+                        .streamingMode()
+                        .checkpointIntervalMs(100)
+                        .parallelism(1)
+                        .build();
         sEnv.executeSql(createCatalogSql("testCatalog", path));
         sEnv.executeSql("USE CATALOG testCatalog");
 

diff --git a/...link/paimon-flink-common/src/test/java/org/apache/paimon/flink/util/AbstractTestBase.java b/...link/paimon-flink-common/src/test/java/org/apache/paimon/flink/util/AbstractTestBase.java
@@ -46,7 +46,7 @@
 /** Similar to Flink's AbstractTestBase but using Junit5. */
 public class AbstractTestBase {
 
-    private static final int DEFAULT_PARALLELISM = 8;
+    private static final int DEFAULT_PARALLELISM = 16;
 
     @RegisterExtension
     protected static final MiniClusterWithClientExtension MINI_CLUSTER_EXTENSION =