cshuo commented on code in PR #19819:
URL: https://github.com/apache/hudi/pull/19819#discussion_r3920319606
##########
hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/common/TestHoodieFlinkEngineContext.java:
##########
@@ -89,4 +99,40 @@ public void testMapToPair() {
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 -> {
+ ForkJoinPool executingPool = ForkJoinTask.getPool();
+ if (executingPool == null) {
+ executedOutsideForkJoinPool.set(true);
+ } else {
+ executingPools.add(executingPool);
+ }
+ return values;
+ }, 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);
Review Comment:
Fixed in 8e08e59057b4. Added a regression test that asserts the
executeParallelStream HoodieException message and preserves the original
processFunc failure as the root cause.
##########
hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/common/TestHoodieFlinkEngineContext.java:
##########
@@ -89,4 +99,40 @@ public void testMapToPair() {
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 -> {
+ ForkJoinPool executingPool = ForkJoinTask.getPool();
+ if (executingPool == null) {
+ executedOutsideForkJoinPool.set(true);
+ } else {
+ executingPools.add(executingPool);
+ }
+ return values;
Review Comment:
Fixed in 8e08e59057b4. The returned iterator now records the executing pool
from both hasNext() and next(), so the test covers iterator consumption as well
as processFunc invocation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
