cshuo commented on code in PR #19202:
URL: https://github.com/apache/hudi/pull/19202#discussion_r3592183075
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/AbstractSplitReaderFunction.java:
##########
@@ -52,6 +77,85 @@ public AbstractSplitReaderFunction(
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) {
+ ValidationUtils.checkState(currentIterator != null,
+ "readBatch called before open for split " + split.splitId());
+ RowDataSerializer serializer = getCopySerializer();
+ List<RowData> buffer = new ArrayList<>();
+ try {
+ while (buffer.size() < batchSize && currentIterator.hasNext()) {
+ RowData next = currentIterator.next();
+ RowData copy = serializer.copy(next);
+ copy.setRowKind(next.getRowKind());
Review Comment:
`RowDataSerializer.copy(next)` already preserves `RowKind` (including the
generic-row path, which explicitly copies it). Therefore,
`copy.setRowKind(next.getRowKind())` is redundant.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/BatchRecords.java:
##########
@@ -97,29 +104,14 @@ public Set<String> finishedSplits() {
@Override
public void recycle() {
Review Comment:
`BatchRecords#recycle()` is currently a no-op. Could you create a follow-up
issue to explore reusing consumed batches (and their row objects) through an
object pool? It could reduce allocation pressure on the source side when
processing many small batches.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java:
##########
@@ -101,6 +101,11 @@ private 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
Review Comment:
Now the CdcFileSplitsIterator in source V2 is also read and closed in one
single thread, the split-fetcher thread, so there should be no concurrent
operations on the iterator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
