This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new bda726b82f [flink] Fix source split reader release reliability (#8183)
bda726b82f is described below
commit bda726b82f8ceba4238dc01ffebbc8dec35c8339
Author: QuakeWang <[email protected]>
AuthorDate: Sat Jun 20 18:42:27 2026 +0800
[flink] Fix source split reader release reliability (#8183)
The source split readers keep a single pooled iterator and may also hold
`currentFirstBatch` after seeking past restored records.
The old design had two reliability gaps:
- `close()` only closed the lazy record reader, so an already fetched
`currentFirstBatch` could be left unreleased.
- `releaseBatch()` returned the pooled iterator only after the
underlying batch release succeeded. If `iterator.releaseBatch()` threw,
the single-entry pool lost its iterator and later fetches could block
forever.
This PR releases `currentFirstBatch` during close, always returns the
pooled iterator in a `finally` block, and aligns the CDC reader's
no-split fetch path with the common reader by failing clearly instead of
returning without a valid split.
---
.../cdc/source/reader/CDCSourceSplitReader.java | 25 ++-
.../source/reader/CDCSourceSplitReaderTest.java | 195 +++++++++++++++++++++
.../flink/source/FileStoreSourceSplitReader.java | 23 ++-
.../source/FileStoreSourceSplitReaderTest.java | 174 ++++++++++++++++++
4 files changed, 406 insertions(+), 11 deletions(-)
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReader.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReader.java
index 9aeb48567d..59d3f5799c 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReader.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReader.java
@@ -185,9 +185,19 @@ public class CDCSourceSplitReader
@Override
public void close() throws Exception {
currentSchemaChangeEvents.clear();
- if (currentReader != null) {
- if (currentReader.lazyRecordReader != null) {
- currentReader.lazyRecordReader.close();
+ try {
+ if (currentFirstBatch != null) {
+ try {
+ currentFirstBatch.releaseBatch();
+ } finally {
+ currentFirstBatch = null;
+ }
+ }
+ } finally {
+ if (currentReader != null) {
+ if (currentReader.lazyRecordReader != null) {
+ currentReader.lazyRecordReader.close();
+ }
}
}
}
@@ -199,7 +209,7 @@ public class CDCSourceSplitReader
final TableAwareFileStoreSourceSplit nextSplit = splits.poll();
if (nextSplit == null) {
- return;
+ throw new IOException("Cannot fetch from another split - no split
remaining");
}
// update metric when split changes
@@ -320,8 +330,11 @@ public class CDCSourceSplitReader
@Override
public void releaseBatch() {
- this.iterator.releaseBatch();
- pool.recycler().recycle(this);
+ try {
+ this.iterator.releaseBatch();
+ } finally {
+ pool.recycler().recycle(this);
+ }
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReaderTest.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReaderTest.java
index 82199811b5..8461ed2a66 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReaderTest.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReaderTest.java
@@ -26,12 +26,14 @@ import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.flink.pipeline.cdc.source.CDCSource;
import
org.apache.paimon.flink.pipeline.cdc.source.TableAwareFileStoreSourceSplit;
import
org.apache.paimon.flink.source.FileStoreSourceReaderTest.DummyMetricGroup;
import org.apache.paimon.flink.source.TestChangelogDataReadWrite;
import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.options.Options;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
@@ -69,6 +71,12 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.OptionalLong;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -76,6 +84,7 @@ import static
org.apache.paimon.flink.LogicalTypeConversion.toDataType;
import static
org.apache.paimon.flink.source.FileStoreSourceSplitSerializerTest.newSourceSplit;
import static org.apache.paimon.io.DataFileTestUtils.row;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link CDCSourceSplitReader}. */
public class CDCSourceSplitReaderTest {
@@ -444,6 +453,68 @@ public class CDCSourceSplitReaderTest {
reader.close();
}
+ @Test
+ public void testNoSplit() throws Exception {
+ TestChangelogDataReadWrite rw = new
TestChangelogDataReadWrite(tablePath);
+ CDCSourceSplitReader reader = createReader(rw.createReadWithKey());
+ assertThatThrownBy(reader::fetch).hasMessageContaining("no split
remaining");
+ reader.close();
+ }
+
+ @Test
+ public void testRecycleIteratorWhenReleaseBatchFails() throws Exception {
+ CDCSourceSplitReader reader =
+ createReader(
+ new TestingTableRead(
+ new SingleBatchRecordReader(new
FailingReleaseIterator())));
+ try {
+ assignSplit(
+ reader,
+ new TableAwareFileStoreSourceSplit(
+ "id1",
+ new TestingSplit(),
+ 0,
+ Identifier.create(DATABASE, TABLE),
+ 1L,
+ 1L));
+
+ RecordsWithSplitIds<RecordIterator<Event>> records =
reader.fetch();
+ assertThatThrownBy(records::recycle).hasMessageContaining("release
failed");
+
+ RecordsWithSplitIds<RecordIterator<Event>> finishedRecords =
+ fetchWithoutWaitingForPool(reader);
+
assertThat(finishedRecords.finishedSplits()).isEqualTo(Collections.singleton("id1"));
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ public void testCloseReleasesCurrentFirstBatch() throws Exception {
+ TrackingRecordIterator iterator = new TrackingRecordIterator();
+ CDCSourceSplitReader reader =
+ createReader(new TestingTableRead(new
SingleBatchRecordReader(iterator)));
+ try {
+ assignSplit(
+ reader,
+ new TableAwareFileStoreSourceSplit(
+ "id1",
+ new TestingSplit(),
+ 1,
+ Identifier.create(DATABASE, TABLE),
+ 1L,
+ 1L));
+ reader.wakeUp();
+
+ RecordsWithSplitIds<RecordIterator<Event>> records =
reader.fetch();
+ assertThat(records.finishedSplits()).isEmpty();
+ assertThat(records.nextSplit()).isNull();
+ } finally {
+ reader.close();
+ }
+ assertThat(iterator.released()).isTrue();
+ }
+
@Test
public void testPauseOrResumeSplits() throws Exception {
TestChangelogDataReadWrite rw = new
TestChangelogDataReadWrite(tablePath);
@@ -622,6 +693,22 @@ public class CDCSourceSplitReaderTest {
reader.handleSplitsChanges(splitsChange);
}
+ private RecordsWithSplitIds<RecordIterator<Event>>
fetchWithoutWaitingForPool(
+ CDCSourceSplitReader reader) throws Exception {
+ ExecutorService executorService = Executors.newSingleThreadExecutor();
+ Future<RecordsWithSplitIds<RecordIterator<Event>>> future =
+ executorService.submit(reader::fetch);
+ try {
+ return future.get(5, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ reader.wakeUp();
+ future.get(5, TimeUnit.SECONDS);
+ throw new AssertionError("Timed out waiting for split reader
fetch.", e);
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
public static TableAwareFileStoreSourceSplit newSourceSplit(
String id, BinaryRow partition, int bucket, List<DataFileMeta>
files) {
return newSourceSplit(id, partition, bucket, files, false, 0);
@@ -732,4 +819,112 @@ public class CDCSourceSplitReaderTest {
return schemaChangeEvents;
}
}
+
+ private static class TestingTableRead implements TableRead {
+
+ private final RecordReader<InternalRow> recordReader;
+
+ private TestingTableRead(RecordReader<InternalRow> recordReader) {
+ this.recordReader = recordReader;
+ }
+
+ @Override
+ public TableRead withMetricRegistry(MetricRegistry registry) {
+ return this;
+ }
+
+ @Override
+ public TableRead executeFilter() {
+ return this;
+ }
+
+ @Override
+ public TableRead withIOManager(IOManager ioManager) {
+ return this;
+ }
+
+ @Override
+ public RecordReader<InternalRow> createReader(Split split) {
+ return recordReader;
+ }
+ }
+
+ private static class SingleBatchRecordReader implements
RecordReader<InternalRow> {
+
+ private final RecordReader.RecordIterator<InternalRow> iterator;
+ private boolean returned;
+
+ private
SingleBatchRecordReader(RecordReader.RecordIterator<InternalRow> iterator) {
+ this.iterator = iterator;
+ }
+
+ @Nullable
+ @Override
+ public RecordReader.RecordIterator<InternalRow> readBatch() {
+ if (returned) {
+ return null;
+ }
+
+ returned = true;
+ return iterator;
+ }
+
+ @Override
+ public void close() {}
+ }
+
+ private static class FailingReleaseIterator
+ implements RecordReader.RecordIterator<InternalRow> {
+
+ @Nullable
+ @Override
+ public InternalRow next() {
+ return null;
+ }
+
+ @Override
+ public void releaseBatch() {
+ throw new RuntimeException("release failed");
+ }
+ }
+
+ private static class TrackingRecordIterator
+ implements RecordReader.RecordIterator<InternalRow> {
+
+ private boolean returned;
+ private boolean released;
+
+ @Nullable
+ @Override
+ public InternalRow next() {
+ if (returned) {
+ return null;
+ }
+
+ returned = true;
+ return GenericRow.of(1L);
+ }
+
+ @Override
+ public void releaseBatch() {
+ released = true;
+ }
+
+ private boolean released() {
+ return released;
+ }
+ }
+
+ private static class TestingSplit implements Split {
+
+ @Override
+ public long rowCount() {
+ return 0;
+ }
+
+ @Override
+ public OptionalLong mergedRowCount() {
+ return OptionalLong.empty();
+ }
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java
index b49b9adb94..992867ab69 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java
@@ -191,9 +191,19 @@ public class FileStoreSourceSplitReader
@Override
public void close() throws Exception {
- if (currentReader != null) {
- if (currentReader.lazyRecordReader != null) {
- currentReader.lazyRecordReader.close();
+ try {
+ if (currentFirstBatch != null) {
+ try {
+ currentFirstBatch.releaseBatch();
+ } finally {
+ currentFirstBatch = null;
+ }
+ }
+ } finally {
+ if (currentReader != null) {
+ if (currentReader.lazyRecordReader != null) {
+ currentReader.lazyRecordReader.close();
+ }
}
}
}
@@ -319,8 +329,11 @@ public class FileStoreSourceSplitReader
@Override
public void releaseBatch() {
- this.iterator.releaseBatch();
- pool.recycler().recycle(this);
+ try {
+ this.iterator.releaseBatch();
+ } finally {
+ pool.recycler().recycle(this);
+ }
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceSplitReaderTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceSplitReaderTest.java
index d08bc94a40..135cb38dab 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceSplitReaderTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceSplitReaderTest.java
@@ -20,13 +20,18 @@ package org.apache.paimon.flink.source;
import org.apache.paimon.KeyValue;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
import
org.apache.paimon.flink.source.FileStoreSourceReaderTest.DummyMetricGroup;
import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.metrics.MetricRegistry;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.utils.RecordWriter;
@@ -52,6 +57,12 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.OptionalLong;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -341,6 +352,45 @@ public class FileStoreSourceSplitReaderTest {
reader.close();
}
+ @Test
+ public void testRecycleIteratorWhenReleaseBatchFails() throws Exception {
+ FileStoreSourceSplitReader reader =
+ createReader(
+ new TestingTableRead(
+ new SingleBatchRecordReader(new
FailingReleaseIterator())),
+ null);
+ try {
+ assignSplit(reader, new FileStoreSourceSplit("id1", new
TestingSplit()));
+
+ RecordsWithSplitIds<RecordIterator<RowData>> records =
reader.fetch();
+ assertThatThrownBy(records::recycle).hasMessageContaining("release
failed");
+
+ RecordsWithSplitIds<RecordIterator<RowData>> finishedRecords =
+ fetchWithoutWaitingForPool(reader);
+
assertThat(finishedRecords.finishedSplits()).isEqualTo(Collections.singleton("id1"));
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ public void testCloseReleasesCurrentFirstBatch() throws Exception {
+ TrackingRecordIterator iterator = new TrackingRecordIterator();
+ FileStoreSourceSplitReader reader =
+ createReader(new TestingTableRead(new
SingleBatchRecordReader(iterator)), null);
+ try {
+ assignSplit(reader, new FileStoreSourceSplit("id1", new
TestingSplit(), 1));
+ reader.wakeUp();
+
+ RecordsWithSplitIds<RecordIterator<RowData>> records =
reader.fetch();
+ assertThat(records.finishedSplits()).isEmpty();
+ assertThat(records.nextSplit()).isNull();
+ } finally {
+ reader.close();
+ }
+ assertThat(iterator.released()).isTrue();
+ }
+
@Test
public void testLimit() throws Exception {
TestChangelogDataReadWrite rw = new
TestChangelogDataReadWrite(tempDir.toString());
@@ -494,4 +544,128 @@ public class FileStoreSourceSplitReaderTest {
new SplitsAddition<>(Collections.singletonList(split));
reader.handleSplitsChanges(splitsChange);
}
+
+ private RecordsWithSplitIds<RecordIterator<RowData>>
fetchWithoutWaitingForPool(
+ FileStoreSourceSplitReader reader) throws Exception {
+ ExecutorService executorService = Executors.newSingleThreadExecutor();
+ Future<RecordsWithSplitIds<RecordIterator<RowData>>> future =
+ executorService.submit(reader::fetch);
+ try {
+ return future.get(5, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ reader.wakeUp();
+ future.get(5, TimeUnit.SECONDS);
+ throw new AssertionError("Timed out waiting for split reader
fetch.", e);
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
+ private static class TestingTableRead implements TableRead {
+
+ private final RecordReader<InternalRow> recordReader;
+
+ private TestingTableRead(RecordReader<InternalRow> recordReader) {
+ this.recordReader = recordReader;
+ }
+
+ @Override
+ public TableRead withMetricRegistry(MetricRegistry registry) {
+ return this;
+ }
+
+ @Override
+ public TableRead executeFilter() {
+ return this;
+ }
+
+ @Override
+ public TableRead withIOManager(IOManager ioManager) {
+ return this;
+ }
+
+ @Override
+ public RecordReader<InternalRow> createReader(Split split) {
+ return recordReader;
+ }
+ }
+
+ private static class SingleBatchRecordReader implements
RecordReader<InternalRow> {
+
+ private final RecordReader.RecordIterator<InternalRow> iterator;
+ private boolean returned;
+
+ private
SingleBatchRecordReader(RecordReader.RecordIterator<InternalRow> iterator) {
+ this.iterator = iterator;
+ }
+
+ @Nullable
+ @Override
+ public RecordReader.RecordIterator<InternalRow> readBatch() {
+ if (returned) {
+ return null;
+ }
+
+ returned = true;
+ return iterator;
+ }
+
+ @Override
+ public void close() {}
+ }
+
+ private static class FailingReleaseIterator
+ implements RecordReader.RecordIterator<InternalRow> {
+
+ @Nullable
+ @Override
+ public InternalRow next() {
+ return null;
+ }
+
+ @Override
+ public void releaseBatch() {
+ throw new RuntimeException("release failed");
+ }
+ }
+
+ private static class TrackingRecordIterator
+ implements RecordReader.RecordIterator<InternalRow> {
+
+ private boolean returned;
+ private boolean released;
+
+ @Nullable
+ @Override
+ public InternalRow next() {
+ if (returned) {
+ return null;
+ }
+
+ returned = true;
+ return GenericRow.of(1L);
+ }
+
+ @Override
+ public void releaseBatch() {
+ released = true;
+ }
+
+ private boolean released() {
+ return released;
+ }
+ }
+
+ private static class TestingSplit implements Split {
+
+ @Override
+ public long rowCount() {
+ return 0;
+ }
+
+ @Override
+ public OptionalLong mergedRowCount() {
+ return OptionalLong.empty();
+ }
+ }
}
