This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 72989b47d36f fix(flink): run mapGroupsByKey in dedicated fork join
pool (#19819)
72989b47d36f is described below
commit 72989b47d36f841252d5943000ae2e7b2c6094f4
Author: Shuo Cheng <[email protected]>
AuthorDate: Thu Sep 3 11:58:25 2026 +0800
fix(flink): run mapGroupsByKey in dedicated fork join pool (#19819)
* fix(flink): run mapGroupsByKey in dedicated fork join pool
---
.../client/common/HoodieFlinkEngineContext.java | 11 ++-
.../common/TestHoodieFlinkEngineContext.java | 84 ++++++++++++++++++++++
2 files changed, 93 insertions(+), 2 deletions(-)
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
index f88a24117047..a66fa7688486 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
@@ -245,11 +245,18 @@ public class HoodieFlinkEngineContext extends
HoodieEngineContext {
boolean preservesPartitioning) {
// Group values by key and apply the function to each group in parallel
List<Iterable<V>> groupedValues =
data.groupByKey().values().collectAsList();
+ if (groupedValues.isEmpty()) {
+ return HoodieListData.eager(Collections.emptyList());
+ }
+
// Process each group in parallel using parallel stream
List<R> results = executeParallelStream(
groupedValues.parallelStream(),
- stream -> stream.map(values ->
throwingMapWrapper(processFunc).apply(new
ClosableSortingIterator<>(values.iterator()))),
-
groupedValues.size()).flatMap(CollectionUtils::toStream).collect(Collectors.toList());
+ stream -> stream
+ .map(values -> throwingMapWrapper(processFunc).apply(new
ClosableSortingIterator<>(values.iterator())))
+ .flatMap(CollectionUtils::toStream)
+ .collect(Collectors.toList()),
+ groupedValues.size());
return HoodieListData.eager(results);
}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/common/TestHoodieFlinkEngineContext.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/common/TestHoodieFlinkEngineContext.java
index 66dd39cb3469..a506d57836b7 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/common/TestHoodieFlinkEngineContext.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/common/TestHoodieFlinkEngineContext.java
@@ -18,7 +18,12 @@
package org.apache.hudi.client.common;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.data.HoodieListPairData;
+import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.util.collection.ImmutablePair;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@@ -27,8 +32,15 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinTask;
+import java.util.concurrent.atomic.AtomicBoolean;
/**
* Unit test against HoodieFlinkEngineContext.
@@ -89,4 +101,76 @@ public class TestHoodieFlinkEngineContext {
Assertions.assertEquals(resultMap.get("spark"), resultMap.get("flink"));
}
+ @Test
+ public void testMapGroupsByKeyUsesDedicatedForkJoinPool() {
+ HoodiePairData<Integer, Integer> input =
HoodieListPairData.eager(Arrays.asList(
+ Pair.of(1, 3), Pair.of(1, 1), Pair.of(2, 4), Pair.of(2, 2), Pair.of(3,
5)));
+ Set<ForkJoinPool> executingPools = ConcurrentHashMap.newKeySet();
+ AtomicBoolean executedOutsideForkJoinPool = new AtomicBoolean(false);
+
+ HoodieData<Integer> result = context.mapGroupsByKey(input, values -> {
+ recordExecutingPool(executingPools, executedOutsideForkJoinPool);
+ return new Iterator<Integer>() {
+ @Override
+ public boolean hasNext() {
+ recordExecutingPool(executingPools, executedOutsideForkJoinPool);
+ return values.hasNext();
+ }
+
+ @Override
+ public Integer next() {
+ recordExecutingPool(executingPools, executedOutsideForkJoinPool);
+ return values.next();
+ }
+ };
+ }, Arrays.asList(1, 2, 3), false);
+
+ Assertions.assertFalse(executedOutsideForkJoinPool.get());
+ Assertions.assertEquals(1, executingPools.size());
+ Assertions.assertNotSame(ForkJoinPool.commonPool(),
executingPools.iterator().next());
+ Assertions.assertEquals(3,
executingPools.iterator().next().getParallelism());
+ List<Integer> actual = result.collectAsList();
+ Collections.sort(actual);
+ Assertions.assertEquals(Arrays.asList(1, 2, 3, 4, 5), actual);
+ }
+
+ @Test
+ public void testMapGroupsByKeyWithEmptyInput() {
+ HoodiePairData<Integer, Integer> input =
HoodieListPairData.eager(Collections.emptyList());
+
+ HoodieData<Integer> result = context.mapGroupsByKey(
+ input, values -> values, Collections.emptyList(), false);
+
+ Assertions.assertTrue(result.collectAsList().isEmpty());
+ }
+
+ @Test
+ public void testMapGroupsByKeyWrapsProcessFunctionFailure() {
+ HoodiePairData<Integer, Integer> input = HoodieListPairData.eager(
+ Collections.singletonList(Pair.of(1, 1)));
+ IllegalStateException originalFailure = new IllegalStateException("process
function failure");
+
+ HoodieException failure = Assertions.assertThrows(HoodieException.class,
() ->
+ context.mapGroupsByKey(input, values -> {
+ throw originalFailure;
+ }, Collections.singletonList(1), false));
+
+ Assertions.assertEquals("Failed to execute parallel stream with dedicated
ForkJoinPool.", failure.getMessage());
+ Throwable rootCause = failure;
+ while (rootCause.getCause() != null) {
+ rootCause = rootCause.getCause();
+ }
+ Assertions.assertSame(originalFailure, rootCause);
+ }
+
+ private static void recordExecutingPool(Set<ForkJoinPool> executingPools,
+ AtomicBoolean
executedOutsideForkJoinPool) {
+ ForkJoinPool executingPool = ForkJoinTask.getPool();
+ if (executingPool == null) {
+ executedOutsideForkJoinPool.set(true);
+ } else {
+ executingPools.add(executingPool);
+ }
+ }
+
}
