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 24903099c838 perf(flink): use a shared work-stealing split pool for
Source V2 bounded reads (#19520)
24903099c838 is described below
commit 24903099c838a58256567fbecbdafc7aa4c4dc8d
Author: ericyuan915 <[email protected]>
AuthorDate: Fri Aug 7 02:09:58 2026 -0700
perf(flink): use a shared work-stealing split pool for Source V2 bounded
reads (#19520)
* perf(flink): use a shared work-stealing split pool for Source V2 bounded
reads
Bounded reads inherit the streaming split provider, which pins every split
to one subtask at discovery and never rebalances, so a subtask that drew a
heavier share keeps working while its peers sit idle. On a bounded COW
backfill (~16.4K splits, parallelism 32) the readers finished 78 minutes
apart and the job took 3.80 h instead of 2.77 h.
The affinity pays for itself only in streaming, where a file group
accumulates log files across commits and successive splits of one file id
must stay on one reader. A bounded read has exactly one split per file
group, no cross-commit continuation and no ordering relationship between
splits, so any reader can read any split.
Add GlobalHoodieSplitProvider, a single shared pool ordered by the existing
HoodieSourceSplitComparator whose getNext ignores the subtask id, and select
it on the non-streaming branch of HoodieSource.createEnumerator. Streaming
keeps DefaultHoodieSplitProvider and the existing assigners. The provider is
now chosen on the streaming/bounded branch rather than before it, so restore
replays checkpointed splits into whichever provider was chosen. No
enumerator
change is needed: with one pool, getNext returning empty already means
globally drained.
Closes #19516
* test(flink): cover the metadata-driven bounded incremental branch, not
just full scan
READ_START_COMMIT=earliest leaves the analyzer's startInstant empty, so
IncrementalInputSplits.inputSplits() took its fullTableScan branch and the
COW_INCREMENTAL parameter never reached the metadata-driven branch that the
one-split-per-file-group argument cites.
Split the parameter in two: COW_INCREMENTAL now starts from a real completed
commit so fullTableScan is false, and COW_INCREMENTAL_FROM_EARLIEST keeps
the
full-scan fallback covered.
To make the branch observable, the modes that start from a real commit
write a
last commit that only touches par5 and par6. The metadata-driven branch
derives
its read partitions from that commit alone, while a full table scan lists
par1
through par6, so asserting the split partitions pins which branch produced
them.
---
.../java/org/apache/hudi/source/HoodieSource.java | 23 +-
.../source/split/GlobalHoodieSplitProvider.java | 134 ++++++
.../TestHoodieSourceEnumeratorRouting.java | 472 +++++++++++++++++++++
.../TestHoodieStaticSplitEnumerator.java | 85 ++++
.../split/TestGlobalHoodieSplitProvider.java | 298 +++++++++++++
5 files changed, 1006 insertions(+), 6 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java
index 7126dfd290ef..a4f8084cfba0 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java
@@ -34,6 +34,7 @@ import org.apache.hudi.source.reader.HoodieSourceReader;
import org.apache.hudi.source.reader.function.SplitReaderFunction;
import org.apache.hudi.source.split.DefaultHoodieSplitDiscover;
import org.apache.hudi.source.split.DefaultHoodieSplitProvider;
+import org.apache.hudi.source.split.GlobalHoodieSplitProvider;
import org.apache.hudi.source.split.HoodieContinuousSplitDiscover;
import org.apache.hudi.source.split.HoodieSourceSplit;
import org.apache.hudi.source.split.HoodieSourceSplitSerializer;
@@ -135,23 +136,33 @@ public class HoodieSource<T> extends FileIndexReader
implements Source<T, Hoodie
private SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState>
createEnumerator(
SplitEnumeratorContext<HoodieSourceSplit> enumContext,
@Nullable HoodieSplitEnumeratorState enumeratorState) {
+ final boolean streaming = scanContext.isStreaming();
+
+ // Streaming keeps per-subtask assignment (DefaultHoodieSplitProvider) so
that a file id's
+ // successive incremental splits stay affine to one reader. Bounded reads
instead use a shared
+ // work-stealing pool: the full split set is known up front and each split
is independent and
+ // order-free (one split per file group, no cross-commit continuation), so
any reader can read
+ // any split. Serving from one pool keeps every reader busy until it is
drained, which removes
+ // the straggler tail that count-balanced, non-stealing assignment
produces.
HoodieSplitProvider splitProvider;
- HoodieSplitAssigner splitAssigner =
HoodieSplitAssigners.createHoodieSplitAssigner(
- scanContext.getConf(), enumContext.currentParallelism());
-
- if (enumeratorState == null) {
+ if (streaming) {
+ HoodieSplitAssigner splitAssigner =
HoodieSplitAssigners.createHoodieSplitAssigner(
+ scanContext.getConf(), enumContext.currentParallelism());
splitProvider = new DefaultHoodieSplitProvider(splitAssigner);
} else {
+ splitProvider = new GlobalHoodieSplitProvider();
+ }
+
+ if (enumeratorState != null) {
log.info(
"Hoodie source restored {} splits from state for table {}",
enumeratorState.getPendingSplitStates().size(), tableName);
List<HoodieSourceSplit> pendingSplits =
enumeratorState.getPendingSplitStates().stream().map(HoodieSourceSplitState::getSplit).collect(Collectors.toList());
- splitProvider = new DefaultHoodieSplitProvider(splitAssigner);
splitProvider.onDiscoveredSplits(pendingSplits);
}
- if (scanContext.isStreaming()) {
+ if (streaming) {
HoodieContinuousSplitDiscover discover = new DefaultHoodieSplitDiscover(
scanContext);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/GlobalHoodieSplitProvider.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/GlobalHoodieSplitProvider.java
new file mode 100644
index 000000000000..a3a582ece4ca
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/GlobalHoodieSplitProvider.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.split;
+
+import org.apache.hudi.common.util.Option;
+
+import javax.annotation.Nullable;
+
+import java.util.Collection;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.stream.Collectors;
+
+/**
+ * Split provider that serves splits from a single shared pool, ignoring the
requesting subtask id
+ * (work stealing): whichever reader asks next gets the next pending split, so
all readers stay busy
+ * until the pool is fully drained.
+ *
+ * <p>Intended for BOUNDED (batch) reads driven by
+ * {@link org.apache.hudi.source.enumerator.HoodieStaticSplitEnumerator}.
There the complete split
+ * set is known up front and every split is independent: exactly one split per
file group, no
+ * cross-commit continuation and no ordering relationship between splits. That
holds for all bounded
+ * query modes, including the CDC one, where a file group's changes are sorted
inside a single split
+ * rather than spread over several. Any reader can therefore safely read any
split.
+ *
+ * <p>Contrast with {@link DefaultHoodieSplitProvider}, which pins each split
to one subtask (by
+ * hashing the file id, or round-robin on the split number) and never
rebalances: a subtask that
+ * drew a heavier share runs long while its peers sit idle. Because that
assignment balances split
+ * <em>count</em> rather than bytes or records, and cannot steal, even small
per-subtask skew is
+ * unrecoverable and shows up as a declining tail at the end of a bounded read.
+ *
+ * <p>NOT used for streaming reads: the continuous enumerator keeps
per-subtask assignment (via
+ * {@link DefaultHoodieSplitProvider}) so that a file id's successive
incremental splits stay affine
+ * to one reader, and so bucket id to subtask alignment is preserved for
bucket index tables.
+ *
+ * <p>Splits are served oldest-commit-first via {@link
HoodieSourceSplitComparator}, the same
+ * ordering the per-subtask queues use. Thread safe: a {@link
PriorityBlockingQueue} backs the pool,
+ * so {@link #pendingSplitCount()} can be read from the I/O threads for the
unassigned splits gauge
+ * while the coordinator thread assigns.
+ */
+public class GlobalHoodieSplitProvider implements HoodieSplitProvider {
+ public static final int INITIAL_POOL_CAPACITY = 20;
+
+ // Shared pool of unassigned splits, ordered by commit time (oldest first).
+ private final Queue<HoodieSourceSplit> pendingSplits;
+ private CompletableFuture<Void> availableFuture;
+
+ public GlobalHoodieSplitProvider() {
+ this.pendingSplits =
+ new PriorityBlockingQueue<>(INITIAL_POOL_CAPACITY, new
HoodieSourceSplitComparator());
+ }
+
+ @Override
+ public Option<HoodieSourceSplit> getNext(int taskId, @Nullable String
hostname) {
+ // Work stealing: the subtask id and hostname are intentionally ignored,
so any requesting
+ // reader gets the next split from the shared pool. Empty means the pool
is globally drained;
+ // for the static enumerator (shouldWaitForMoreSplits() == false) that
correctly triggers
+ // signalNoMoreSplits for the requesting reader.
+ HoodieSourceSplit next = pendingSplits.poll();
+ return next == null ? Option.empty() : Option.of(next);
+ }
+
+ @Override
+ public void onDiscoveredSplits(Collection<HoodieSourceSplit> splits) {
+ addSplits(splits);
+ }
+
+ @Override
+ public void onUnassignedSplits(Collection<HoodieSourceSplit> splits) {
+ // Splits handed back by a failed reader (addSplitsBack) return to the
shared pool and are
+ // picked up by whichever reader asks next, which need not be the failed
subtask. Readers that
+ // already received no-more-splits are done, but any reader still asking
can claim them.
+ addSplits(splits);
+ }
+
+ private void addSplits(Collection<HoodieSourceSplit> splits) {
+ if (splits.isEmpty()) {
+ return;
+ }
+ pendingSplits.addAll(splits);
+ completeAvailableFuturesIfNeeded();
+ }
+
+ @Override
+ public Collection<HoodieSourceSplitState> state() {
+ return pendingSplits.stream()
+ .map(split -> new HoodieSourceSplitState(split,
HoodieSourceSplitStatus.UNASSIGNED))
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void> isAvailable() {
+ if (availableFuture == null) {
+ availableFuture = new CompletableFuture<>();
+ }
+ return availableFuture;
+ }
+
+ @Override
+ public int pendingSplitCount() {
+ return pendingSplits.size();
+ }
+
+ @Override
+ public long pendingRecords() {
+ throw new UnsupportedOperationException(
+ "Pending records is not supported in GlobalHoodieSplitProvider.");
+ }
+
+ private synchronized void completeAvailableFuturesIfNeeded() {
+ if (availableFuture != null && !pendingSplits.isEmpty()) {
+ availableFuture.complete(null);
+ // Cleared only once completed, so a waiter never loses the future it is
blocked on.
+ availableFuture = null;
+ }
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieSourceEnumeratorRouting.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieSourceEnumeratorRouting.java
new file mode 100644
index 000000000000..3b129099a2a3
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieSourceEnumeratorRouting.java
@@ -0,0 +1,472 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.enumerator;
+
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.configuration.HadoopConfigurations;
+import org.apache.hudi.source.HoodieScanContext;
+import org.apache.hudi.source.HoodieSource;
+import org.apache.hudi.source.reader.HoodieRecordEmitter;
+import org.apache.hudi.source.reader.function.HoodieSplitReaderFunction;
+import org.apache.hudi.source.split.DefaultHoodieSplitProvider;
+import org.apache.hudi.source.split.GlobalHoodieSplitProvider;
+import org.apache.hudi.source.split.HoodieCdcSourceSplit;
+import org.apache.hudi.source.split.HoodieSourceSplit;
+import org.apache.hudi.source.split.HoodieSourceSplitComparator;
+import org.apache.hudi.source.split.HoodieSourceSplitState;
+import org.apache.hudi.source.split.HoodieSourceSplitStatus;
+import org.apache.hudi.source.split.HoodieSplitProvider;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+import org.apache.hudi.table.format.InternalSchemaManager;
+import org.apache.hudi.util.HoodieSchemaConverter;
+import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.TestConfigurations;
+import org.apache.hudi.utils.TestData;
+
+import org.apache.flink.api.connector.source.ReaderInfo;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.connector.source.SplitsAssignment;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests which split provider {@link HoodieSource} wires into the enumerator.
+ *
+ * <p>Bounded reads use the shared work-stealing pool ({@link
GlobalHoodieSplitProvider}); streaming
+ * keeps per-subtask assignment ({@link DefaultHoodieSplitProvider}) so that a
file id's successive
+ * incremental splits stay affine to one reader. Because {@code
HoodieSource.createEnumerator}
+ * handles fresh creation and restore in the same method, both paths are
covered for every mode.
+ *
+ * <p>These tests also assert the property that makes work stealing safe for
bounded reads: every
+ * bounded query mode emits exactly one split per file group, so there is no
cross-commit
+ * continuation and no ordering relationship between splits that a shared pool
could break.
+ *
+ * <p>Lives in the enumerator package so it can read the package-private
+ * {@link AbstractHoodieSplitEnumerator#splitProvider}.
+ */
+public class TestHoodieSourceEnumeratorRouting {
+
+ @TempDir
+ File tempDir;
+
+ private Configuration conf;
+ private StoragePath tablePath;
+ private HoodieTableMetaClient metaClient;
+
+ /**
+ * The bounded query modes {@code HoodieSource.createBatchHoodieSplits()}
covers. All of them are
+ * routed to the shared pool, so all of them are exercised here.
+ *
+ * <p>Incremental appears three times on purpose. {@code
IncrementalInputSplits.inputSplits()}
+ * branches on {@code fullTableScan}, which is true when the query consumes
from the earliest
+ * instant, and the two sides build their file slice set differently: the
full scan lists the
+ * table directly, while the other side derives partitions and files from
the commit metadata of
+ * the instants in range (and, when CDC is on, leaves through the CDC
extractor entirely). Only
+ * covering {@code earliest} would leave the metadata-driven branch untested.
+ */
+ private enum BoundedMode {
+ COW_SNAPSHOT(HoodieTableType.COPY_ON_WRITE,
FlinkOptions.QUERY_TYPE_SNAPSHOT, false, IncrementalStart.NOT_INCREMENTAL),
+ MOR_SNAPSHOT(HoodieTableType.MERGE_ON_READ,
FlinkOptions.QUERY_TYPE_SNAPSHOT, false, IncrementalStart.NOT_INCREMENTAL),
+ MOR_READ_OPTIMIZED(HoodieTableType.MERGE_ON_READ,
FlinkOptions.QUERY_TYPE_READ_OPTIMIZED, false,
IncrementalStart.NOT_INCREMENTAL),
+ COW_INCREMENTAL(HoodieTableType.COPY_ON_WRITE,
FlinkOptions.QUERY_TYPE_INCREMENTAL, false, IncrementalStart.LAST_COMMIT),
+ COW_INCREMENTAL_FROM_EARLIEST(HoodieTableType.COPY_ON_WRITE,
FlinkOptions.QUERY_TYPE_INCREMENTAL, false, IncrementalStart.EARLIEST),
+ COW_INCREMENTAL_CDC(HoodieTableType.COPY_ON_WRITE,
FlinkOptions.QUERY_TYPE_INCREMENTAL, true, IncrementalStart.LAST_COMMIT);
+
+ private final HoodieTableType tableType;
+ private final String queryType;
+ private final boolean cdcEnabled;
+ private final IncrementalStart incrementalStart;
+
+ BoundedMode(HoodieTableType tableType, String queryType, boolean
cdcEnabled, IncrementalStart incrementalStart) {
+ this.tableType = tableType;
+ this.queryType = queryType;
+ this.cdcEnabled = cdcEnabled;
+ this.incrementalStart = incrementalStart;
+ }
+
+ boolean isIncremental() {
+ return incrementalStart != IncrementalStart.NOT_INCREMENTAL;
+ }
+ }
+
+ /**
+ * Where an incremental mode starts reading, which is what decides the
{@code fullTableScan}
+ * branch: {@code earliest} leaves {@code startInstant} empty and takes the
full scan,
+ * a real completion time takes the metadata-driven branch.
+ */
+ private enum IncrementalStart {
+ NOT_INCREMENTAL,
+ EARLIEST,
+ LAST_COMMIT
+ }
+
+ @BeforeEach
+ public void setUp() {
+ conf = TestConfigurations.getDefaultConf(tempDir.getAbsolutePath());
+ tablePath = new StoragePath(tempDir.getAbsolutePath());
+ }
+
+ @ParameterizedTest
+ @EnumSource(BoundedMode.class)
+ public void testBoundedReadUsesSharedSplitPool(BoundedMode mode) throws
Exception {
+ HoodieSource<RowData> source = prepareBoundedSource(mode);
+ MockSplitEnumeratorContext context = new MockSplitEnumeratorContext();
+
+ SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> enumerator =
+ source.createEnumerator(context);
+
+ assertInstanceOf(HoodieStaticSplitEnumerator.class, enumerator,
+ "Bounded read should use the static enumerator for mode " + mode);
+ assertInstanceOf(GlobalHoodieSplitProvider.class, providerOf(enumerator),
+ "Bounded read should use the shared work-stealing pool for mode " +
mode);
+ List<HoodieSourceSplit> splits = pendingSplits(enumerator);
+ assertOneSplitPerFileGroup(splits, mode);
+ if (mode.incrementalStart == IncrementalStart.LAST_COMMIT) {
+ // Guards the parameterization: if the start commit stopped making
fullTableScan false, these
+ // splits would come from a full table listing and cover par1 through
par6, and this mode
+ // would silently stop exercising the metadata-driven branch.
+ assertEquals(new HashSet<>(Arrays.asList("par5", "par6")),
+
splits.stream().map(HoodieSourceSplit::getPartitionPath).collect(Collectors.toSet()),
+ "Mode " + mode + " should read only the partitions written by the
start commit, "
+ + "which is what distinguishes the incremental branch from a
full table scan");
+ }
+ if (mode.cdcEnabled) {
+ // Likewise, a full table scan would bypass the CDC extractor and yield
plain splits.
+ splits.forEach(split -> assertInstanceOf(HoodieCdcSourceSplit.class,
split,
+ "CDC mode should produce CDC splits"));
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(BoundedMode.class)
+ public void testBoundedRestoreKeepsSharedSplitPool(BoundedMode mode) throws
Exception {
+ HoodieSource<RowData> source = prepareBoundedSource(mode);
+ // Snapshot the real splits of this mode, then restore from a subset of
them.
+ List<HoodieSourceSplit> discovered =
+ pendingSplits(source.createEnumerator(new
MockSplitEnumeratorContext()));
+ assertFalse(discovered.isEmpty(), "Expected at least one split for mode "
+ mode);
+ List<HoodieSourceSplit> checkpointed = discovered.subList(0, 1);
+
+ MockSplitEnumeratorContext context = new MockSplitEnumeratorContext();
+ SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> restored =
+ source.restoreEnumerator(context, enumeratorStateOf(checkpointed));
+
+ assertInstanceOf(HoodieStaticSplitEnumerator.class, restored,
+ "Restored bounded read should still use the static enumerator for mode
" + mode);
+ assertInstanceOf(GlobalHoodieSplitProvider.class, providerOf(restored),
+ "Restored bounded read should still use the shared work-stealing pool
for mode " + mode);
+ assertEquals(checkpointed.size(), providerOf(restored).pendingSplitCount(),
+ "Restore should replay exactly the checkpointed splits into the shared
pool "
+ + "and must not re-run split discovery for mode " + mode);
+ }
+
+ /**
+ * A restored pending split is not owned by any subtask: whichever reader
asks for work claims it.
+ * Parameterized over the requesting subtask so the assertion is
deterministic - under per-subtask
+ * pinning only the one subtask the file id hashes to could ever receive it.
+ */
+ @ParameterizedTest
+ @ValueSource(ints = {0, 1, 2, 3})
+ public void testRestoredSplitIsClaimedByWhicheverSubtaskAsks(int
requestingSubtask) throws Exception {
+ HoodieSource<RowData> source =
prepareBoundedSource(BoundedMode.COW_SNAPSHOT);
+ List<HoodieSourceSplit> discovered =
+ pendingSplits(source.createEnumerator(new
MockSplitEnumeratorContext()));
+ List<HoodieSourceSplit> checkpointed = discovered.subList(0, 1);
+
+ MockSplitEnumeratorContext context = new MockSplitEnumeratorContext();
+ SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> restored =
+ source.restoreEnumerator(context, enumeratorStateOf(checkpointed));
+ restored.start();
+ for (int subtask = 0; subtask < 4; subtask++) {
+ context.registerReader(new ReaderInfo(subtask, "localhost"));
+ }
+
+ restored.handleSplitRequest(requestingSubtask, "localhost");
+
+ assertEquals(checkpointed,
context.getAssignedSplits().get(requestingSubtask),
+ "The restored split should go to whichever subtask asked for work");
+ assertFalse(context.getNoMoreSplitsSignaled().contains(requestingSubtask),
+ "The requesting subtask received a split, so it should not be told
no-more-splits");
+ }
+
+ @Test
+ public void testStreamingReadKeepsPerSubtaskProvider() throws Exception {
+ HoodieSource<RowData> source = prepareStreamingSource();
+
+ SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> enumerator =
+ source.createEnumerator(new MockSplitEnumeratorContext());
+
+ assertInstanceOf(HoodieContinuousSplitEnumerator.class, enumerator,
+ "Streaming read should use the continuous enumerator");
+ assertInstanceOf(DefaultHoodieSplitProvider.class, providerOf(enumerator),
+ "Streaming read must keep per-subtask assignment for file id
affinity");
+ }
+
+ @Test
+ public void testStreamingRestoreKeepsPerSubtaskProvider() throws Exception {
+ HoodieSource<RowData> source = prepareStreamingSource();
+ List<HoodieSourceSplit> checkpointed = Collections.singletonList(
+ new HoodieSourceSplit(0, null, Option.empty(), tablePath.toString(),
"par1",
+ FlinkOptions.REALTIME_PAYLOAD_COMBINE, "20260126034717000",
"file-0", Option.empty()));
+
+ SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> restored =
+ source.restoreEnumerator(new MockSplitEnumeratorContext(),
enumeratorStateOf(checkpointed));
+
+ assertInstanceOf(HoodieContinuousSplitEnumerator.class, restored,
+ "Restored streaming read should use the continuous enumerator");
+ assertInstanceOf(DefaultHoodieSplitProvider.class, providerOf(restored),
+ "Restored streaming read must keep per-subtask assignment");
+ assertEquals(1, providerOf(restored).pendingSplitCount(),
+ "Restored split should be replayed into the per-subtask provider");
+ }
+
+ // Helper methods
+
+ private static HoodieSplitProvider providerOf(
+ SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState>
enumerator) {
+ return ((AbstractHoodieSplitEnumerator) enumerator).splitProvider;
+ }
+
+ private static List<HoodieSourceSplit> pendingSplits(
+ SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState>
enumerator) {
+ return providerOf(enumerator).state().stream()
+ .map(HoodieSourceSplitState::getSplit)
+ .collect(Collectors.toList());
+ }
+
+ private static HoodieSplitEnumeratorState
enumeratorStateOf(List<HoodieSourceSplit> splits) {
+ List<HoodieSourceSplitState> states = splits.stream()
+ .map(split -> new HoodieSourceSplitState(split,
HoodieSourceSplitStatus.UNASSIGNED))
+ .collect(Collectors.toList());
+ return new HoodieSplitEnumeratorState(states, Option.empty(),
Option.empty());
+ }
+
+ /**
+ * Asserts the invariant that makes a shared pool safe for a bounded read:
one split per file
+ * group, hence no cross-commit continuation and no ordering relationship
between splits.
+ */
+ private static void assertOneSplitPerFileGroup(List<HoodieSourceSplit>
splits, BoundedMode mode) {
+ assertFalse(splits.isEmpty(), "Expected at least one split for mode " +
mode);
+ Set<String> fileIds = splits.stream()
+ .map(HoodieSourceSplit::getFileId)
+ .collect(Collectors.toSet());
+ assertEquals(splits.size(), fileIds.size(),
+ "Mode " + mode + " must emit exactly one split per file group,
otherwise splits of the "
+ + "same file group could be read concurrently by different
readers");
+ }
+
+ private HoodieSource<RowData> prepareBoundedSource(BoundedMode mode) throws
Exception {
+ conf.set(FlinkOptions.TABLE_TYPE, mode.tableType.name());
+ conf.set(FlinkOptions.READ_AS_STREAMING, false);
+ if (mode.tableType == HoodieTableType.MERGE_ON_READ) {
+ // Compact the first commit so the MOR file groups own a base file (a
read-optimized read
+ // sees nothing otherwise); the second commit below is then written as
logs only, so a
+ // snapshot read exercises real base + log file slices.
+ conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, true);
+ conf.set(FlinkOptions.COMPACTION_DELTA_COMMITS, 1);
+ }
+ if (mode.cdcEnabled) {
+ conf.set(FlinkOptions.CDC_ENABLED, true);
+ conf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, true); // for batch update
+ }
+
+ TestData.writeData(TestData.DATA_SET_INSERT, conf);
+ if (mode.tableType == HoodieTableType.MERGE_ON_READ) {
+ conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
+ }
+ TestData.writeData(TestData.DATA_SET_UPDATE_INSERT, conf);
+ if (mode.incrementalStart == IncrementalStart.LAST_COMMIT) {
+ // A last commit that only touches par5 and par6, so the partitions of
the resulting splits
+ // show which branch produced them: the metadata-driven branch derives
its read partitions
+ // from this commit alone, a full table scan would list par1 through
par6.
+ TestData.writeData(TestData.DATA_SET_INSERT_SEPARATE_PARTITION, conf);
+ }
+ metaClient = StreamerUtil.createMetaClient(conf);
+
+ conf.set(FlinkOptions.QUERY_TYPE, mode.queryType);
+ if (mode.incrementalStart == IncrementalStart.EARLIEST) {
+ conf.set(FlinkOptions.READ_START_COMMIT,
FlinkOptions.START_COMMIT_EARLIEST);
+ } else if (mode.incrementalStart == IncrementalStart.LAST_COMMIT) {
+ conf.set(FlinkOptions.READ_START_COMMIT, lastCompletionTime());
+ }
+ return createSource();
+ }
+
+ private HoodieSource<RowData> prepareStreamingSource() throws Exception {
+ conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.MERGE_ON_READ.name());
+ conf.set(FlinkOptions.READ_AS_STREAMING, true);
+
+ TestData.writeData(TestData.DATA_SET_INSERT, conf);
+ metaClient = StreamerUtil.createMetaClient(conf);
+
+ return createSource();
+ }
+
+ private String lastCompletionTime() {
+ List<String> commits =
metaClient.getCommitsTimeline().filterCompletedInstants()
+ .getInstantsAsStream()
+ .map(HoodieInstant::getCompletionTime)
+ .collect(Collectors.toList());
+ assertTrue(commits.size() > 1, "Expected more than one commit to read
changes from");
+ return commits.get(commits.size() - 1);
+ }
+
+ private HoodieSource<RowData> createSource() {
+ RowType rowType = TestConfigurations.ROW_TYPE;
+ HoodieScanContext scanContext = HoodieScanContext.builder()
+ .conf(conf)
+ .path(tablePath)
+ .rowType(rowType)
+ .startInstant(conf.get(FlinkOptions.READ_START_COMMIT))
+ .endInstant(conf.get(FlinkOptions.READ_END_COMMIT))
+
.maxCompactionMemoryInBytes(conf.get(FlinkOptions.COMPACTION_MAX_MEMORY))
+ .maxPendingSplits(1000)
+ .skipCompaction(conf.get(FlinkOptions.READ_STREAMING_SKIP_COMPACT))
+ .skipClustering(conf.get(FlinkOptions.READ_STREAMING_SKIP_CLUSTERING))
+
.skipInsertOverwrite(conf.get(FlinkOptions.READ_STREAMING_SKIP_INSERT_OVERWRITE))
+ .cdcEnabled(conf.get(FlinkOptions.CDC_ENABLED))
+ .isStreaming(conf.get(FlinkOptions.READ_AS_STREAMING))
+ .build();
+ HoodieSchema schema = HoodieSchemaConverter.convertToSchema(rowType);
+ HadoopStorageConfiguration hadoopConf =
+ new
HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(conf));
+ InternalSchemaManager internalSchemaManager =
InternalSchemaManager.get(hadoopConf, metaClient);
+
+ return new HoodieSource<>(
+ scanContext,
+ () -> new HoodieSplitReaderFunction(
+ conf,
+ schema,
+ schema,
+ internalSchemaManager,
+ conf.get(FlinkOptions.MERGE_TYPE),
+ Collections.emptyList(),
+ false),
+ new HoodieSourceSplitComparator(),
+ metaClient,
+ new HoodieRecordEmitter<>());
+ }
+
+ /**
+ * Minimal mock of {@link SplitEnumeratorContext} for the wiring assertions
above.
+ */
+ private static class MockSplitEnumeratorContext implements
SplitEnumeratorContext<HoodieSourceSplit> {
+ private final Map<Integer, ReaderInfo> registeredReaders = new HashMap<>();
+ private final Map<Integer, List<HoodieSourceSplit>> assignedSplits = new
HashMap<>();
+ private final List<Integer> noMoreSplitsSignaled = new ArrayList<>();
+
+ void registerReader(ReaderInfo readerInfo) {
+ registeredReaders.put(readerInfo.getSubtaskId(), readerInfo);
+ }
+
+ Map<Integer, List<HoodieSourceSplit>> getAssignedSplits() {
+ return assignedSplits;
+ }
+
+ List<Integer> getNoMoreSplitsSignaled() {
+ return noMoreSplitsSignaled;
+ }
+
+ @Override
+ public SplitEnumeratorMetricGroup metricGroup() {
+ return UnregisteredMetricsGroup.createSplitEnumeratorMetricGroup();
+ }
+
+ @Override
+ public void sendEventToSourceReader(int subtaskId, SourceEvent event) {
+ // No-op for testing
+ }
+
+ @Override
+ public int currentParallelism() {
+ return Math.max(registeredReaders.size(), 1);
+ }
+
+ @Override
+ public Map<Integer, ReaderInfo> registeredReaders() {
+ return new HashMap<>(registeredReaders);
+ }
+
+ @Override
+ public void assignSplits(SplitsAssignment<HoodieSourceSplit>
newSplitAssignments) {
+ newSplitAssignments.assignment().forEach((subtask, splits) ->
+ assignedSplits.computeIfAbsent(subtask, k -> new
ArrayList<>()).addAll(splits));
+ }
+
+ @Override
+ public void assignSplit(HoodieSourceSplit split, int subtask) {
+ assignedSplits.computeIfAbsent(subtask, k -> new
ArrayList<>()).add(split);
+ }
+
+ @Override
+ public void signalNoMoreSplits(int subtask) {
+ noMoreSplitsSignaled.add(subtask);
+ }
+
+ @Override
+ public <T> void callAsync(Callable<T> callable, BiConsumer<T, Throwable>
handler) {
+ // No-op: split discovery is not exercised by these wiring tests.
+ }
+
+ @Override
+ public <T> void callAsync(Callable<T> callable, BiConsumer<T, Throwable>
handler, long initialDelay, long period) {
+ // No-op: split discovery is not exercised by these wiring tests.
+ }
+
+ @Override
+ public void runInCoordinatorThread(Runnable runnable) {
+ runnable.run();
+ }
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieStaticSplitEnumerator.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieStaticSplitEnumerator.java
index 70a564bd6fb4..84822c21cdd6 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieStaticSplitEnumerator.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/enumerator/TestHoodieStaticSplitEnumerator.java
@@ -20,6 +20,7 @@ package org.apache.hudi.source.enumerator;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.source.split.DefaultHoodieSplitProvider;
+import org.apache.hudi.source.split.GlobalHoodieSplitProvider;
import org.apache.hudi.source.split.HoodieSourceSplit;
import org.apache.hudi.source.split.SplitRequestEvent;
import org.apache.hudi.source.split.assign.HoodieSplitNumberAssigner;
@@ -252,6 +253,90 @@ public class TestHoodieStaticSplitEnumerator {
"Should throw IllegalArgumentException for unknown source event type");
}
+ @Test
+ public void testGlobalProviderWorkStealingAcrossSubtasks() {
+ // With the shared work-stealing pool a single reader can drain every
split: the enumerator no
+ // longer pins splits to a subtask. DefaultHoodieSplitProvider with a
number/hash assigner would
+ // hand most of these to other subtasks and starve reader 0.
+ GlobalHoodieSplitProvider globalProvider = new GlobalHoodieSplitProvider();
+ HoodieStaticSplitEnumerator globalEnumerator =
+ new HoodieStaticSplitEnumerator("test-table", context, globalProvider);
+ globalProvider.onDiscoveredSplits(Arrays.asList(split1, split2, split3));
+ globalEnumerator.start();
+
+ context.registerReader(new ReaderInfo(0, "localhost"));
+ context.registerReader(new ReaderInfo(1, "localhost"));
+
+ // Reader 0 keeps finishing and asking for more; it takes all three splits
by itself.
+ globalEnumerator.handleSplitRequest(0, "localhost");
+ globalEnumerator.handleSplitRequest(0, "localhost");
+ globalEnumerator.handleSplitRequest(0, "localhost");
+
+ assertEquals(3, context.getAssignedSplits().get(0).size(),
+ "A single reader should be able to steal the entire pool");
+ assertFalse(context.getNoMoreSplitsSignaled().contains(0),
+ "No-more-splits must not fire while the pool still had splits");
+ }
+
+ @Test
+ public void testGlobalProviderSignalsNoMoreSplitsOnlyWhenPoolEmpty() {
+ GlobalHoodieSplitProvider globalProvider = new GlobalHoodieSplitProvider();
+ HoodieStaticSplitEnumerator globalEnumerator =
+ new HoodieStaticSplitEnumerator("test-table", context, globalProvider);
+ globalProvider.onDiscoveredSplits(Collections.singletonList(split1)); //
one split, two readers
+ globalEnumerator.start();
+
+ context.registerReader(new ReaderInfo(0, "localhost"));
+ context.registerReader(new ReaderInfo(1, "localhost"));
+
+ globalEnumerator.handleSplitRequest(0, "localhost"); // reader 0 takes the
only split
+ globalEnumerator.handleSplitRequest(1, "localhost"); // reader 1 finds the
shared pool empty
+
+ assertTrue(context.getAssignedSplits().containsKey(0), "Reader 0 should
receive the split");
+ assertFalse(context.getNoMoreSplitsSignaled().contains(0),
+ "Reader 0 got a split, so it should not be told no-more-splits");
+ assertTrue(context.getNoMoreSplitsSignaled().contains(1),
+ "Reader 1 should be told no-more-splits once the shared pool is
drained");
+ }
+
+ @Test
+ public void
testGlobalProviderAddSplitsBackAfterOtherReadersGotNoMoreSplits() {
+ // Failure recovery once the pool has already been drained and some
readers have finished:
+ // the split a failed reader hands back must land in the shared pool and
stay claimable by a
+ // subtask that is neither the failed one nor an already-finished one.
Under per-subtask
+ // pinning it would instead be re-pinned to hash(fileId), possibly a
reader that is already
+ // done, and never be read.
+ GlobalHoodieSplitProvider globalProvider = new GlobalHoodieSplitProvider();
+ HoodieStaticSplitEnumerator globalEnumerator =
+ new HoodieStaticSplitEnumerator("test-table", context, globalProvider);
+ globalProvider.onDiscoveredSplits(Collections.singletonList(split1));
+ globalEnumerator.start();
+
+ context.registerReader(new ReaderInfo(0, "localhost"));
+ context.registerReader(new ReaderInfo(1, "localhost"));
+ context.registerReader(new ReaderInfo(2, "localhost"));
+
+ globalEnumerator.handleSplitRequest(0, "localhost"); // reader 0 takes the
only split
+ globalEnumerator.handleSplitRequest(1, "localhost"); // pool is drained,
reader 1 finishes
+ assertTrue(context.getNoMoreSplitsSignaled().contains(1),
+ "Reader 1 should already have been told no-more-splits");
+
+ // Reader 0 fails mid-split and its split is returned.
+ context.unregisterReader(0);
+ globalEnumerator.addSplitsBack(Collections.singletonList(split1), 0);
+ assertEquals(1, globalProvider.pendingSplitCount(),
+ "Returned split should be back in the shared pool");
+
+ // Reader 2, which has neither failed nor finished, claims it.
+ globalEnumerator.handleSplitRequest(2, "localhost");
+
+ assertEquals(Collections.singletonList(split1),
context.getAssignedSplits().get(2),
+ "A different, still-running subtask should claim the returned split");
+ assertFalse(context.getNoMoreSplitsSignaled().contains(2),
+ "Reader 2 got the returned split, so it should not be told
no-more-splits");
+ assertEquals(0, globalProvider.pendingSplitCount(), "Pool should be
drained again");
+ }
+
private HoodieSourceSplit createTestSplit(int splitNum, String fileId) {
return new HoodieSourceSplit(
splitNum,
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestGlobalHoodieSplitProvider.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestGlobalHoodieSplitProvider.java
new file mode 100644
index 000000000000..807c306c504f
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/split/TestGlobalHoodieSplitProvider.java
@@ -0,0 +1,298 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.source.split;
+
+import org.apache.hudi.common.util.Option;
+
+import org.junit.jupiter.api.BeforeEach;
+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.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Test cases for {@link GlobalHoodieSplitProvider}.
+ *
+ * <p>The distinguishing behavior from {@link DefaultHoodieSplitProvider} is
that splits are NOT
+ * pinned to a subtask: any requesting subtask gets the next split from a
single shared pool (work
+ * stealing), and the pool drains fully regardless of which subtasks do the
asking.
+ */
+public class TestGlobalHoodieSplitProvider {
+ private GlobalHoodieSplitProvider provider;
+ private HoodieSourceSplit split1;
+ private HoodieSourceSplit split2;
+ private HoodieSourceSplit split3;
+
+ @BeforeEach
+ public void setUp() {
+ provider = new GlobalHoodieSplitProvider();
+ split1 = createTestSplit(1, "file1");
+ split2 = createTestSplit(2, "file2");
+ split3 = createTestSplit(3, "file3");
+ }
+
+ @Test
+ public void testGetNextFromEmptyProvider() {
+ assertFalse(provider.getNext(0, null).isPresent(),
+ "Should return empty option when no splits available");
+ }
+
+ @Test
+ public void testAnySubtaskGetsNextSplit() {
+ provider.onDiscoveredSplits(Arrays.asList(split1, split2, split3));
+
+ // Three unrelated subtask ids each pull one split from the shared pool;
together they drain it,
+ // and every discovered split is handed out exactly once.
+ Set<String> served = new HashSet<>();
+ served.add(requireSplit(provider.getNext(0, null)).splitId());
+ served.add(requireSplit(provider.getNext(5, null)).splitId());
+ served.add(requireSplit(provider.getNext(99, "some-host")).splitId());
+
+ assertEquals(
+ new HashSet<>(Arrays.asList(split1.splitId(), split2.splitId(),
split3.splitId())),
+ served,
+ "Every split should be served exactly once across arbitrary subtasks");
+ assertFalse(provider.getNext(0, null).isPresent(), "Pool should be
drained");
+ assertEquals(0, provider.pendingSplitCount());
+ }
+
+ @Test
+ public void testSingleSubtaskCanDrainEntirePool() {
+ // Work stealing: one reader may take every split.
DefaultHoodieSplitProvider would instead pin
+ // most of these to other subtasks and starve this one.
+ provider.onDiscoveredSplits(Arrays.asList(split1, split2, split3));
+
+ assertTrue(provider.getNext(7, null).isPresent());
+ assertTrue(provider.getNext(7, null).isPresent());
+ assertTrue(provider.getNext(7, null).isPresent());
+ assertFalse(provider.getNext(7, null).isPresent(),
+ "Fourth request should be empty once the single reader drained the
pool");
+ }
+
+ @Test
+ public void testServedOldestCommitFirstRegardlessOfSubtask() {
+ HoodieSourceSplit early = createSplitWithCommit(1, "20260126034716930",
"file_early");
+ HoodieSourceSplit middle = createSplitWithCommit(2, "20260126034717000",
"file_middle");
+ HoodieSourceSplit late = createSplitWithCommit(3, "20260126034718000",
"file_late");
+
+ // Discover out of order and request from different subtasks: ordering is
by commit time, not by
+ // requester or insertion order.
+ provider.onDiscoveredSplits(Arrays.asList(late, early, middle));
+
+ assertEquals(early.splitId(), requireSplit(provider.getNext(3,
null)).splitId());
+ assertEquals(middle.splitId(), requireSplit(provider.getNext(8,
null)).splitId());
+ assertEquals(late.splitId(), requireSplit(provider.getNext(0,
null)).splitId());
+ }
+
+ @Test
+ public void testOnUnassignedSplitsReturnedToPoolForAnySubtask() {
+ provider.onDiscoveredSplits(Collections.singletonList(split1));
+ HoodieSourceSplit taken = requireSplit(provider.getNext(0, null));
+ assertEquals(0, provider.pendingSplitCount());
+
+ // A failed reader hands the split back; a different subtask can pick it
up.
+ provider.onUnassignedSplits(Collections.singletonList(taken));
+ assertEquals(1, provider.pendingSplitCount(), "Returned split should be
back in the pool");
+ assertEquals(taken.splitId(), requireSplit(provider.getNext(4,
null)).splitId());
+ }
+
+ @Test
+ public void testPendingSplitCount() {
+ assertEquals(0, provider.pendingSplitCount(), "Initially should have 0
pending splits");
+
+ provider.onDiscoveredSplits(Arrays.asList(split1, split2, split3));
+ assertEquals(3, provider.pendingSplitCount());
+
+ provider.getNext(0, null);
+ provider.getNext(1, null);
+ assertEquals(1, provider.pendingSplitCount(),
+ "Count should drop as splits are served to any subtask");
+ }
+
+ @Test
+ public void testMultipleDiscoveryCalls() {
+ provider.onDiscoveredSplits(Collections.singletonList(split1));
+ provider.onDiscoveredSplits(Arrays.asList(split2, split3));
+ assertEquals(3, provider.pendingSplitCount(), "All discovered splits
accumulate in the pool");
+ }
+
+ @Test
+ public void testEmptyDiscoveredSplits() {
+ provider.onDiscoveredSplits(Collections.emptyList());
+ assertEquals(0, provider.pendingSplitCount());
+ assertFalse(provider.getNext(0, null).isPresent());
+ }
+
+ @Test
+ public void testState() {
+ provider.onDiscoveredSplits(Arrays.asList(split1, split2, split3));
+
+ Collection<HoodieSourceSplitState> states = provider.state();
+ assertEquals(3, states.size(), "State should contain all pending splits");
+ for (HoodieSourceSplitState state : states) {
+ assertEquals(HoodieSourceSplitStatus.UNASSIGNED, state.getStatus(),
+ "Pending splits should be UNASSIGNED");
+ }
+ }
+
+ @Test
+ public void testStateAfterConsumingSomeSplits() {
+ provider.onDiscoveredSplits(Arrays.asList(split1, split2, split3));
+ provider.getNext(0, null);
+ provider.getNext(1, null);
+
+ assertEquals(1, provider.state().size(), "State should only reflect the
remaining split");
+ }
+
+ @Test
+ public void testStateRoundTripsThroughRediscovery() {
+ // Mirrors the enumerator restore path: snapshot pending splits, rebuild a
fresh provider and
+ // re-discover them. No split is lost or duplicated.
+ provider.onDiscoveredSplits(Arrays.asList(split1, split2, split3));
+ provider.getNext(0, null); // one assigned, two remain pending in the
checkpoint
+
+ List<HoodieSourceSplit> checkpointed = new ArrayList<>();
+ for (HoodieSourceSplitState state : provider.state()) {
+ checkpointed.add(state.getSplit());
+ }
+
+ GlobalHoodieSplitProvider restored = new GlobalHoodieSplitProvider();
+ restored.onDiscoveredSplits(checkpointed);
+ assertEquals(2, restored.pendingSplitCount());
+ assertTrue(restored.getNext(0, null).isPresent());
+ assertTrue(restored.getNext(0, null).isPresent());
+ assertFalse(restored.getNext(0, null).isPresent());
+ }
+
+ @Test
+ public void testIsAvailable() {
+ CompletableFuture<Void> future = provider.isAvailable();
+ assertNotNull(future, "isAvailable should return a future");
+ assertFalse(future.isDone(), "Future should not be completed with no
splits");
+ assertSame(future, provider.isAvailable(),
+ "The same future should be returned until it completes");
+ }
+
+ @Test
+ public void testIsAvailableCompletesOnDiscovery() {
+ CompletableFuture<Void> future = provider.isAvailable();
+
+ provider.onDiscoveredSplits(Collections.singletonList(split1));
+
+ assertTrue(future.isDone(), "Future should complete once splits land in
the pool");
+ assertFalse(provider.isAvailable().isDone(),
+ "A fresh, uncompleted future should be handed out afterwards");
+ }
+
+ @Test
+ public void testPendingRecordsUnsupported() {
+ assertThrows(UnsupportedOperationException.class, () ->
provider.pendingRecords());
+ }
+
+ @Test
+ public void testConcurrentDrainServesEachSplitExactlyOnce() throws Exception
{
+ final int splitCount = 500;
+ final int readerCount = 8;
+ List<HoodieSourceSplit> splits = new ArrayList<>();
+ for (int i = 0; i < splitCount; i++) {
+ splits.add(createTestSplit(i, "file" + i));
+ }
+ provider.onDiscoveredSplits(splits);
+
+ ConcurrentLinkedQueue<String> served = new ConcurrentLinkedQueue<>();
+ CountDownLatch start = new CountDownLatch(1);
+ List<Future<?>> drains = new ArrayList<>();
+ ExecutorService readers = Executors.newFixedThreadPool(readerCount);
+ try {
+ for (int reader = 0; reader < readerCount; reader++) {
+ final int subtaskId = reader;
+ drains.add(readers.submit(() -> {
+ start.await();
+ Option<HoodieSourceSplit> next;
+ while ((next = provider.getNext(subtaskId, null)).isPresent()) {
+ served.add(next.get().splitId());
+ }
+ return null;
+ }));
+ }
+ start.countDown();
+ readers.shutdown();
+ assertTrue(readers.awaitTermination(30, TimeUnit.SECONDS), "Readers
should drain the pool");
+ for (Future<?> drain : drains) {
+ drain.get(); // surface any failure inside a reader thread
+ }
+ } finally {
+ readers.shutdownNow();
+ }
+
+ assertEquals(splitCount, served.size(), "No split should be served twice");
+ assertEquals(splitCount, new HashSet<>(served).size(), "No split should be
lost");
+ assertEquals(0, provider.pendingSplitCount(), "Pool should be fully
drained");
+ }
+
+ private static HoodieSourceSplit requireSplit(Option<HoodieSourceSplit>
option) {
+ assertTrue(option.isPresent(), "Expected a split to be available");
+ return option.get();
+ }
+
+ private HoodieSourceSplit createSplitWithCommit(int splitNum, String
latestCommit, String basePath) {
+ return new HoodieSourceSplit(
+ splitNum,
+ basePath,
+ Option.empty(),
+ "/table/path",
+ "/table/path/partition1",
+ "read_optimized",
+ latestCommit,
+ "file" + splitNum,
+ Option.empty());
+ }
+
+ private HoodieSourceSplit createTestSplit(int splitNum, String fileId) {
+ return new HoodieSourceSplit(
+ splitNum,
+ "40e603a8-3cc1-4d09-b0a5-1432992b4bf7_1-0" + splitNum +
"_20260126034717000.parquet",
+ Option.empty(),
+ "/table/path",
+ "/table/path/partition1",
+ "read_optimized",
+ "2026012603471700" + splitNum,
+ fileId,
+ Option.empty());
+ }
+}
