You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2026/07/17 00:51:28 UTC
(hudi) branch master updated: fix(flink): remove the Source V2 read teardown race via materialized bounded minibatches (#19202)
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 d10b868d90c9 fix(flink): remove the Source V2 read teardown race via materialized bounded minibatches (#19202)
d10b868d90c9 is described below
commit d10b868d90c97bd7297c0ac18fcb7fcf3deac8af
Author: Vova Kolmakov <wo...@gmail.com>
AuthorDate: Fri Jul 17 07:51:08 2026 +0700
fix(flink): remove the Source V2 read teardown race via materialized bounded minibatches (#19202)
* fix(flink): make CDC read iterator teardown thread-safe to de-flake stream-read ITs
CdcFileSplitsIterator is drained on the Flink task thread but can be closed from the split-fetcher thread during job teardown, racing on its non-volatile recordIterator/imageManager fields (and the nested BaseImageIterator.cdcItr) and surfacing an NPE that flakes testStreamReadFromSpecifiedCommitWithChangelog.
Synchronize hasNext()/next()/close() on the iterator and add a closed flag so a concurrent close() cannot null out state mid-read and the iterator reports a well-defined end-of-stream afterwards. Extend the test's isAcceptableTerminalFailure teardown-race tolerance to the CDC-iterator frame, since BatchRecords drives hasNext()/next() as separate calls and a close() between them still ends a force-terminated drain with a benign NoSuchElementException.
* addressed review comments: split next() guard and use class refs for CDC teardown-frame check
* test(flink): retry a bare await timeout in the CDC stream-read IT
The flink2.1 CI shard failed testStreamReadFromSpecifiedCommitWithChangelog with
"Unexpected job failure" caused by a TimeoutException from tableResult.await(): the
streaming read had not collected its expected rows within the window (CI-load slowness,
not the teardown race this PR already tolerates and retries). A bare TimeoutException
was rethrown as an AssertionError, bypassing submitAndFetchWithRetry, which only retries
short results.
Treat a bare await timeout as a retryable short read: cancel the still-running job and
return the rows collected so far so the retry loop re-submits a fresh job. Also widen
the await window from 30s to 60s so a slow shard is less likely to time out at all.
Test-only; no production change.
* addressed review comments: use equals for CDC teardown-frame class match
* addressed review comments: redesign Source V2 read path to materialize bounded minibatches
The reviewers asked for a root-cause fix rather than the CdcFileSplitsIterator synchronization mitigation. The real flaw was that BatchRecords did lazy reading: it held a live ClosableIterator drained on the Flink task thread, while a forced-cancel teardown closed that same iterator on the split-fetcher thread, so read and close genuinely ran on different threads.
This reworks the Source V2 read path so a reader function is a per-split cursor that owns the record iterator, the CdcImageManager and the file-group readers. HoodieSourceSplitReader.fetch() opens a split, reads one bounded minibatch (DEFAULT_MINI_BATCH_SIZE), and closes the split's resources on the same split-fetcher thread on EOF, failure or cancellation. BatchRecords now carries a materialized list instead of a live iterator, so read and close happen on one thread and the cross-thr [...]
Because Flink's columnar readers and the CDC/MOR row projections return the same reused RowData on every next(), readBatch deep-copies each record before buffering it and re-applies the RowKind, which the serializer does not round-trip.
With read and close single-threaded, the CdcFileSplitsIterator synchronization and closed-flag guards are reverted, and the test-side teardown-race tolerance in ITTestHoodieDataSource is removed: isAcceptableTerminalFailure now accepts only the SuccessException happy path. The independent await-timeout retry is kept.
* addressed review comments: narrow the await-timeout retry to a bare top-level TimeoutException
TableResult.await(long, TimeUnit) throws its own timeout bare, so only a top-level TimeoutException means the await window elapsed with the job still running. A genuine job failure arrives wrapped in an ExecutionException that may itself embed a TimeoutException (checkpoint expiry, RPC timeout); walking the whole cause chain misclassified such a failure as a slow shard, cancelled and retried it, and then reported a row-count mismatch with the real cause discarded. isAwaitTimeout now i [...]
* addressed review comments: regroup copySerializer field and rename test open-count accessor
* addressed review comments: drop redundant RowKind re-apply after RowDataSerializer.copy
* addressed review comments: unblock fetch() on wakeUp and close the file-group reader on init failure
P1: fetch() now drains an I/O-backed minibatch, so wakeUp() sets a volatile flag the drain polls between records; on a wake-up fetch() returns promptly without finishing/closing, keeping the split-close on the fetcher thread.
P2: HoodieSplitReaderFunction.createRecordIterator retains the HoodieFileGroupReader in a local and closes it when getClosableIterator() fails, suppressing the close error onto the original exception.
---------
Co-authored-by: Vova Kolmakov <wo...@apache.org>
---
.../apache/hudi/source/reader/BatchRecords.java | 64 ++--
.../source/reader/HoodieSourceSplitReader.java | 80 ++--
.../function/AbstractSplitReaderFunction.java | 107 +++++-
.../function/HoodieCdcSplitReaderFunction.java | 27 +-
.../reader/function/HoodieSplitReaderFunction.java | 43 ++-
.../reader/function/SplitReaderFunction.java | 46 ++-
.../apache/hudi/table/format/cdc/CdcIterators.java | 5 +
.../hudi/source/reader/TestBatchRecords.java | 249 +++---------
.../source/reader/TestHoodieSourceSplitReader.java | 418 ++++++++++++++++++---
.../function/TestAbstractSplitReaderFunction.java | 153 +++++++-
.../function/TestHoodieCdcSplitReaderFunction.java | 11 +-
.../function/TestHoodieSplitReaderFunction.java | 92 ++++-
.../apache/hudi/table/ITTestHoodieDataSource.java | 148 +++-----
13 files changed, 992 insertions(+), 451 deletions(-)
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/BatchRecords.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/BatchRecords.java
index c69b7634ad1a..95db8365ac7b 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/BatchRecords.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/BatchRecords.java
@@ -19,7 +19,6 @@
package org.apache.hudi.source.reader;
import org.apache.hudi.common.util.ValidationUtils;
-import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.flink.connector.base.source.reader.RecordsBySplits;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
@@ -27,41 +26,50 @@ import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import javax.annotation.Nullable;
import java.util.Collections;
+import java.util.List;
import java.util.Set;
/**
- * Implementation of RecordsWithSplitIds with a list record inside.
+ * Implementation of {@link RecordsWithSplitIds} backed by a materialized, bounded minibatch of
+ * records for a single split.
*
- * Type parameters: <T> – record type
+ * <p>The records are already drained (and copied) on the split-fetcher thread before this batch is
+ * enqueued, so {@link #nextRecordFromSplit()} only walks an in-memory list and holds no live I/O
+ * resource. Record reading and resource teardown therefore stay on the same (split-fetcher) thread,
+ * which removes the cross-thread teardown race a live-iterator batch would be exposed to. The
+ * underlying iterator and its I/O resources are owned and closed by the reader function (see
+ * {@code AbstractSplitReaderFunction#closeCurrentSplit}).
+ *
+ * @param <T> record type
*/
public class BatchRecords<T> implements RecordsWithSplitIds<HoodieRecordWithPosition<T>> {
private String splitId;
private String nextSplitId;
- private final ClosableIterator<T> recordIterator;
+ private final List<T> records;
private final Set<String> finishedSplits;
private final HoodieRecordWithPosition<T> recordAndPosition;
- // point to current read position within the records list
- private int position;
+ // points to the current read position within the records list
+ private int index;
BatchRecords(
String splitId,
- ClosableIterator<T> recordIterator,
+ List<T> records,
int fileOffset,
long startingRecordOffset,
Set<String> finishedSplits) {
ValidationUtils.checkArgument(
finishedSplits != null, "finishedSplits can be empty but not null");
ValidationUtils.checkArgument(
- recordIterator != null, "recordIterator can be empty but not null");
+ records != null, "records can be empty but not null");
this.splitId = splitId;
this.nextSplitId = splitId;
- this.recordIterator = recordIterator;
+ this.records = records;
this.finishedSplits = finishedSplits;
this.recordAndPosition = new HoodieRecordWithPosition<>();
this.recordAndPosition.set(null, fileOffset, startingRecordOffset);
- this.position = 0;
+ this.index = 0;
}
@Nullable
@@ -69,7 +77,7 @@ public class BatchRecords<T> implements RecordsWithSplitIds<HoodieRecordWithPosi
public String nextSplit() {
if (splitId.equals(nextSplitId)) {
// set the nextSplitId to null to indicate no more splits
- // this class only contains record for one split
+ // this class only contains records for one split
nextSplitId = null;
return splitId;
} else {
@@ -80,12 +88,11 @@ public class BatchRecords<T> implements RecordsWithSplitIds<HoodieRecordWithPosi
@Nullable
@Override
public HoodieRecordWithPosition<T> nextRecordFromSplit() {
- if (recordIterator.hasNext()) {
- recordAndPosition.record(recordIterator.next());
- position = position + 1;
+ if (index < records.size()) {
+ recordAndPosition.record(records.get(index));
+ index++;
return recordAndPosition;
} else {
- recordIterator.close();
return null;
}
}
@@ -97,29 +104,14 @@ public class BatchRecords<T> implements RecordsWithSplitIds<HoodieRecordWithPosi
@Override
public void recycle() {
- if (recordIterator != null) {
- recordIterator.close();
- }
- }
-
- public void seek(long startingRecordOffset) {
- for (long i = 0; i < startingRecordOffset; ++i) {
- if (recordIterator.hasNext()) {
- position = position + 1;
- recordIterator.next();
- } else {
- throw new IllegalStateException(
- String.format(
- "Invalid starting record offset %d for split %s",
- startingRecordOffset,
- splitId));
- }
- }
+ // No-op: the minibatch is fully materialized, so there is no live iterator or I/O resource to
+ // release here. The underlying reader is owned and closed by the reader function on the
+ // split-fetcher thread (AbstractSplitReaderFunction#closeCurrentSplit).
}
public static <T> BatchRecords<T> forRecords(
- String splitId, ClosableIterator<T> recordIterator, int fileOffset, long startingRecordOffset) {
- return new BatchRecords<>(splitId, recordIterator, fileOffset, startingRecordOffset, Set.of());
+ String splitId, List<T> records, int fileOffset, long startingRecordOffset) {
+ return new BatchRecords<>(splitId, records, fileOffset, startingRecordOffset, Set.of());
}
public static <T> RecordsWithSplitIds<HoodieRecordWithPosition<T>> lastBatchRecords(String splitId) {
@@ -130,4 +122,4 @@ public class BatchRecords<T> implements RecordsWithSplitIds<HoodieRecordWithPosi
// in SourceReaderBase for bounded (batch) reads.
return new RecordsBySplits<>(Collections.emptyMap(), Set.of(splitId));
}
-}
\ No newline at end of file
+}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieSourceSplitReader.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieSourceSplitReader.java
index 04c55e6a2963..e53c5f82eed9 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieSourceSplitReader.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieSourceSplitReader.java
@@ -45,21 +45,31 @@ import java.util.Set;
/**
* The split reader of Hoodie source.
*
- * <p>Each call to {@link #fetch()} reads one split and returns it as a single
- * {@link RecordsWithSplitIds} batch. Flink's {@code SourceReaderBase} is responsible for
- * draining all records from the batch (via {@code nextRecordFromSplit()}) and marking
- * the split finished (via {@code finishedSplits()}) before calling {@link #fetch()} again.
+ * <p>Each call to {@link #fetch()} returns one bounded minibatch of the currently open split, so a
+ * single split spans multiple {@code fetch()} calls. When the open split is exhausted its resources
+ * are closed on this (split-fetcher) thread and a finish signal ({@code finishedSplits()}) is
+ * returned so Flink's {@code SourceReaderBase} can advance / reach end-of-input. All record reading
+ * and resource teardown for a split therefore happen on the same thread, which removes the
+ * cross-thread teardown race a live-iterator batch would be exposed to.
*
* @param <T> record type
*/
@Slf4j
public class HoodieSourceSplitReader<T> implements SplitReader<HoodieRecordWithPosition<T>, HoodieSourceSplit> {
+ // Upper bound on the number of records materialized per fetch() call (one minibatch). Kept as a
+ // fixed constant for now (mirrors RecordIterators.DEFAULT_BATCH_SIZE); it can be promoted to a
+ // Flink option later if a tunable per-fetch bound is ever needed.
+ private static final int DEFAULT_MINI_BATCH_SIZE = 2048;
+
private final SerializableComparator<HoodieSourceSplit> splitComparator;
private final Queue<HoodieSourceSplit> splits;
private final FlinkStreamReadMetrics readerMetrics;
private final SplitReaderFunction<T> readerFunction;
private final Option<RecordLimiter> recordLimiter;
private transient HoodieSourceSplit currentSplit;
+ // Set by wakeUp() (possibly from another thread) to stop the in-flight minibatch drain promptly.
+ // Reset at the start of every fetch(), so it only ever means "a wakeUp() landed during THIS fetch".
+ private volatile boolean wokenUp;
public HoodieSourceSplitReader(
String tableName,
@@ -77,27 +87,47 @@ public class HoodieSourceSplitReader<T> implements SplitReader<HoodieRecordWithP
@Override
public RecordsWithSplitIds<HoodieRecordWithPosition<T>> fetch() throws IOException {
- // finish current split.
- if (currentSplit != null) {
- return finishSplit();
+ // A wakeUp() only needs to unblock an in-progress fetch(); Flink's SplitFetcher drives shutdown
+ // off its own 'closed' flag (set before wakeUp() and checked before the next fetch()), not off a
+ // lasting wakeUp effect. Start each cycle from a clean flag so the drain below reacts only to a
+ // wakeUp that lands during THIS fetch.
+ wokenUp = false;
+ if (currentSplit == null) {
+ // Limit already satisfied: drain any remaining locally-queued splits as immediately finished
+ // so that Flink's SourceReaderBase can reach end-of-input cleanly.
+ if (recordLimiter.map(RecordLimiter::isLimitReached).orElse(false)) {
+ return drainRemainingAsSplitsFinished();
+ }
+ HoodieSourceSplit nextSplit = splits.poll();
+ if (nextSplit == null) {
+ // return an empty result, which will lead to split fetch to be idle.
+ // SplitFetcherManager will then close idle fetcher.
+ return new RecordsBySplits<>(Collections.emptyMap(), Collections.emptySet());
+ }
+ currentSplit = nextSplit;
+ readerFunction.open(currentSplit);
}
- // Limit already satisfied: drain any remaining locally-queued splits as immediately finished
- // so that Flink's SourceReaderBase can reach end-of-input cleanly.
- if (recordLimiter.map(RecordLimiter::isLimitReached).orElse(false)) {
- return drainRemainingAsSplitsFinished();
+ // Read the next bounded minibatch of the open split, unless the global limit is already reached.
+ if (!recordLimiter.map(RecordLimiter::isLimitReached).orElse(false)) {
+ BatchRecords<T> batch = readerFunction.readBatch(currentSplit, DEFAULT_MINI_BATCH_SIZE, () -> wokenUp);
+ if (batch != null) {
+ // Partial (woken) or full minibatch; the split is not finished either way.
+ return recordLimiter.map(rl -> rl.wrap(batch)).orElse(batch);
+ }
+ if (wokenUp) {
+ // Woken before any record was buffered: return promptly WITHOUT finishing or closing the
+ // split, so it resumes on the next fetch(), or the fetcher observes shutdown and closes it
+ // on this (split-fetcher) thread. This branch must stay inside the !isLimitReached block:
+ // a wakeUp coinciding with the limit-reached path below must still finish the split.
+ return new RecordsBySplits<>(Collections.emptyMap(), Collections.emptySet());
+ }
}
- HoodieSourceSplit nextSplit = splits.poll();
- if (nextSplit != null) {
- currentSplit = nextSplit;
- RecordsWithSplitIds<HoodieRecordWithPosition<T>> records = readerFunction.read(nextSplit);
- return recordLimiter.map(rl -> rl.wrap(records)).orElse(records);
- } else {
- // return an empty result, which will lead to split fetch to be idle.
- // SplitFetcherManager will then close idle fetcher.
- return new RecordsBySplits<>(Collections.emptyMap(), Collections.emptySet());
- }
+ // Split exhausted (or the limit was reached mid-split): close its resources on this
+ // (split-fetcher) thread first, then emit the finish signal so SourceReaderBase can advance.
+ readerFunction.closeCurrentSplit();
+ return finishSplit();
}
@Override
@@ -120,13 +150,17 @@ public class HoodieSourceSplitReader<T> implements SplitReader<HoodieRecordWithP
@Override
public void wakeUp() {
- // Nothing to do
+ // Flink calls this (while holding SplitFetcher.lock) to unblock a fetch() that is draining a
+ // minibatch, e.g. on shutdown. Keep it a non-blocking plain volatile write; the drain loop in
+ // readBatch polls the flag between records and returns promptly. The actual resource teardown
+ // still happens on the split-fetcher thread via close()/closeCurrentSplit().
+ wokenUp = true;
}
/**
* SourceSplitReader only reads splits sequentially. When waiting for watermark alignment
* the SourceOperator will stop processing and recycling the fetched batches. Based on this the
- * {@code pauseOrResumeSplits} and the {@code wakeUp} are left empty.
+ * {@code pauseOrResumeSplits} is left empty.
* @param splitsToPause splits to pause
* @param splitsToResume splits to resume
*/
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/AbstractSplitReaderFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/AbstractSplitReaderFunction.java
index 0f97b612fa21..d317dd7baa76 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/AbstractSplitReaderFunction.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/AbstractSplitReaderFunction.java
@@ -18,19 +18,40 @@
package org.apache.hudi.source.reader.function;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.configuration.HadoopConfigurations;
import org.apache.hudi.source.ExpressionPredicates;
+import org.apache.hudi.source.reader.BatchRecords;
+import org.apache.hudi.source.split.HoodieSourceSplit;
import org.apache.hudi.table.format.InternalSchemaManager;
import org.apache.hudi.util.FlinkWriteClients;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.types.logical.RowType;
+import java.util.ArrayList;
import java.util.List;
+import java.util.function.BooleanSupplier;
/**
- * Abstract implementation of SplitReaderFunction.
+ * Abstract implementation of {@link SplitReaderFunction} that provides the per-split cursor
+ * machinery shared by all reader functions.
+ *
+ * <p>A subclass only supplies how to create the record iterator for a split
+ * ({@link #createRecordIterator(HoodieSourceSplit)}) and the {@link RowType} of the records it
+ * produces ({@link #producedRowType()}); this base drives {@link #open(HoodieSourceSplit)},
+ * {@link #readBatch(HoodieSourceSplit, int, java.util.function.BooleanSupplier)}, {@link #closeCurrentSplit()} and {@link #close()},
+ * all on the single split-fetcher thread.
+ *
+ * <p>Because Flink's columnar readers (and the CDC/MOR row projections) return the same reused
+ * {@link RowData} object on every {@code next()}, {@link #readBatch} copies each record before
+ * buffering it - materializing raw references would make every entry in a minibatch alias the last
+ * row. {@link RowDataSerializer#copy(RowData)} preserves the record's {@code RowKind}, so no
+ * re-apply is needed.
*/
public abstract class AbstractSplitReaderFunction implements SplitReaderFunction<RowData> {
@@ -40,6 +61,11 @@ public abstract class AbstractSplitReaderFunction implements SplitReaderFunction
protected final boolean emitDelete;
private transient HoodieWriteConfig writeConfig;
private transient org.apache.hadoop.conf.Configuration hadoopConf;
+ private transient RowDataSerializer copySerializer;
+
+ // Per-split cursor state (split-fetcher thread only).
+ private transient ClosableIterator<RowData> currentIterator;
+ private transient long nextRecordOffset;
public AbstractSplitReaderFunction(
Configuration conf,
@@ -52,6 +78,85 @@ public abstract class AbstractSplitReaderFunction implements SplitReaderFunction
this.emitDelete = emitDelete;
}
+ /**
+ * Creates the record iterator (and its underlying I/O resources) for {@code split}. Closing the
+ * returned iterator must release all of those resources.
+ */
+ protected abstract ClosableIterator<RowData> createRecordIterator(HoodieSourceSplit split);
+
+ /** The {@link RowType} of the records produced by {@link #createRecordIterator}. */
+ protected abstract RowType producedRowType();
+
+ @Override
+ public void open(HoodieSourceSplit split) {
+ this.currentIterator = createRecordIterator(split);
+ try {
+ // Skip the records already emitted before the last checkpoint so a recovered split resumes at
+ // the right position; matches the validation the old BatchRecords#seek performed.
+ long consumed = split.getConsumed();
+ for (long i = 0; i < consumed; i++) {
+ if (currentIterator.hasNext()) {
+ currentIterator.next();
+ } else {
+ throw new IllegalStateException(
+ String.format("Invalid starting record offset %d for split %s", consumed, split.splitId()));
+ }
+ }
+ this.nextRecordOffset = consumed;
+ } catch (RuntimeException | Error e) {
+ // Close on failure so the split's I/O resources are released even if the resume-skip fails.
+ closeCurrentSplit();
+ throw e;
+ }
+ }
+
+ @Override
+ public BatchRecords<RowData> readBatch(HoodieSourceSplit split, int batchSize, BooleanSupplier wakeupSignal) {
+ ValidationUtils.checkState(currentIterator != null,
+ "readBatch called before open for split " + split.splitId());
+ RowDataSerializer serializer = getCopySerializer();
+ List<RowData> buffer = new ArrayList<>();
+ try {
+ // Poll wakeupSignal between records so a wakeUp() lands promptly: materialization stops early
+ // and whatever is buffered so far is returned as a partial minibatch (null if nothing yet).
+ while (buffer.size() < batchSize && !wakeupSignal.getAsBoolean() && currentIterator.hasNext()) {
+ RowData next = currentIterator.next();
+ buffer.add(serializer.copy(next));
+ }
+ } catch (RuntimeException | Error e) {
+ // Close on failure so the split's I/O resources are released even if a mid-drain read fails.
+ closeCurrentSplit();
+ throw e;
+ }
+ if (buffer.isEmpty()) {
+ return null;
+ }
+ long startingRecordOffset = nextRecordOffset;
+ nextRecordOffset += buffer.size();
+ return BatchRecords.forRecords(split.splitId(), buffer, split.getFileOffset(), startingRecordOffset);
+ }
+
+ @Override
+ public void closeCurrentSplit() {
+ if (currentIterator != null) {
+ currentIterator.close();
+ currentIterator = null;
+ }
+ nextRecordOffset = 0;
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeCurrentSplit();
+ }
+
+ private RowDataSerializer getCopySerializer() {
+ if (copySerializer == null) {
+ copySerializer = new RowDataSerializer(producedRowType());
+ }
+ return copySerializer;
+ }
+
protected HoodieWriteConfig getWriteConfig() {
if (writeConfig == null) {
writeConfig = FlinkWriteClients.getHoodieClientConfig(conf);
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java
index b539bc835f14..2f6afd833534 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieCdcSplitReaderFunction.java
@@ -40,8 +40,6 @@ import org.apache.hudi.configuration.OptionsResolver;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.source.ExpressionPredicates;
-import org.apache.hudi.source.reader.BatchRecords;
-import org.apache.hudi.source.reader.HoodieRecordWithPosition;
import org.apache.hudi.source.split.HoodieCdcSourceSplit;
import org.apache.hudi.source.split.HoodieSourceSplit;
import org.apache.hudi.table.format.FilePathUtils;
@@ -56,9 +54,9 @@ import org.apache.hudi.table.format.mor.MergeOnReadTableState;
import org.apache.hudi.util.StreamerUtil;
import lombok.extern.slf4j.Slf4j;
-import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
import org.apache.hadoop.fs.Path;
import java.io.IOException;
@@ -81,7 +79,6 @@ public class HoodieCdcSplitReaderFunction extends AbstractSplitReaderFunction {
private final List<DataType> fieldTypes;
private final MergeOnReadTableState tableState;
private transient HoodieTableMetaClient metaClient;
- private transient ClosableIterator<RowData> currentIterator;
// Fallback reader for non-CDC splits (e.g. snapshot reads when read.start-commit='earliest')
private transient HoodieSplitReaderFunction fallbackReaderFunction;
@@ -110,12 +107,12 @@ public class HoodieCdcSplitReaderFunction extends AbstractSplitReaderFunction {
}
@Override
- public RecordsWithSplitIds<HoodieRecordWithPosition<RowData>> read(HoodieSourceSplit split) {
+ protected ClosableIterator<RowData> createRecordIterator(HoodieSourceSplit split) {
if (!(split instanceof HoodieCdcSourceSplit)) {
// Non-CDC splits arrive when reading from 'earliest' with no prior CDC history
// (i.e. instantRange is empty → snapshot path). Fall back to the standard MOR reader
// which emits all records as INSERT rows, matching the expected snapshot behaviour.
- return getFallbackReaderFunction().read(split);
+ return getFallbackReaderFunction().createRecordIterator(split);
}
HoodieCdcSourceSplit cdcSplit = (HoodieCdcSourceSplit) split;
@@ -132,21 +129,15 @@ public class HoodieCdcSplitReaderFunction extends AbstractSplitReaderFunction {
mode,
imageManager);
- currentIterator = new CdcIterators.CdcFileSplitsIterator(cdcSplit.getChanges(), imageManager, recordIteratorFunc);
- BatchRecords<RowData> records = BatchRecords.forRecords(
- split.splitId(), currentIterator, split.getFileOffset(), split.getConsumed());
- records.seek(split.getConsumed());
- return records;
+ // The CdcFileSplitsIterator owns the imageManager and its per-split record iterators; closing it
+ // (via the base class closeCurrentSplit) releases them. The base class handles the consumed-offset
+ // skip and the minibatch materialization uniformly with the MOR/COW path.
+ return new CdcIterators.CdcFileSplitsIterator(cdcSplit.getChanges(), imageManager, recordIteratorFunc);
}
@Override
- public void close() throws Exception {
- if (currentIterator != null) {
- currentIterator.close();
- }
- if (fallbackReaderFunction != null) {
- fallbackReaderFunction.close();
- }
+ protected RowType producedRowType() {
+ return tableState.getRequiredRowType();
}
// -------------------------------------------------------------------------
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieSplitReaderFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieSplitReaderFunction.java
index 96ade6d4d77b..0ebc5c39cdbb 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieSplitReaderFunction.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieSplitReaderFunction.java
@@ -29,16 +29,15 @@ import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.source.ExpressionPredicates;
-import org.apache.hudi.source.reader.BatchRecords;
-import org.apache.hudi.source.reader.HoodieRecordWithPosition;
import org.apache.hudi.source.split.HoodieSourceSplit;
import org.apache.hudi.table.format.FormatUtils;
import org.apache.hudi.table.format.InternalSchemaManager;
+import org.apache.hudi.util.HoodieSchemaConverter;
import org.apache.hudi.util.StreamerUtil;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
import java.io.IOException;
import java.util.Collections;
@@ -52,7 +51,6 @@ public class HoodieSplitReaderFunction extends AbstractSplitReaderFunction {
private final HoodieSchema tableSchema;
private final HoodieSchema requiredSchema;
private final String mergeType;
- private transient HoodieFileGroupReader<RowData> fileGroupReader;
public HoodieSplitReaderFunction(
Configuration configuration,
@@ -72,28 +70,39 @@ public class HoodieSplitReaderFunction extends AbstractSplitReaderFunction {
}
@Override
- public RecordsWithSplitIds<HoodieRecordWithPosition<RowData>> read(HoodieSourceSplit split) {
- final String splitId = split.splitId();
+ protected ClosableIterator<RowData> createRecordIterator(HoodieSourceSplit split) {
HoodieTableMetaClient metaClient = StreamerUtil.metaClientForReader(conf, getHadoopConf());
-
+ // Closing the returned iterator cascade-closes the whole HoodieFileGroupReader, so the base
+ // class only has to close the iterator in closeCurrentSplit(). But getClosableIterator() runs
+ // initRecordIterators(), which opens the reader's base-file iterator / record buffer before the
+ // wrapping iterator is returned; if it throws, the reader is only a local here and nothing else
+ // would close it. Keep it in a local and close it in the failure path.
+ HoodieFileGroupReader<RowData> fileGroupReader = createFileGroupReader(split, metaClient);
try {
- this.fileGroupReader = createFileGroupReader(split, metaClient);
- final ClosableIterator<RowData> recordIterator = fileGroupReader.getClosableIterator();
- BatchRecords<RowData> records = BatchRecords.forRecords(splitId, recordIterator, split.getFileOffset(), split.getConsumed());
- records.seek(split.getConsumed());
- return records;
+ return fileGroupReader.getClosableIterator();
} catch (IOException e) {
+ closeSuppressing(fileGroupReader, e);
throw new HoodieIOException("Failed to read from file group: " + split.getFileId(), e);
+ } catch (RuntimeException | Error e) {
+ closeSuppressing(fileGroupReader, e);
+ throw e;
}
}
- @Override
- public void close() throws Exception {
- if (fileGroupReader != null) {
- fileGroupReader.close();
+ /** Closes {@code reader}, attaching any close failure to {@code primary} as a suppressed exception. */
+ private static void closeSuppressing(HoodieFileGroupReader<RowData> reader, Throwable primary) {
+ try {
+ reader.close();
+ } catch (Exception closeError) {
+ primary.addSuppressed(closeError);
}
}
+ @Override
+ protected RowType producedRowType() {
+ return HoodieSchemaConverter.convertToRowType(requiredSchema);
+ }
+
/**
* Creates a {@link HoodieFileGroupReader} for the given split.
*
@@ -101,7 +110,7 @@ public class HoodieSplitReaderFunction extends AbstractSplitReaderFunction {
* @param metaClient The table meta client for schema and config resolution
* @return A {@link HoodieFileGroupReader} instance
*/
- private HoodieFileGroupReader<RowData> createFileGroupReader(HoodieSourceSplit split, HoodieTableMetaClient metaClient) {
+ protected HoodieFileGroupReader<RowData> createFileGroupReader(HoodieSourceSplit split, HoodieTableMetaClient metaClient) {
// Create FileSlice from split information
FileSlice fileSlice = new FileSlice(
new HoodieFileGroupId(split.getPartitionPath(), split.getFileId()),
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/SplitReaderFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/SplitReaderFunction.java
index 6f7bf0f18ebe..46fb343aa3d4 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/SplitReaderFunction.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/SplitReaderFunction.java
@@ -18,19 +18,55 @@
package org.apache.hudi.source.reader.function;
-import org.apache.hudi.source.reader.HoodieRecordWithPosition;
+import org.apache.hudi.source.reader.BatchRecords;
import org.apache.hudi.source.split.HoodieSourceSplit;
-import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
-
import java.io.Serializable;
+import java.util.function.BooleanSupplier;
/**
- * Interface for split read function.
+ * Interface for a split read function.
+ *
+ * <p>A reader function is a stateful, per-split cursor driven entirely on the Flink split-fetcher
+ * thread by {@link org.apache.hudi.source.reader.HoodieSourceSplitReader#fetch()}:
+ * {@link #open(HoodieSourceSplit)} creates the record iterator and its underlying I/O resources for
+ * a split, {@link #readBatch(HoodieSourceSplit, int, java.util.function.BooleanSupplier)} drains the next bounded minibatch, and
+ * {@link #closeCurrentSplit()} releases the split's resources once it is exhausted. Because open,
+ * read and close all run on the same thread, no record or I/O resource is ever touched concurrently.
+ *
+ * @param <T> record type
*/
public interface SplitReaderFunction<T> extends Serializable {
- RecordsWithSplitIds<HoodieRecordWithPosition<T>> read(HoodieSourceSplit split);
+ /**
+ * Opens {@code split} for reading: creates the record iterator and its underlying I/O resources,
+ * and skips the records already consumed ({@link HoodieSourceSplit#getConsumed()}) so a recovered
+ * split resumes at the right position.
+ */
+ void open(HoodieSourceSplit split);
+
+ /**
+ * Drains up to {@code batchSize} records from the currently open split into a materialized
+ * {@link BatchRecords} minibatch. Returns {@code null} once the split is exhausted.
+ *
+ * <p>{@code wakeupSignal} is polled between records: once it returns {@code true} materialization
+ * stops early and the records buffered so far are returned as a (non-finishing) partial minibatch;
+ * if nothing has been buffered yet {@code null} is returned. This lets a blocking {@code fetch()}
+ * unblock promptly on {@link org.apache.hudi.source.reader.HoodieSourceSplitReader#wakeUp()}
+ * without touching any resource off the split-fetcher thread.
+ */
+ BatchRecords<T> readBatch(HoodieSourceSplit split, int batchSize, BooleanSupplier wakeupSignal);
+
+ /**
+ * Closes the currently open split's iterator and I/O resources. Called when the split is
+ * exhausted, a read fails, or the read is stopped early. Safe to call when no split is open.
+ */
+ void closeCurrentSplit();
+ /**
+ * Closes the reader function entirely (idempotent). Invoked by
+ * {@link org.apache.hudi.source.reader.HoodieSourceSplitReader#close()} on the split-fetcher
+ * thread.
+ */
void close() throws Exception;
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java
index a4a0c33d6eca..104276583ae0 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java
@@ -101,6 +101,11 @@ public final class CdcIterators {
/**
* Iterates over an ordered sequence of {@link HoodieCDCFileSplit}s, delegating
* per-split record reading to a user-supplied factory function.
+ *
+ * <p>Not thread-safe by design: in the Source V2 read path this iterator is created, drained into a
+ * materialized minibatch, and closed entirely on the single split-fetcher thread (see
+ * {@code AbstractSplitReaderFunction}); the legacy {@link CdcInputFormat} path likewise reads and
+ * closes it on one thread. No method is ever invoked concurrently, so no synchronization is needed.
*/
public static class CdcFileSplitsIterator implements ClosableIterator<RowData> {
private CdcImageManager imageManager;
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/TestBatchRecords.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/TestBatchRecords.java
index 69af5702f1f4..5b111dddcc03 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/TestBatchRecords.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/TestBatchRecords.java
@@ -18,14 +18,11 @@
package org.apache.hudi.source.reader;
-import org.apache.hudi.common.util.collection.ClosableIterator;
-
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
import java.util.Set;
@@ -36,16 +33,15 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
- * Test cases for {@link BatchRecords}.
+ * Test cases for {@link BatchRecords}, which now holds a materialized, bounded minibatch of records.
*/
public class TestBatchRecords {
@Test
- public void testForRecordsWithEmptyIterator() {
+ public void testForRecordsWithEmptyList() {
String splitId = "test-split-1";
- ClosableIterator<String> emptyIterator = createClosableIterator(Collections.emptyList());
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, emptyIterator, 0, 0L);
+ BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, Collections.emptyList(), 0, 0L);
assertNotNull(batchRecords);
assertEquals(splitId, batchRecords.nextSplit());
@@ -57,9 +53,8 @@ public class TestBatchRecords {
public void testForRecordsWithMultipleRecords() {
String splitId = "test-split-2";
List<String> records = Arrays.asList("record1", "record2", "record3");
- ClosableIterator<String> iterator = createClosableIterator(records);
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, iterator, 0, 0L);
+ BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, records, 0, 0L);
// Verify split ID
assertEquals(splitId, batchRecords.nextSplit());
@@ -86,46 +81,13 @@ public class TestBatchRecords {
assertNull(batchRecords.nextRecordFromSplit());
}
- @Test
- public void testSeekToStartingOffset() {
- String splitId = "test-split-3";
- List<String> records = Arrays.asList("record1", "record2", "record3", "record4", "record5");
- ClosableIterator<String> iterator = createClosableIterator(records);
-
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, iterator, 0, 2L);
- batchRecords.seek(2L);
-
- // After seeking to offset 2, we should start from record3
- batchRecords.nextSplit();
-
- HoodieRecordWithPosition<String> record = batchRecords.nextRecordFromSplit();
- assertNotNull(record);
- assertEquals("record3", record.record());
- }
-
- @Test
- public void testSeekBeyondAvailableRecords() {
- String splitId = "test-split-4";
- List<String> records = Arrays.asList("record1", "record2");
- ClosableIterator<String> iterator = createClosableIterator(records);
-
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, iterator, 0, 0L);
-
- IllegalStateException exception = assertThrows(IllegalStateException.class, () -> {
- batchRecords.seek(10L);
- });
-
- assertTrue(exception.getMessage().contains("Invalid starting record offset"));
- }
-
@Test
public void testFileOffsetPersistence() {
String splitId = "test-split-5";
int fileOffset = 5;
List<String> records = Arrays.asList("record1", "record2");
- ClosableIterator<String> iterator = createClosableIterator(records);
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, iterator, fileOffset, 0L);
+ BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, records, fileOffset, 0L);
batchRecords.nextSplit();
HoodieRecordWithPosition<String> record1 = batchRecords.nextRecordFromSplit();
@@ -140,12 +102,11 @@ public class TestBatchRecords {
@Test
public void testConstructorWithFinishedSplits() {
String splitId = "test-split-7";
- List<String> records = Arrays.asList("record1");
- ClosableIterator<String> iterator = createClosableIterator(records);
+ List<String> records = Collections.singletonList("record1");
Set<String> finishedSplits = new HashSet<>(Arrays.asList("split1", "split2"));
BatchRecords<String> batchRecords = new BatchRecords<>(
- splitId, iterator, 0, 0L, finishedSplits);
+ splitId, records, 0, 0L, finishedSplits);
assertEquals(2, batchRecords.finishedSplits().size());
assertTrue(batchRecords.finishedSplits().contains("split1"));
@@ -157,10 +118,9 @@ public class TestBatchRecords {
String splitId = "test-split-8";
long startingRecordOffset = 10L;
List<String> records = Arrays.asList("A", "B", "C");
- ClosableIterator<String> iterator = createClosableIterator(records);
BatchRecords<String> batchRecords = BatchRecords.forRecords(
- splitId, iterator, 0, startingRecordOffset);
+ splitId, records, 0, startingRecordOffset);
batchRecords.nextSplit();
// First record should be at startingRecordOffset + 1
@@ -176,13 +136,37 @@ public class TestBatchRecords {
assertEquals(startingRecordOffset + 3, record3.recordOffset());
}
+ @Test
+ public void testOffsetContinuityAcrossMinibatches() {
+ // Two consecutive minibatches of the same split: the reader function threads the running
+ // record offset so that offsets stay monotonic across batch boundaries (batch 2 starts where
+ // batch 1 left off). This mirrors AbstractSplitReaderFunction#readBatch.
+ String splitId = "test-split-continuity";
+
+ BatchRecords<String> batch1 = BatchRecords.forRecords(splitId, Arrays.asList("a", "b", "c"), 0, 0L);
+ batch1.nextSplit();
+ assertEquals(1L, batch1.nextRecordFromSplit().recordOffset());
+ assertEquals(2L, batch1.nextRecordFromSplit().recordOffset());
+ assertEquals(3L, batch1.nextRecordFromSplit().recordOffset());
+ assertNull(batch1.nextRecordFromSplit());
+
+ // batch 2 opens at startingRecordOffset = 3 (the count already consumed by batch 1)
+ BatchRecords<String> batch2 = BatchRecords.forRecords(splitId, Arrays.asList("d", "e"), 0, 3L);
+ batch2.nextSplit();
+ HoodieRecordWithPosition<String> d = batch2.nextRecordFromSplit();
+ assertEquals("d", d.record());
+ assertEquals(4L, d.recordOffset());
+ HoodieRecordWithPosition<String> e = batch2.nextRecordFromSplit();
+ assertEquals("e", e.record());
+ assertEquals(5L, e.recordOffset());
+ }
+
@Test
public void testSplitIdReturnedOnlyOnce() {
String splitId = "test-split-9";
- List<String> records = Arrays.asList("record1");
- ClosableIterator<String> iterator = createClosableIterator(records);
+ List<String> records = Collections.singletonList("record1");
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, iterator, 0, 0L);
+ BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, records, 0, 0L);
assertEquals(splitId, batchRecords.nextSplit());
assertNull(batchRecords.nextSplit());
@@ -191,187 +175,64 @@ public class TestBatchRecords {
}
@Test
- public void testRecycleClosesIterator() {
+ public void testRecycleIsNoOp() {
+ // The minibatch is fully materialized, so recycle() releases nothing and is a harmless no-op:
+ // it must not throw and must not affect the still-readable records.
String splitId = "test-split-10";
List<String> records = Arrays.asList("record1", "record2");
- MockClosableIterator<String> mockIterator = new MockClosableIterator<>(records);
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, mockIterator, 0, 0L);
+ BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, records, 0, 0L);
+ batchRecords.nextSplit();
batchRecords.recycle();
- assertTrue(mockIterator.isClosed(), "Iterator should be closed after recycle");
- }
-
- @Test
- public void testRecycleWithNullIterator() {
- // Test that recycle handles null iterator gracefully (though in practice this shouldn't happen)
- // This tests the null check in recycle() method
- String splitId = "test-split-11";
- ClosableIterator<String> emptyIterator = createClosableIterator(Collections.emptyList());
-
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, emptyIterator, 0, 0L);
-
- // Should not throw exception
- batchRecords.recycle();
+ assertNotNull(batchRecords.nextRecordFromSplit(), "records remain readable after recycle");
+ assertNotNull(batchRecords.nextRecordFromSplit());
+ assertNull(batchRecords.nextRecordFromSplit());
}
@Test
public void testNextRecordFromSplitAfterExhaustion() {
String splitId = "test-split-12";
- List<String> records = Arrays.asList("record1");
- ClosableIterator<String> iterator = createClosableIterator(records);
+ List<String> records = Collections.singletonList("record1");
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, iterator, 0, 0L);
+ BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, records, 0, 0L);
batchRecords.nextSplit();
// Read the only record
assertNotNull(batchRecords.nextRecordFromSplit());
- // After exhaustion, should return null
+ // After exhaustion, should return null (and stay null)
assertNull(batchRecords.nextRecordFromSplit());
assertNull(batchRecords.nextRecordFromSplit());
}
- @Test
- public void testSeekWithZeroOffset() {
- String splitId = "test-split-13";
- List<String> records = Arrays.asList("record1", "record2", "record3");
- ClosableIterator<String> iterator = createClosableIterator(records);
-
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, iterator, 0, 0L);
-
- // Seeking to 0 should not skip any records
- batchRecords.seek(0L);
- batchRecords.nextSplit();
-
- HoodieRecordWithPosition<String> record = batchRecords.nextRecordFromSplit();
- assertNotNull(record);
- assertEquals("record1", record.record());
- }
-
@Test
public void testConstructorNullValidation() {
String splitId = "test-split-14";
- List<String> records = Arrays.asList("record1");
- ClosableIterator<String> iterator = createClosableIterator(records);
+ List<String> records = Collections.singletonList("record1");
// Test null finishedSplits
- assertThrows(IllegalArgumentException.class, () -> {
- new BatchRecords<>(splitId, iterator, 0, 0L, null);
- });
-
- // Test null recordIterator
- assertThrows(IllegalArgumentException.class, () -> {
- new BatchRecords<>(splitId, null, 0, 0L, new HashSet<>());
- });
+ assertThrows(IllegalArgumentException.class, () ->
+ new BatchRecords<>(splitId, records, 0, 0L, null));
+
+ // Test null records
+ assertThrows(IllegalArgumentException.class, () ->
+ new BatchRecords<>(splitId, null, 0, 0L, new HashSet<>()));
}
@Test
public void testRecordPositionReusability() {
String splitId = "test-split-15";
List<String> records = Arrays.asList("A", "B", "C");
- ClosableIterator<String> iterator = createClosableIterator(records);
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, iterator, 0, 0L);
+ BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, records, 0, 0L);
batchRecords.nextSplit();
HoodieRecordWithPosition<String> pos1 = batchRecords.nextRecordFromSplit();
HoodieRecordWithPosition<String> pos2 = batchRecords.nextRecordFromSplit();
- // Should reuse the same object
+ // Should reuse the same object (safe because SourceReaderBase emits each record before the next)
assertTrue(pos1 == pos2, "Should reuse the same HoodieRecordWithPosition object");
}
-
- @Test
- public void testSeekUpdatesPosition() {
- String splitId = "test-split-16";
- List<String> records = Arrays.asList("r1", "r2", "r3", "r4", "r5");
- ClosableIterator<String> iterator = createClosableIterator(records);
-
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, iterator, 5, 10L);
-
- // Seek to offset 3
- batchRecords.seek(3L);
-
- batchRecords.nextSplit();
-
- // After seeking 3, next record should be r4 (4th record)
- HoodieRecordWithPosition<String> record = batchRecords.nextRecordFromSplit();
- assertNotNull(record);
- assertEquals("r4", record.record());
- }
-
- @Test
- public void testIteratorClosedAfterExhaustion() {
- String splitId = "test-split-17";
- List<String> records = Arrays.asList("record1");
- MockClosableIterator<String> mockIterator = new MockClosableIterator<>(records);
-
- BatchRecords<String> batchRecords = BatchRecords.forRecords(splitId, mockIterator, 0, 0L);
- batchRecords.nextSplit();
-
- // Read records
- batchRecords.nextRecordFromSplit();
-
- // Trigger close operation
- batchRecords.nextRecordFromSplit();
-
- // After exhaustion, nextRecordFromSplit should close the iterator
- assertTrue(mockIterator.isClosed(), "Iterator should be closed after exhaustion");
- }
-
- /**
- * Helper method to create a ClosableIterator from a list of items.
- */
- private <T> ClosableIterator<T> createClosableIterator(List<T> items) {
- Iterator<T> iterator = items.iterator();
- return new ClosableIterator<T>() {
- @Override
- public void close() {
- // No-op for test
- }
-
- @Override
- public boolean hasNext() {
- return iterator.hasNext();
- }
-
- @Override
- public T next() {
- return iterator.next();
- }
- };
- }
-
- /**
- * Mock closable iterator for testing close behavior.
- */
- private static class MockClosableIterator<T> implements ClosableIterator<T> {
- private final Iterator<T> iterator;
- private boolean closed = false;
-
- public MockClosableIterator(List<T> items) {
- this.iterator = items.iterator();
- }
-
- @Override
- public void close() {
- closed = true;
- }
-
- @Override
- public boolean hasNext() {
- return iterator.hasNext();
- }
-
- @Override
- public T next() {
- return iterator.next();
- }
-
- public boolean isClosed() {
- return closed;
- }
- }
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/TestHoodieSourceSplitReader.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/TestHoodieSourceSplitReader.java
index 4dbae4896e77..ccb6c2bbf893 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/TestHoodieSourceSplitReader.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/TestHoodieSourceSplitReader.java
@@ -19,7 +19,6 @@
package org.apache.hudi.source.reader;
import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.source.reader.function.SplitReaderFunction;
import org.apache.hudi.source.split.HoodieSourceSplit;
import org.apache.hudi.source.split.SerializableComparator;
@@ -33,11 +32,17 @@ import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -183,13 +188,235 @@ public class TestHoodieSourceSplitReader {
}
@Test
- public void testWakeUp() {
+ public void testWakeUp() throws IOException {
TestSplitReaderFunction readerFunction = new TestSplitReaderFunction();
HoodieSourceSplitReader<String> reader =
new HoodieSourceSplitReader<>(TABLE_NAME, readerContext, readerFunction, null, Option.empty());
- // wakeUp is a no-op, should not throw any exception
+ // wakeUp() now sets a flag but must not throw, and the flag is reset at the top of each fetch()
+ // so a wakeUp with no in-flight drain leaves the next fetch() unaffected.
reader.wakeUp();
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> result = reader.fetch();
+ assertNotNull(result);
+ assertNull(result.nextSplit());
+ }
+
+ // -------------------------------------------------------------------------
+ // wakeUp — cooperative cancellation of the minibatch drain
+ // -------------------------------------------------------------------------
+
+ @Test
+ public void testWakeUpMidDrainReturnsPartialBatchAndResumes() throws IOException {
+ // A wakeUp() landing after the first record stops the drain between records: fetch() returns the
+ // 1 record buffered so far as a NON-finishing batch without closing the split; a later fetch()
+ // resumes the rest with continuous offsets, and the split is closed only once at true EOF.
+ List<String> testData = Arrays.asList("r1", "r2", "r3", "r4", "r5");
+ TestSplitReaderFunction readerFunction = new TestSplitReaderFunction(testData);
+ HoodieSourceSplitReader<String> reader =
+ new HoodieSourceSplitReader<>(TABLE_NAME, readerContext, readerFunction, null, Option.empty());
+ boolean[] fired = {false};
+ readerFunction.setDrainProbe((buffered, hasNext) -> {
+ if (buffered == 1 && !fired[0]) {
+ fired[0] = true;
+ reader.wakeUp();
+ }
+ });
+
+ HoodieSourceSplit split = createTestSplit(1, "file1");
+ reader.handleSplitsChanges(new SplitsAddition<>(Collections.singletonList(split)));
+
+ // First fetch: woken after r1 -> partial batch of 1, split not finished, not closed.
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b1 = reader.fetch();
+ assertEquals(split.splitId(), b1.nextSplit());
+ HoodieRecordWithPosition<String> first = b1.nextRecordFromSplit();
+ assertNotNull(first);
+ assertEquals("r1", first.record());
+ assertEquals(1L, first.recordOffset());
+ assertNull(b1.nextRecordFromSplit(), "drain must stop at the first record on wake-up");
+ assertTrue(b1.finishedSplits().isEmpty(), "woken split must not be finished");
+ assertEquals(0, readerFunction.getCloseCurrentSplitCount(), "woken split must not be closed");
+
+ // Second fetch: resumes the remaining records with continuous offsets (2..5).
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b2 = reader.fetch();
+ assertEquals(split.splitId(), b2.nextSplit());
+ HoodieRecordWithPosition<String> next = b2.nextRecordFromSplit();
+ assertNotNull(next);
+ assertEquals("r2", next.record());
+ assertEquals(2L, next.recordOffset(), "offset continues across the wake boundary");
+ assertEquals(3, drainRecordCount(b2), "r3, r4, r5 remain");
+ assertTrue(b2.finishedSplits().isEmpty());
+
+ // Third fetch: true EOF -> finish signal, split closed exactly once, opened exactly once.
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b3 = reader.fetch();
+ assertTrue(b3.finishedSplits().contains(split.splitId()));
+ assertEquals(1, readerFunction.getCloseCurrentSplitCount());
+ assertEquals(1, readerFunction.getOpenCount());
+ }
+
+ @Test
+ public void testWakeUpBeforeAnyRecordReturnsEmptyNonFinishingBatch() throws IOException {
+ // A wakeUp() landing before any record is buffered must return an empty NON-finishing batch, not
+ // a finish signal: the split stays open (not closed) and resumes on the next fetch().
+ List<String> testData = Arrays.asList("r1", "r2", "r3");
+ TestSplitReaderFunction readerFunction = new TestSplitReaderFunction(testData);
+ HoodieSourceSplitReader<String> reader =
+ new HoodieSourceSplitReader<>(TABLE_NAME, readerContext, readerFunction, null, Option.empty());
+ boolean[] fired = {false};
+ readerFunction.setDrainProbe((buffered, hasNext) -> {
+ // Wake at the start of the drain (count 0) while data is still available.
+ if (buffered == 0 && hasNext && !fired[0]) {
+ fired[0] = true;
+ reader.wakeUp();
+ }
+ });
+
+ HoodieSourceSplit split = createTestSplit(1, "file1");
+ reader.handleSplitsChanges(new SplitsAddition<>(Collections.singletonList(split)));
+
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b1 = reader.fetch();
+ assertNull(b1.nextSplit(), "empty non-finishing batch carries no split records");
+ assertTrue(b1.finishedSplits().isEmpty(), "must not finish the split on wake-up");
+ assertEquals(0, readerFunction.getCloseCurrentSplitCount(), "split must stay open");
+
+ // Next fetch resumes and returns all records (the one-shot wake has fired).
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b2 = reader.fetch();
+ assertEquals(split.splitId(), b2.nextSplit());
+ assertEquals(3, drainRecordCount(b2));
+ }
+
+ @Test
+ public void testWakeUpCoincidingWithEofDefersFinishByOneFetch() throws IOException {
+ // readBatch returning null is ambiguous between true-EOF and woken-empty. When a wakeUp lands
+ // exactly at genuine exhaustion, fetch() returns an empty non-finishing batch once (deferring the
+ // finish), and the next fetch() finishes the split - it must never stall.
+ List<String> testData = Arrays.asList("r1", "r2");
+ TestSplitReaderFunction readerFunction = new TestSplitReaderFunction(testData);
+ HoodieSourceSplitReader<String> reader =
+ new HoodieSourceSplitReader<>(TABLE_NAME, readerContext, readerFunction, null, Option.empty());
+ boolean[] fired = {false};
+ readerFunction.setDrainProbe((buffered, hasNext) -> {
+ // Wake only at the start of a drain that finds the cursor already exhausted.
+ if (buffered == 0 && !hasNext && !fired[0]) {
+ fired[0] = true;
+ reader.wakeUp();
+ }
+ });
+
+ HoodieSourceSplit split = createTestSplit(1, "file1");
+ reader.handleSplitsChanges(new SplitsAddition<>(Collections.singletonList(split)));
+
+ // First fetch drains both records (cursor still had data at drain start, so no wake).
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b1 = reader.fetch();
+ assertEquals(2, drainRecordCount(b1));
+ assertEquals(0, readerFunction.getCloseCurrentSplitCount());
+
+ // Second fetch: cursor is exhausted at drain start -> wake fires -> empty non-finishing, deferred.
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b2 = reader.fetch();
+ assertTrue(b2.finishedSplits().isEmpty(), "finish deferred by the coinciding wake-up");
+ assertEquals(0, readerFunction.getCloseCurrentSplitCount());
+
+ // Third fetch: no wake now -> true EOF finishes and closes the split.
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b3 = reader.fetch();
+ assertTrue(b3.finishedSplits().contains(split.splitId()));
+ assertEquals(1, readerFunction.getCloseCurrentSplitCount());
+ }
+
+ @Test
+ public void testWakeUpWithLimitReachedStillFinishesSplit() throws IOException {
+ // Guard: the woken-resume short-circuit must live strictly on the readBatch-returned-null path.
+ // A wakeUp coinciding with the limit-reached path must still finish the split, never loop forever
+ // returning non-finishing batches. The limiter wakes the reader exactly when the limit is reached.
+ List<String> testData = Arrays.asList("r1", "r2", "r3");
+ TestSplitReaderFunction readerFunction = new TestSplitReaderFunction(testData);
+ AtomicReference<HoodieSourceSplitReader<String>> holder = new AtomicReference<>();
+ RecordLimiter wakingLimiter = new RecordLimiter(2L) {
+ @Override
+ public boolean isLimitReached() {
+ boolean reached = super.isLimitReached();
+ if (reached && holder.get() != null) {
+ holder.get().wakeUp();
+ }
+ return reached;
+ }
+ };
+ HoodieSourceSplitReader<String> reader =
+ new HoodieSourceSplitReader<>(TABLE_NAME, readerContext, readerFunction, null, Option.of(wakingLimiter));
+ holder.set(reader);
+
+ HoodieSourceSplit split = createTestSplit(1, "file1");
+ reader.handleSplitsChanges(new SplitsAddition<>(Collections.singletonList(split)));
+
+ // First fetch returns the split data; the limit wrapper caps consumption at 2 records.
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b1 = reader.fetch();
+ assertEquals(split.splitId(), b1.nextSplit());
+ assertEquals(2, drainRecordCount(b1), "limit caps drained records at 2");
+
+ // Second fetch hits the limit-reached path (which wakes the reader); it must finish, not defer.
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b2 = reader.fetch();
+ assertTrue(b2.finishedSplits().contains(split.splitId()), "limit-reached path must finish the split");
+ assertEquals(1, readerFunction.getCloseCurrentSplitCount());
+ }
+
+ @Test
+ public void testWakeUpPartialBatchRespectsLimit() throws IOException {
+ // A partial (woken) batch must not double-count against the pushed-down limit: with limit=3 over a
+ // 5-record split and a wake after the first record, the total emitted across batches stays at 3.
+ List<String> testData = Arrays.asList("r1", "r2", "r3", "r4", "r5");
+ TestSplitReaderFunction readerFunction = new TestSplitReaderFunction(testData);
+ HoodieSourceSplitReader<String> reader =
+ new HoodieSourceSplitReader<>(TABLE_NAME, readerContext, readerFunction, null, Option.of(new RecordLimiter(3L)));
+ boolean[] fired = {false};
+ readerFunction.setDrainProbe((buffered, hasNext) -> {
+ if (buffered == 1 && !fired[0]) {
+ fired[0] = true;
+ reader.wakeUp();
+ }
+ });
+
+ HoodieSourceSplit split = createTestSplit(1, "file1");
+ reader.handleSplitsChanges(new SplitsAddition<>(Collections.singletonList(split)));
+
+ int total = 0;
+ // Drain fetches until the split is finished; assert the limit caps the total at 3.
+ for (int i = 0; i < 10; i++) {
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> batch = reader.fetch();
+ if (batch.nextSplit() != null) {
+ total += drainRecordCount(batch);
+ }
+ if (batch.finishedSplits().contains(split.splitId())) {
+ break;
+ }
+ }
+ assertEquals(3, total, "pushed-down limit must cap the total across partial woken batches");
+ }
+
+ @Test
+ public void testCloseReleasesWokenStillOpenSplit() throws Exception {
+ // Unit proxy for the real shutdown path (SplitFetcher.run() -> splitReader.close() on the fetcher
+ // thread): a split left open by a wake-up is released when the reader is closed.
+ List<String> testData = Arrays.asList("r1", "r2", "r3");
+ TestSplitReaderFunction readerFunction = new TestSplitReaderFunction(testData);
+ HoodieSourceSplitReader<String> reader =
+ new HoodieSourceSplitReader<>(TABLE_NAME, readerContext, readerFunction, null, Option.empty());
+ boolean[] fired = {false};
+ readerFunction.setDrainProbe((buffered, hasNext) -> {
+ if (buffered == 1 && !fired[0]) {
+ fired[0] = true;
+ reader.wakeUp();
+ }
+ });
+
+ HoodieSourceSplit split = createTestSplit(1, "file1");
+ reader.handleSplitsChanges(new SplitsAddition<>(Collections.singletonList(split)));
+
+ // Woken fetch leaves the split open (not closed).
+ reader.fetch();
+ assertEquals(0, readerFunction.getCloseCurrentSplitCount());
+
+ // close() on the (split-fetcher) thread releases the still-open split.
+ reader.close();
+ assertEquals(1, readerFunction.getCloseCurrentSplitCount());
+ assertTrue(readerFunction.isClosed());
}
@Test
@@ -221,7 +448,7 @@ public class TestHoodieSourceSplitReader {
reader.fetch();
- assertEquals(1, readerFunction.getReadCount());
+ assertEquals(1, readerFunction.getOpenCount());
assertEquals(split, readerFunction.getLastReadSplit());
}
@@ -246,7 +473,7 @@ public class TestHoodieSourceSplitReader {
assertNotNull(result);
assertNull(result.nextSplit());
- assertEquals(0, readerFunction.getReadCount(), "Should not read any splits");
+ assertEquals(0, readerFunction.getOpenCount(), "Should not read any splits");
}
@Test
@@ -347,8 +574,8 @@ public class TestHoodieSourceSplitReader {
assertTrue(drainBatch.finishedSplits().contains(split2.splitId()),
"split2 should be drained as finished once limit is reached");
assertNull(drainBatch.nextSplit());
- // readerFunction.read() was called only once (for split1, never for split2)
- assertEquals(1, readerFunction.getReadCount());
+ // readerFunction was opened only once (for split1, never for split2)
+ assertEquals(1, readerFunction.getOpenCount());
}
@Test
@@ -368,8 +595,8 @@ public class TestHoodieSourceSplitReader {
assertTrue(finished.contains(split1.splitId()));
assertTrue(finished.contains(split2.splitId()));
assertNull(batch.nextSplit());
- // readerFunction.read() was never called
- assertEquals(0, readerFunction.getReadCount());
+ // readerFunction was never opened
+ assertEquals(0, readerFunction.getOpenCount());
}
@Test
@@ -449,6 +676,78 @@ public class TestHoodieSourceSplitReader {
assertEquals(5, drainRecordCount(batch));
}
+ // -------------------------------------------------------------------------
+ // Minibatch / resume tests
+ // -------------------------------------------------------------------------
+
+ @Test
+ public void testSplitSpanningMultipleMinibatches() throws IOException {
+ // A split larger than the mini-batch bound (2048) is emitted as multiple non-finished batches,
+ // then one finish signal. The reader function is opened once and closed once across the split,
+ // and the record offset stays continuous across the minibatch boundary.
+ int n = 2049;
+ List<String> testData = IntStream.range(0, n).mapToObj(i -> "r" + i).collect(Collectors.toList());
+ TestSplitReaderFunction readerFunction = new TestSplitReaderFunction(testData);
+ HoodieSourceSplitReader<String> reader =
+ new HoodieSourceSplitReader<>(TABLE_NAME, readerContext, readerFunction, null, Option.empty());
+
+ HoodieSourceSplit split = createTestSplit(1, "file1");
+ reader.handleSplitsChanges(new SplitsAddition<>(Collections.singletonList(split)));
+
+ // First minibatch: 2048 records, offsets 1..2048, split not yet finished.
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b1 = reader.fetch();
+ assertEquals(split.splitId(), b1.nextSplit());
+ long lastOffset = 0L;
+ int c1 = 0;
+ HoodieRecordWithPosition<String> rec;
+ while ((rec = b1.nextRecordFromSplit()) != null) {
+ lastOffset = rec.recordOffset();
+ c1++;
+ }
+ assertEquals(2048, c1);
+ assertEquals(2048L, lastOffset);
+ assertTrue(b1.finishedSplits().isEmpty(), "split not finished after the first minibatch");
+
+ // Second minibatch: the remaining record, offset continues at 2049.
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b2 = reader.fetch();
+ assertEquals(split.splitId(), b2.nextSplit());
+ HoodieRecordWithPosition<String> only = b2.nextRecordFromSplit();
+ assertNotNull(only);
+ assertEquals(2049L, only.recordOffset());
+ assertNull(b2.nextRecordFromSplit());
+ assertTrue(b2.finishedSplits().isEmpty());
+
+ // Third fetch: split exhausted -> finish signal.
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> b3 = reader.fetch();
+ assertTrue(b3.finishedSplits().contains(split.splitId()));
+
+ assertEquals(1, readerFunction.getOpenCount(), "split opened exactly once across minibatches");
+ assertEquals(1, readerFunction.getCloseCurrentSplitCount(), "split closed exactly once at EOF");
+ }
+
+ @Test
+ public void testResumeSkipsConsumedRecords() throws IOException {
+ // A recovered split carries a consumed offset; open() skips that many records and the emitted
+ // offsets resume at consumed+1.
+ List<String> testData = Arrays.asList("r1", "r2", "r3", "r4", "r5");
+ TestSplitReaderFunction readerFunction = new TestSplitReaderFunction(testData);
+ HoodieSourceSplitReader<String> reader =
+ new HoodieSourceSplitReader<>(TABLE_NAME, readerContext, readerFunction, null, Option.empty());
+
+ HoodieSourceSplit split = createTestSplit(1, "file1");
+ split.updatePosition(0, 2L); // 2 records already consumed before recovery
+ reader.handleSplitsChanges(new SplitsAddition<>(Collections.singletonList(split)));
+
+ RecordsWithSplitIds<HoodieRecordWithPosition<String>> batch = reader.fetch();
+ assertEquals(split.splitId(), batch.nextSplit());
+ HoodieRecordWithPosition<String> first = batch.nextRecordFromSplit();
+ assertNotNull(first);
+ assertEquals("r3", first.record(), "should resume past the 2 consumed records");
+ assertEquals(3L, first.recordOffset(), "offset resumes at consumed + 1");
+ // r4, r5 remain
+ assertEquals(2, drainRecordCount(batch));
+ }
+
/**
* Fetches the next batch that contains actual split data, skipping split-finish signal batches.
* Split-finish batches have non-empty {@code finishedSplits()} but no records.
@@ -495,14 +794,29 @@ public class TestHoodieSourceSplitReader {
}
/**
- * Test implementation of SplitReaderFunction.
+ * Test implementation of the stateful {@link SplitReaderFunction} cursor contract: {@code open}
+ * materializes the split's data and honors the consumed-offset skip, {@code readBatch} drains a
+ * bounded minibatch, and {@code closeCurrentSplit}/{@code close} release it.
*/
private static class TestSplitReaderFunction implements SplitReaderFunction<String> {
private final List<String> testData;
- private int readCount = 0;
+ private int openCount = 0;
+ private int closeCurrentSplitCount = 0;
private HoodieSourceSplit lastReadSplit = null;
private boolean closed = false;
+ // per-split cursor
+ private Iterator<String> cursor;
+ private long nextRecordOffset;
+
+ // Optional hook invoked as (bufferedCount, cursorHasNext) at readBatch start (count 0) and after
+ // each buffered record, so a test can call reader.wakeUp() at a precise point in the drain.
+ private BiConsumer<Integer, Boolean> drainProbe;
+
+ void setDrainProbe(BiConsumer<Integer, Boolean> drainProbe) {
+ this.drainProbe = drainProbe;
+ }
+
public TestSplitReaderFunction() {
this(Collections.emptyList());
}
@@ -512,25 +826,63 @@ public class TestHoodieSourceSplitReader {
}
@Override
- public RecordsWithSplitIds<HoodieRecordWithPosition<String>> read(HoodieSourceSplit split) {
- readCount++;
+ public void open(HoodieSourceSplit split) {
+ openCount++;
lastReadSplit = split;
- ClosableIterator<String> iterator = createClosableIterator(testData);
- return BatchRecords.forRecords(
- split.splitId(),
- iterator,
- split.getFileOffset(),
- split.getConsumed()
- );
+ cursor = testData.iterator();
+ long consumed = split.getConsumed();
+ for (long i = 0; i < consumed; i++) {
+ if (cursor.hasNext()) {
+ cursor.next();
+ } else {
+ throw new IllegalStateException(
+ "Invalid starting record offset " + consumed + " for split " + split.splitId());
+ }
+ }
+ nextRecordOffset = consumed;
}
@Override
- public void close() throws Exception {
+ public BatchRecords<String> readBatch(HoodieSourceSplit split, int batchSize, BooleanSupplier wakeupSignal) {
+ List<String> buffer = new ArrayList<>();
+ if (drainProbe != null) {
+ drainProbe.accept(0, cursor.hasNext());
+ }
+ while (buffer.size() < batchSize && !wakeupSignal.getAsBoolean() && cursor.hasNext()) {
+ buffer.add(cursor.next());
+ if (drainProbe != null) {
+ drainProbe.accept(buffer.size(), cursor.hasNext());
+ }
+ }
+ if (buffer.isEmpty()) {
+ return null;
+ }
+ long startingRecordOffset = nextRecordOffset;
+ nextRecordOffset += buffer.size();
+ return BatchRecords.forRecords(split.splitId(), buffer, split.getFileOffset(), startingRecordOffset);
+ }
+
+ @Override
+ public void closeCurrentSplit() {
+ closeCurrentSplitCount++;
+ cursor = null;
+ }
+
+ @Override
+ public void close() {
+ // Mirror AbstractSplitReaderFunction#close(): releasing the reader also releases any split
+ // still open (e.g. one left open by a wake-up), on the split-fetcher thread.
+ closeCurrentSplit();
closed = true;
}
- public int getReadCount() {
- return readCount;
+ // Number of splits opened; mirrors the old per-split read() count.
+ public int getOpenCount() {
+ return openCount;
+ }
+
+ public int getCloseCurrentSplitCount() {
+ return closeCurrentSplitCount;
}
public HoodieSourceSplit getLastReadSplit() {
@@ -540,25 +892,5 @@ public class TestHoodieSourceSplitReader {
public boolean isClosed() {
return closed;
}
-
- private ClosableIterator<String> createClosableIterator(List<String> items) {
- Iterator<String> iterator = items.iterator();
- return new ClosableIterator<String>() {
- @Override
- public void close() {
- // No-op
- }
-
- @Override
- public boolean hasNext() {
- return iterator.hasNext();
- }
-
- @Override
- public String next() {
- return iterator.next();
- }
- };
- }
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestAbstractSplitReaderFunction.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestAbstractSplitReaderFunction.java
index d785be90bb87..85cf06a665b5 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestAbstractSplitReaderFunction.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestAbstractSplitReaderFunction.java
@@ -18,20 +18,26 @@
package org.apache.hudi.source.reader.function;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.source.ExpressionPredicates;
+import org.apache.hudi.source.reader.BatchRecords;
import org.apache.hudi.source.reader.HoodieRecordWithPosition;
import org.apache.hudi.source.split.HoodieSourceSplit;
import org.apache.hudi.table.format.InternalSchemaManager;
import org.apache.hudi.utils.TestConfigurations;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.expressions.FieldReferenceExpression;
import org.apache.flink.table.expressions.ValueLiteralExpression;
import org.apache.flink.table.types.AtomicDataType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.types.RowKind;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -39,10 +45,13 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.util.Collections;
import java.util.List;
+import java.util.function.BooleanSupplier;
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.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
@@ -70,8 +79,8 @@ public class TestAbstractSplitReaderFunction {
/**
* Minimal concrete implementation that exposes the protected helper methods
- * ({@code getWriteConfig()} / {@code getHadoopConf()}) for testing, and leaves
- * {@code read()} / {@code close()} as no-ops.
+ * ({@code getWriteConfig()} / {@code getHadoopConf()}) for testing. The template methods return an
+ * empty iterator / a trivial row type; the constructor and singleton tests never drive the cursor.
*/
private static class MinimalSplitReaderFunction extends AbstractSplitReaderFunction {
@@ -84,12 +93,13 @@ public class TestAbstractSplitReaderFunction {
}
@Override
- public RecordsWithSplitIds<HoodieRecordWithPosition<RowData>> read(HoodieSourceSplit split) {
- return null;
+ protected ClosableIterator<RowData> createRecordIterator(HoodieSourceSplit split) {
+ return ClosableIterator.wrap(Collections.<RowData>emptyIterator());
}
@Override
- public void close() throws Exception {
+ protected RowType producedRowType() {
+ return RowType.of(new IntType());
}
HoodieWriteConfig writeConfigForTest() {
@@ -243,4 +253,135 @@ public class TestAbstractSplitReaderFunction {
assertSame(wc2, fn2.writeConfigForTest(),
"fn2's writeConfig must remain the same singleton across calls");
}
+
+ // -----------------------------------------------------------------------
+ // readBatch — copy-on-materialize (object-reuse regression guard)
+ // -----------------------------------------------------------------------
+
+ @Test
+ public void testReadBatchCopiesReusedRecordObjects() {
+ // Columnar readers return the SAME mutable RowData object on every next(); readBatch must copy
+ // each record, otherwise every entry in a materialized minibatch would alias the last row.
+ ReusedObjectSplitReaderFunction fn =
+ new ReusedObjectSplitReaderFunction(conf, mockInternalSchemaManager, 3);
+ HoodieSourceSplit split = createSplit();
+
+ fn.open(split);
+ BatchRecords<RowData> batch = fn.readBatch(split, 10, () -> false);
+ assertNotNull(batch);
+ batch.nextSplit();
+
+ RowData r0 = batch.nextRecordFromSplit().record();
+ RowData r1 = batch.nextRecordFromSplit().record();
+ RowData r2 = batch.nextRecordFromSplit().record();
+ assertNull(batch.nextRecordFromSplit());
+
+ // Distinct values despite the source reusing a single object.
+ assertEquals(0, r0.getInt(0));
+ assertEquals(1, r1.getInt(0));
+ assertEquals(2, r2.getInt(0));
+ assertNotSame(r0, r1, "records must be copies, not the reused source object");
+ assertNotSame(r1, r2);
+ // RowKind is preserved across the copy.
+ assertEquals(RowKind.DELETE, r0.getRowKind());
+ assertEquals(RowKind.INSERT, r1.getRowKind());
+ assertEquals(RowKind.DELETE, r2.getRowKind());
+ }
+
+ // -----------------------------------------------------------------------
+ // readBatch — wake-up signal (cooperative cancellation between records)
+ // -----------------------------------------------------------------------
+
+ @Test
+ public void testReadBatchStopsOnWakeupSignal() {
+ // The wakeupSignal is polled between records; once it trips, materialization stops early and the
+ // records buffered so far are returned as a partial minibatch with continuous offsets.
+ ReusedObjectSplitReaderFunction fn =
+ new ReusedObjectSplitReaderFunction(conf, mockInternalSchemaManager, 5);
+ HoodieSourceSplit split = createSplit();
+ fn.open(split);
+
+ // Returns false, false, true: the loop buffers 2 records, then the 3rd poll stops it.
+ int[] polls = {0};
+ BooleanSupplier signal = () -> (++polls[0]) > 2;
+
+ BatchRecords<RowData> batch = fn.readBatch(split, 10, signal);
+ assertNotNull(batch);
+ batch.nextSplit();
+
+ // nextRecordFromSplit() returns the same reused position wrapper each call, so read each record's
+ // value/offset before advancing. Offsets are 1-based and continuous from the starting offset (0).
+ HoodieRecordWithPosition<RowData> rec = batch.nextRecordFromSplit();
+ assertNotNull(rec);
+ assertEquals(0, rec.record().getInt(0));
+ assertEquals(1L, rec.recordOffset());
+
+ rec = batch.nextRecordFromSplit();
+ assertNotNull(rec);
+ assertEquals(1, rec.record().getInt(0));
+ assertEquals(2L, rec.recordOffset());
+
+ assertNull(batch.nextRecordFromSplit(), "materialization must stop at 2 records on wake-up");
+ }
+
+ @Test
+ public void testReadBatchReturnsNullWhenWokenBeforeAnyRecord() {
+ // A wake-up that lands before the first record is buffered yields an empty batch, signalled as
+ // null (the same sentinel as EOF); HoodieSourceSplitReader.fetch() disambiguates the two.
+ ReusedObjectSplitReaderFunction fn =
+ new ReusedObjectSplitReaderFunction(conf, mockInternalSchemaManager, 5);
+ HoodieSourceSplit split = createSplit();
+ fn.open(split);
+
+ assertNull(fn.readBatch(split, 10, () -> true));
+ }
+
+ private HoodieSourceSplit createSplit() {
+ return new HoodieSourceSplit(
+ 1, "base", Option.of(Collections.emptyList()), "/tbl", "/part",
+ "read_optimized", "19700101000000000", "file1", Option.empty());
+ }
+
+ /**
+ * Reader function whose iterator returns the SAME mutable {@link GenericRowData} instance on every
+ * {@code next()} (mimicking a columnar reader), used to prove readBatch copies each record.
+ */
+ private static class ReusedObjectSplitReaderFunction extends AbstractSplitReaderFunction {
+ private final int count;
+
+ ReusedObjectSplitReaderFunction(Configuration conf, InternalSchemaManager ism, int count) {
+ super(conf, Collections.emptyList(), ism, false);
+ this.count = count;
+ }
+
+ @Override
+ protected ClosableIterator<RowData> createRecordIterator(HoodieSourceSplit split) {
+ return new ClosableIterator<RowData>() {
+ private final GenericRowData reused = new GenericRowData(1);
+ private int i = 0;
+
+ @Override
+ public boolean hasNext() {
+ return i < count;
+ }
+
+ @Override
+ public RowData next() {
+ reused.setField(0, i);
+ reused.setRowKind(i % 2 == 0 ? RowKind.DELETE : RowKind.INSERT);
+ i++;
+ return reused; // same object every call
+ }
+
+ @Override
+ public void close() {
+ }
+ };
+ }
+
+ @Override
+ protected RowType producedRowType() {
+ return RowType.of(new IntType());
+ }
+ }
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestHoodieCdcSplitReaderFunction.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestHoodieCdcSplitReaderFunction.java
index b0d0579da084..d5dc3eca5770 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestHoodieCdcSplitReaderFunction.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestHoodieCdcSplitReaderFunction.java
@@ -156,11 +156,11 @@ public class TestHoodieCdcSplitReaderFunction {
1, "base.parquet", Option.empty(), tempDir.getAbsolutePath(),
"", "read_optimized", "20230101000000000", "file-1", Option.empty());
- Exception ex = assertThrows(Exception.class, () -> function.read(nonCdcSplit));
+ Exception ex = assertThrows(Exception.class, () -> function.open(nonCdcSplit));
assertNotNull(ex);
// Must not be IllegalArgumentException (which the old type-guard wrongly threw)
if (ex instanceof IllegalArgumentException) {
- throw new AssertionError("read() should not throw IllegalArgumentException for non-CDC split; "
+ throw new AssertionError("open() should not throw IllegalArgumentException for non-CDC split; "
+ "it should fall through to the fallback reader", ex);
}
}
@@ -223,7 +223,7 @@ public class TestHoodieCdcSplitReaderFunction {
// -------------------------------------------------------------------------
@Test
- public void testReadAcceptsCdcSourceSplitType() {
+ public void testReadAcceptsCdcSourceSplitType() throws Exception {
// Verify that HoodieCdcSourceSplit is accepted (cast doesn't throw).
// Actual I/O would require a real Hoodie table, so we only check the
// type-guard passes by catching the downstream I/O error rather than
@@ -237,7 +237,8 @@ public class TestHoodieCdcSplitReaderFunction {
1, tempDir.getAbsolutePath(), 128 * 1024 * 1024L, "file-cdc",
EMPTY_PARTITION_PATH, changes, "read_optimized", "20230101000000000");
- // Should not throw exception
- function.read(cdcSplit);
+ // Opening the split creates the CDC iterator lazily (no I/O yet); it must not throw.
+ function.open(cdcSplit);
+ function.close();
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestHoodieSplitReaderFunction.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestHoodieSplitReaderFunction.java
index 062b0c0ff11a..e6f8b76f986d 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestHoodieSplitReaderFunction.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/function/TestHoodieSplitReaderFunction.java
@@ -23,11 +23,17 @@ import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.internal.InternalSchema;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.read.HoodieFileGroupReader;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.source.ExpressionPredicates;
+import org.apache.hudi.source.split.HoodieSourceSplit;
import org.apache.hudi.table.format.InternalSchemaManager;
+import org.apache.hudi.util.StreamerUtil;
import org.apache.hudi.utils.TestConfigurations;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.expressions.FieldReferenceExpression;
import org.apache.flink.table.expressions.ValueLiteralExpression;
import org.apache.flink.table.types.AtomicDataType;
@@ -35,14 +41,23 @@ import org.apache.flink.table.types.logical.VarCharType;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockedStatic;
import java.io.File;
+import java.io.IOException;
import java.util.Collections;
import java.util.List;
+import static org.junit.jupiter.api.Assertions.assertEquals;
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.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
@@ -284,7 +299,8 @@ public class TestHoodieSplitReaderFunction {
@Test
public void testReadMethodSignature() {
- // Verify that the read method returns CloseableIterator
+ // Verify the reader function constructs via its public signature (the read path is driven
+ // through open/readBatch on the split-fetcher thread).
HoodieSplitReaderFunction function =
new HoodieSplitReaderFunction(
conf,
@@ -444,4 +460,78 @@ public class TestHoodieSplitReaderFunction {
assertNotNull(function);
}
+
+ // -------------------------------------------------------------------------
+ // createRecordIterator — file-group reader cleanup when iterator init fails
+ // -------------------------------------------------------------------------
+
+ @Test
+ public void testCreateRecordIteratorClosesReaderWhenInitFails() throws Exception {
+ // getClosableIterator() runs initRecordIterators(), which can open the reader's I/O resources and
+ // then throw. The reader is only a local in createRecordIterator, so the failure path must close
+ // it (nothing else would), preserving the original failure as the cause.
+ HoodieFileGroupReader<RowData> reader = mockReader();
+ IOException initFailure = new IOException("init failed");
+ when(reader.getClosableIterator()).thenThrow(initFailure);
+
+ HoodieSplitReaderFunction function = readerFunctionReturning(reader);
+ HoodieSourceSplit split = createSplit();
+
+ try (MockedStatic<StreamerUtil> mockedStreamerUtil = mockStatic(StreamerUtil.class)) {
+ mockedStreamerUtil.when(() -> StreamerUtil.metaClientForReader(any(), any()))
+ .thenReturn(mockMetaClient);
+
+ HoodieIOException thrown =
+ assertThrows(HoodieIOException.class, () -> function.createRecordIterator(split));
+ assertSame(initFailure, thrown.getCause(), "the original init failure must be the cause");
+ }
+ verify(reader, times(1)).close();
+ }
+
+ @Test
+ public void testCreateRecordIteratorSuppressesCloseError() throws Exception {
+ // A close() failure on the cleanup path must not mask the init failure: it is attached as a
+ // suppressed exception on the original.
+ HoodieFileGroupReader<RowData> reader = mockReader();
+ IOException initFailure = new IOException("init failed");
+ IOException closeFailure = new IOException("close failed");
+ when(reader.getClosableIterator()).thenThrow(initFailure);
+ doThrow(closeFailure).when(reader).close();
+
+ HoodieSplitReaderFunction function = readerFunctionReturning(reader);
+ HoodieSourceSplit split = createSplit();
+
+ try (MockedStatic<StreamerUtil> mockedStreamerUtil = mockStatic(StreamerUtil.class)) {
+ mockedStreamerUtil.when(() -> StreamerUtil.metaClientForReader(any(), any()))
+ .thenReturn(mockMetaClient);
+
+ HoodieIOException thrown =
+ assertThrows(HoodieIOException.class, () -> function.createRecordIterator(split));
+ assertSame(initFailure, thrown.getCause());
+ assertEquals(1, initFailure.getSuppressed().length, "close failure must be suppressed, not lost");
+ assertSame(closeFailure, initFailure.getSuppressed()[0]);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static HoodieFileGroupReader<RowData> mockReader() {
+ return mock(HoodieFileGroupReader.class);
+ }
+
+ private HoodieSplitReaderFunction readerFunctionReturning(HoodieFileGroupReader<RowData> reader) {
+ return new HoodieSplitReaderFunction(
+ conf, tableSchema, requiredSchema, mockInternalSchemaManager,
+ "AVRO_PAYLOAD", Collections.emptyList(), false) {
+ @Override
+ protected HoodieFileGroupReader<RowData> createFileGroupReader(HoodieSourceSplit split, HoodieTableMetaClient metaClient) {
+ return reader;
+ }
+ };
+ }
+
+ private static HoodieSourceSplit createSplit() {
+ return new HoodieSourceSplit(
+ 1, "base", Option.of(Collections.emptyList()), "/tbl", "/part",
+ "read_optimized", "19700101000000000", "file1", Option.empty());
+ }
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
index 07c3f0c05ecd..a656a1314eac 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
@@ -99,6 +99,7 @@ import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
@@ -4133,10 +4134,11 @@ public class ITTestHoodieDataSource {
* the collected rows.
*
* <p>The streaming job is terminated by a forced {@link CollectSinkTableFactory.SuccessException} once
- * {@code expectedNum} rows are collected. A benign teardown race (see {@link #isAcceptableTerminalFailure})
- * can instead end the job before all rows are emitted, leaving a short result. Re-reading the already
- * committed table is idempotent, so retry up to {@link #MAX_STREAM_READ_ATTEMPTS} times when the result
- * is short; this keeps the race from surfacing as a confusing row-count assertion failure.
+ * {@code expectedNum} rows are collected. On a slow CI shard the {@code await} window can elapse before
+ * the sink reaches {@code expectedNum} (a bare timeout - see {@link #isAwaitTimeout}), leaving a short
+ * result. Re-reading the already committed table is idempotent, so retry up to
+ * {@link #MAX_STREAM_READ_ATTEMPTS} times when the result is short; this keeps a slow shard from
+ * surfacing as a confusing row-count (or "Unexpected job failure") assertion failure.
*/
private List<Row> submitAndFetchWithRetry(TableEnvironment tEnv, String select, String sinkDDL, int expectedNum) {
List<Row> rows = Collections.emptyList();
@@ -4174,46 +4176,35 @@ public class ITTestHoodieDataSource {
private List<Row> fetchResultWithExpectedNum(TableEnvironment tEnv, TableResult tableResult) {
try {
// wait the continuous streaming query to be terminated by forced exception with expected row number
- // and max waiting timeout is 30s
- tableResult.await(30, TimeUnit.SECONDS);
+ // and max waiting timeout is 60s (kept generous so a slow CI shard does not time out before the
+ // sink collects its rows; a bare timeout is still handled as a retryable short read below)
+ tableResult.await(60, TimeUnit.SECONDS);
} catch (Throwable e) {
- // Acceptable terminal causes:
- // 1. SuccessException: the sink reached its expected row count and intentionally
- // threw to terminate the streaming job. This is the happy path.
- // 2. IOException("Stream is closed!") wrapped as HoodieIOException: a benign
- // error-attribution race between the source-side cascading-shutdown path and
- // the sink-side SuccessException terminator. When the sink throws
- // SuccessException to end the job, the chained source's SplitFetcher can close
- // the underlying Hadoop FSDataInputStream while the mailbox is still draining
- // a BatchRecords queued earlier; the next row-group read on the now-closed
- // stream surfaces an IOException("Stream is closed!"). With
- // restart-strategy.fixed-delay.attempts=0 (set in beforeEach to keep tests
- // deterministic) that IOException becomes the job's reported failure cause
- // instead of the sink's SuccessException, even though the sink has already
- // collected the expected rows by then - i.e. the functional outcome is
- // unchanged, only the error-attribution differs. Production paths correctly
- // fail the job on stream-closed-mid-read (the right behavior for real I/O
- // failures), so this tolerance is scoped to the SuccessException-based test
- // pattern below and is NOT mirrored in production code.
- // 3. NullPointerException from ParquetColumnarRowSplitReader#readNextRowGroup: the
- // same benign teardown race as (2), observed with different timing. When the
- // SplitFetcher's close() fully completes first, ParquetColumnarRowSplitReader#close
- // nulls out its `reader` field, so the in-flight row-group read on the task thread
- // surfaces as a NullPointerException (reader.readNextRowGroup() on a null reader)
- // instead of an IOException("Stream is closed!"). Same functional outcome - the
- // sink has already collected the expected rows - only the error symptom differs.
- // Tolerated narrowly (an NPE originating from that exact frame) for the same
- // reason as (2), and likewise NOT mirrored in production code.
- if (!isAcceptableTerminalFailure(e)) {
- throw new AssertionError("Unexpected job failure", e);
- }
- // The races (2)/(3) usually fire after the sink has collected its expected rows, but can also fire
- // before - ending the read with a short result. Log the tolerated cause so an incomplete read is
- // diagnosable; submitAndFetchWithRetry re-reads when the collected count is below the expectation.
- if (!isSuccessException(e)) {
- log.warn("Streaming read terminated by a tolerated teardown race ({}); collected {} rows so far.",
- describeTerminalCause(e),
+ // The only acceptable terminal cause is the sink reaching its expected row count and throwing
+ // SuccessException to terminate the streaming job (the happy path). The Source V2 read path now
+ // reads and closes each split's I/O on a single (split-fetcher) thread, so the former teardown
+ // races (a closed Parquet stream / a closed CDC iterator surfacing on the task thread) can no
+ // longer happen; any other terminal failure is a real error and fails the test.
+ //
+ // A bare await-window TimeoutException is not a terminal failure at all - the sink simply had not
+ // reached expectedNum yet (typically CI-load slowness), so the job is still running. Cancel it and
+ // let submitAndFetchWithRetry re-submit a fresh job, rather than treating a slow shard as a hard
+ // failure.
+ if (isAwaitTimeout(e)) {
+ // Cancel the still-running job (best-effort, bounded) so it cannot keep writing to the shared
+ // CollectSinkTableFactory.RESULT after the retry's re-submit clears it.
+ tableResult.getJobClient().ifPresent(jobClient -> {
+ try {
+ jobClient.cancel().get(30, TimeUnit.SECONDS);
+ } catch (Exception ignored) {
+ // best-effort cancel; the subsequent re-submit clears RESULT and starts a fresh job
+ }
+ });
+ log.warn("Streaming read did not reach the expected row count within the await window; "
+ + "cancelled the job and will retry. Collected {} rows so far.",
CollectSinkTableFactory.RESULT.values().stream().mapToInt(List::size).sum());
+ } else if (!isSuccessException(e)) {
+ throw new AssertionError("Unexpected job failure", e);
}
}
tEnv.executeSql("DROP TABLE IF EXISTS sink");
@@ -4223,56 +4214,20 @@ public class ITTestHoodieDataSource {
}
/**
- * Whether {@code e} (or any of its causes) is one of the terminal failures that
- * {@link #fetchResultWithExpectedNum} is allowed to swallow. See the comment at the call
- * site for the rationale.
- */
- private static boolean isAcceptableTerminalFailure(Throwable e) {
- Throwable cur = e;
- while (cur != null) {
- if (cur instanceof CollectSinkTableFactory.SuccessException) {
- return true;
- }
- String msg = cur.getMessage();
- if (msg != null && msg.contains("Stream is closed")) {
- return true;
- }
- // The NPE twin of the "Stream is closed!" teardown race (cause #3 at the call site):
- // a NullPointerException whose own stack trace originates from
- // ParquetColumnarRowSplitReader#readNextRowGroup, i.e. reader.readNextRowGroup() ran on a
- // null `reader` that ParquetColumnarRowSplitReader#close had just nulled out. Scoped to
- // that exact frame so genuine NPEs - and the legitimate IOException("expecting more
- // rows...") thrown from the same method - still fail the test.
- if (isNullPointerException(cur) && containsReadNextRowGroupFrame(cur)) {
- return true;
- }
- cur = cur.getCause();
- }
- return false;
- }
-
- /**
- * True for a real {@link NullPointerException} as well as one wrapped in Flink's
- * {@code SerializedThrowable} when the failure is propagated back from the cluster (its
- * {@code toString()} preserves the original {@code java.lang.NullPointerException} prefix).
+ * Whether {@code e} is a bare {@link TimeoutException} thrown directly by
+ * {@link org.apache.flink.table.api.TableResult#await(long, TimeUnit)} - i.e. the await window elapsed
+ * before the sink reached its expected row count and threw
+ * {@link CollectSinkTableFactory.SuccessException}, leaving the job still running (never terminated),
+ * so the caller cancels it and retries the read rather than swallowing it.
+ *
+ * <p>Only the top-level exception is inspected, never the cause chain: {@code await} throws its own
+ * timeout bare, whereas a genuine job failure arrives wrapped in an {@link ExecutionException} that
+ * may itself embed a {@link TimeoutException} (checkpoint expiry, RPC timeout). Walking the chain
+ * would misclassify such a real failure as a slow shard - cancel it, retry, and finally report a
+ * row-count mismatch with the true cause discarded.
*/
- private static boolean isNullPointerException(Throwable t) {
- return t instanceof NullPointerException
- || t.toString().startsWith(NullPointerException.class.getName());
- }
-
- /**
- * Whether {@code t}'s stack trace (preserved even through {@code SerializedThrowable})
- * contains a {@code ParquetColumnarRowSplitReader#readNextRowGroup} frame.
- */
- private static boolean containsReadNextRowGroupFrame(Throwable t) {
- for (StackTraceElement frame : t.getStackTrace()) {
- if (frame.getClassName().endsWith("ParquetColumnarRowSplitReader")
- && "readNextRowGroup".equals(frame.getMethodName())) {
- return true;
- }
- }
- return false;
+ private static boolean isAwaitTimeout(Throwable e) {
+ return e instanceof TimeoutException;
}
/**
@@ -4287,15 +4242,4 @@ public class ITTestHoodieDataSource {
}
return false;
}
-
- /**
- * Short description of {@code e}'s root cause, for logging which tolerated terminal failure fired.
- */
- private static String describeTerminalCause(Throwable e) {
- Throwable root = e;
- while (root.getCause() != null) {
- root = root.getCause();
- }
- return root.getClass().getSimpleName() + ": " + root.getMessage();
- }
}