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 2463bd103b [flink] Honor snapshot limit in exactly-once monitor source
(#9677)
2463bd103b is described below
commit 2463bd103be56295bc53f68fbd3e1f92bb80dbe4
Author: jianguotian <[email protected]>
AuthorDate: Tue Sep 8 16:40:17 2026 +0800
[flink] Honor snapshot limit in exactly-once monitor source (#9677)
---
.../flink/source/operator/MonitorSource.java | 68 +++++++++++-
.../flink/source/operator/OperatorSourceTest.java | 122 +++++++++++++++++++++
2 files changed, 188 insertions(+), 2 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
index 3fbec3697b..740cb59311 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.source.operator;
+import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.flink.NestedProjectedRowData;
import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
@@ -26,6 +27,7 @@ import org.apache.paimon.flink.source.PaimonDataStreamSource;
import org.apache.paimon.flink.source.SimpleSourceSplit;
import org.apache.paimon.flink.source.SplitListState;
import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.options.Options;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.ChannelComputer;
import org.apache.paimon.table.source.DataSplit;
@@ -56,7 +58,9 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.util.ArrayDeque;
import java.util.ArrayList;
+import java.util.Deque;
import java.util.List;
import java.util.NavigableMap;
import java.util.OptionalLong;
@@ -96,16 +100,27 @@ public class MonitorSource extends
AbstractNonCoordinatedSource<Split> {
private final long monitorInterval;
private final boolean emitSnapshotWatermark;
private final boolean isBounded;
+ private final int maxSnapshotCount;
public MonitorSource(
ReadBuilder readBuilder,
long monitorInterval,
boolean emitSnapshotWatermark,
boolean isBounded) {
+ this(readBuilder, monitorInterval, emitSnapshotWatermark, isBounded,
-1);
+ }
+
+ MonitorSource(
+ ReadBuilder readBuilder,
+ long monitorInterval,
+ boolean emitSnapshotWatermark,
+ boolean isBounded,
+ int maxSnapshotCount) {
this.readBuilder = readBuilder;
this.monitorInterval = monitorInterval;
this.emitSnapshotWatermark = emitSnapshotWatermark;
this.isBounded = isBounded;
+ this.maxSnapshotCount = maxSnapshotCount;
}
@Override
@@ -136,6 +151,7 @@ public class MonitorSource extends
AbstractNonCoordinatedSource<Split> {
Long.parseLong(x.split(":")[0]),
Long.parseLong(x.split(":")[1])));
private final TreeMap<Long, Long> nextSnapshotPerCheckpoint = new
TreeMap<>();
+ private final Deque<Long> inFlightNextSnapshots = new ArrayDeque<>();
private CompletableFuture<Void> availableFuture =
CompletableFuture.completedFuture(null);
@Override
@@ -143,8 +159,19 @@ public class MonitorSource extends
AbstractNonCoordinatedSource<Split> {
NavigableMap<Long, Long> nextSnapshots =
nextSnapshotPerCheckpoint.headMap(checkpointId, true);
OptionalLong max =
nextSnapshots.values().stream().mapToLong(Long::longValue).max();
- max.ifPresent(scan::notifyCheckpointComplete);
+ boolean limitReached = snapshotLimitReached();
+ max.ifPresent(
+ completedNextSnapshot -> {
+ scan.notifyCheckpointComplete(completedNextSnapshot);
+ while (!inFlightNextSnapshots.isEmpty()
+ && inFlightNextSnapshots.getFirst() <=
completedNextSnapshot) {
+ inFlightNextSnapshots.removeFirst();
+ }
+ });
nextSnapshots.clear();
+ if (limitReached && !snapshotLimitReached()) {
+ availableFuture.complete(null);
+ }
}
@Override
@@ -190,6 +217,8 @@ public class MonitorSource extends
AbstractNonCoordinatedSource<Split> {
for (Tuple2<Long, Long> tuple2 : nextSnapshotState.get()) {
nextSnapshotPerCheckpoint.put(tuple2.f0, tuple2.f1);
}
+ inFlightNextSnapshots.clear();
+ availableFuture.complete(null);
}
@Override
@@ -199,11 +228,20 @@ public class MonitorSource extends
AbstractNonCoordinatedSource<Split> {
@Override
public InputStatus pollNext(ReaderOutput<Split> readerOutput) throws
Exception {
+ if (snapshotLimitReached()) {
+ return InputStatus.NOTHING_AVAILABLE;
+ }
+
boolean isEmpty;
try {
List<Split> splits = isBounded ? batchScan.plan().splits() :
scan.plan().splits();
isEmpty = splits.isEmpty();
splits.forEach(readerOutput::collect);
+ if (!isBounded && maxSnapshotCount > 0 && !isEmpty) {
+ inFlightNextSnapshots.addLast(
+ Preconditions.checkNotNull(
+ scan.checkpoint(), "Non-empty streaming
plan without state."));
+ }
if (emitSnapshotWatermark && !isBounded) {
Long watermark = scan.watermark();
@@ -231,8 +269,18 @@ public class MonitorSource extends
AbstractNonCoordinatedSource<Split> {
});
return InputStatus.NOTHING_AVAILABLE;
}
+ if (snapshotLimitReached()) {
+ availableFuture = new CompletableFuture<>();
+ return InputStatus.NOTHING_AVAILABLE;
+ }
return InputStatus.MORE_AVAILABLE;
}
+
+ private boolean snapshotLimitReached() {
+ return !isBounded
+ && maxSnapshotCount > 0
+ && inFlightNextSnapshots.size() >= maxSnapshotCount;
+ }
}
public static DataStream<RowData> buildSource(
@@ -307,8 +355,24 @@ public class MonitorSource extends
AbstractNonCoordinatedSource<Split> {
@Nullable Table table,
RowType readType,
boolean blobAsDescriptor) {
+ int maxSnapshotCount =
+ table == null
+ ? -1
+ : Options.fromMap(table.options())
+
.get(FlinkConnectorOptions.SCAN_MAX_SNAPSHOT_COUNT);
+ Preconditions.checkArgument(
+ isBounded
+ || maxSnapshotCount <= 0
+ || env.getCheckpointConfig().isCheckpointingEnabled(),
+ "Option '%s' is only supported for streaming monitor source
when checkpointing is enabled.",
+ FlinkConnectorOptions.SCAN_MAX_SNAPSHOT_COUNT.key());
MonitorSource monitorSource =
- new MonitorSource(readBuilder, monitorInterval,
emitSnapshotWatermark, isBounded);
+ new MonitorSource(
+ readBuilder,
+ monitorInterval,
+ emitSnapshotWatermark,
+ isBounded,
+ maxSnapshotCount);
Source<Split, SimpleSourceSplit, NoOpEnumState> source = monitorSource;
if (table != null) {
source = new PaimonDataStreamSource<>(monitorSource, table);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
index 9c57f27b86..4446278be1 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
@@ -23,6 +23,8 @@ import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.flink.source.FlinkSourceBuilder;
+import org.apache.paimon.flink.source.SimpleSourceSplit;
import org.apache.paimon.flink.utils.TestingMetricUtils;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.Table;
@@ -34,11 +36,16 @@ import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.DataTypes;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
+import org.apache.flink.core.io.InputStatus;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.event.WatermarkEvent;
+import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.operators.SourceOperator;
+import org.apache.flink.streaming.api.transformations.SourceTransformation;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
@@ -68,8 +75,11 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import static org.apache.paimon.CoreOptions.CONSUMER_EXPIRATION_TIME;
import static org.apache.paimon.CoreOptions.CONSUMER_ID;
+import static
org.apache.paimon.flink.FlinkConnectorOptions.SCAN_MAX_SNAPSHOT_COUNT;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link MonitorSource} and {@link ReadOperator}. */
public class OperatorSourceTest {
@@ -92,6 +102,8 @@ public class OperatorSourceTest {
.column("c", DataTypes.INT())
.primaryKey("a")
.option(CONSUMER_ID.key(), "my_consumer")
+ .option(CONSUMER_EXPIRATION_TIME.key(), "1 d")
+ .option(SCAN_MAX_SNAPSHOT_COUNT.key(), "1")
.option("bucket", "1")
.build();
Identifier identifier = Identifier.create("default", "t");
@@ -195,6 +207,116 @@ public class OperatorSourceTest {
}
}
+ @Test
+ public void testMonitorSourceSnapshotLimitRequiresCheckpointing() {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+
+ assertThatThrownBy(
+ () -> new
FlinkSourceBuilder(table).env(env).sourceBounded(false).build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(SCAN_MAX_SNAPSHOT_COUNT.key())
+ .hasMessageContaining("checkpoint");
+ }
+
+ @Test
+ public void testMonitorSourceLimitsSnapshotsUntilCheckpointCompletes()
throws Exception {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.enableCheckpointing(10);
+ DataStream<RowData> dataStream =
+ new
FlinkSourceBuilder(table).env(env).sourceBounded(false).build();
+ SourceTransformation<?, ?, ?> sourceTransformation =
+
dataStream.getTransformation().getTransitivePredecessors().stream()
+ .filter(SourceTransformation.class::isInstance)
+ .map(SourceTransformation.class::cast)
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ @SuppressWarnings("unchecked")
+ SourceReader<Split, SimpleSourceSplit> reader =
+ (SourceReader<Split, SimpleSourceSplit>)
+ sourceTransformation.getSource().createReader(null);
+ TestingReaderOutput<Split> output = new TestingReaderOutput<>();
+
+ writeToTable(1, 1, 1);
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ assertThat(output.getEmittedRecords()).hasSize(1);
+
+ writeToTable(2, 2, 2);
+ assertThat(reader.isAvailable()).isNotDone();
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ assertThat(output.getEmittedRecords()).hasSize(1);
+
+ reader.snapshotState(1L);
+ reader.notifyCheckpointComplete(1L);
+
+ assertThat(reader.isAvailable()).isDone();
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ assertThat(output.getEmittedRecords()).hasSize(2);
+ }
+
+ @Test
+ public void testCheckpointBeforeSnapshotDoesNotReleaseSnapshotLimit()
throws Exception {
+ MonitorSource source = new MonitorSource(table.newReadBuilder(), 10,
false, false, 1);
+ SourceReader<Split, SimpleSourceSplit> reader =
source.createReader(null);
+ TestingReaderOutput<Split> output = new TestingReaderOutput<>();
+
+ reader.snapshotState(1L);
+ writeToTable(1, 1, 1);
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+
+ reader.notifyCheckpointComplete(1L);
+ assertThat(reader.isAvailable()).isNotDone();
+
+ reader.snapshotState(2L);
+ reader.notifyCheckpointComplete(2L);
+ assertThat(reader.isAvailable()).isDone();
+ }
+
+ @Test
+ public void testCompletedCheckpointReleasesCoveredSnapshotCredits() throws
Exception {
+ MonitorSource source = new MonitorSource(table.newReadBuilder(), 10,
false, false, 2);
+ SourceReader<Split, SimpleSourceSplit> reader =
source.createReader(null);
+ TestingReaderOutput<Split> output = new TestingReaderOutput<>();
+
+ writeToTable(1, 1, 1);
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE);
+ reader.snapshotState(1L);
+
+ writeToTable(2, 2, 2);
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ assertThat(output.getEmittedRecords()).hasSize(2);
+
+ reader.notifyCheckpointComplete(1L);
+ assertThat(reader.isAvailable()).isDone();
+
+ writeToTable(3, 3, 3);
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ assertThat(output.getEmittedRecords()).hasSize(3);
+ }
+
+ @Test
+ public void testMonitorSourceSnapshotLimitIsOpenAfterRestore() throws
Exception {
+ MonitorSource source = new MonitorSource(table.newReadBuilder(), 10,
false, false, 1);
+ SourceReader<Split, SimpleSourceSplit> reader =
source.createReader(null);
+
+ writeToTable(1, 1, 1);
+ assertThat(reader.pollNext(new TestingReaderOutput<>()))
+ .isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ List<SimpleSourceSplit> checkpoint = reader.snapshotState(1L);
+
+ MonitorSource restoredSource =
+ new MonitorSource(table.newReadBuilder(), 10, false, false, 1);
+ SourceReader<Split, SimpleSourceSplit> restoredReader =
restoredSource.createReader(null);
+ restoredReader.addSplits(checkpoint);
+ assertThat(restoredReader.isAvailable()).isDone();
+
+ writeToTable(2, 2, 2);
+ TestingReaderOutput<Split> output = new TestingReaderOutput<>();
+
assertThat(restoredReader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ assertThat(output.getEmittedRecords()).hasSize(1);
+ assertThat(readSplit(output.getEmittedRecords().get(0)))
+ .containsExactlyInAnyOrder(Arrays.asList(2, 2, 2));
+ }
+
@Test
public void testReadOperator() throws Exception {
ReadOperator readOperator =
