[HUDI-4619] Change HoodieFlinkEngineContext.flatMap to single concu…

…rrent access.
apache · Aug 15, 2022 · 09f49ab · 09f49ab
1 parent c3621eb
commit 09f49ab
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/...udi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java b/...udi-client-common/src/main/java/org/apache/hudi/table/action/compact/HoodieCompactor.java
@@ -305,7 +305,7 @@ HoodieCompactionPlan generateCompactionPlan(
           return new CompactionOperation(dataFile, partitionPath, logFiles,
               config.getCompactionStrategy().captureMetrics(config, s));
         })
-        .filter(c -> !c.getDeltaFileNames().isEmpty()), 1).stream()
+        .filter(c -> !c.getDeltaFileNames().isEmpty()), partitionPaths.size()).stream()
         .map(CompactionUtils::buildHoodieCompactionOperation).collect(toList());
 
     LOG.info("Total of " + operations.size() + " compactions are retrieved");

diff --git a/...di-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java b/...di-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
@@ -133,7 +133,7 @@ public <I, K, V> List<V> reduceByKey(
 
   @Override
   public <I, O> List<O> flatMap(List<I> data, SerializableFunction<I, Stream<O>> func, int parallelism) {
-    return data.stream().parallel().flatMap(throwingFlatMapWrapper(func)).collect(Collectors.toList());
+    return data.stream().flatMap(throwingFlatMapWrapper(func)).collect(Collectors.toList());
   }
 
   @Override