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 <[email protected]>
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 <[email protected]>
---
.../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();
- }
}
