From 06ef4b7454eff4294958f173656f6191d6504805 Mon Sep 17 00:00:00 2001 From: mingfeng Date: Mon, 7 Sep 2026 07:06:13 -0700 Subject: [PATCH] [flink] Honor snapshot limit in exactly-once monitor source --- .../flink/source/operator/MonitorSource.java | 68 +++++++++- .../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 3fbec3697b95..740cb59311d5 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.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 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 { 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 @@ private class Reader extends AbstractNonCoordinatedSourceReader { Long.parseLong(x.split(":")[0]), Long.parseLong(x.split(":")[1]))); private final TreeMap nextSnapshotPerCheckpoint = new TreeMap<>(); + private final Deque inFlightNextSnapshots = new ArrayDeque<>(); private CompletableFuture availableFuture = CompletableFuture.completedFuture(null); @Override @@ -143,8 +159,19 @@ public void notifyCheckpointComplete(long checkpointId) { NavigableMap 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 void addSplits(List list) { for (Tuple2 tuple2 : nextSnapshotState.get()) { nextSnapshotPerCheckpoint.put(tuple2.f0, tuple2.f1); } + inFlightNextSnapshots.clear(); + availableFuture.complete(null); } @Override @@ -199,11 +228,20 @@ public CompletableFuture isAvailable() { @Override public InputStatus pollNext(ReaderOutput readerOutput) throws Exception { + if (snapshotLimitReached()) { + return InputStatus.NOTHING_AVAILABLE; + } + boolean isEmpty; try { List 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 InputStatus pollNext(ReaderOutput readerOutput) throws Exception { }); 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 buildSource( @@ -307,8 +355,24 @@ public static DataStream buildSource( @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 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 9c57f27b866d..4446278be1d0 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.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.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.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 void before() .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 void testMonitorSource() throws Exception { } } + @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 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 reader = + (SourceReader) + sourceTransformation.getSource().createReader(null); + TestingReaderOutput 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 reader = source.createReader(null); + TestingReaderOutput 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 reader = source.createReader(null); + TestingReaderOutput 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 reader = source.createReader(null); + + writeToTable(1, 1, 1); + assertThat(reader.pollNext(new TestingReaderOutput<>())) + .isEqualTo(InputStatus.NOTHING_AVAILABLE); + List checkpoint = reader.snapshotState(1L); + + MonitorSource restoredSource = + new MonitorSource(table.newReadBuilder(), 10, false, false, 1); + SourceReader restoredReader = restoredSource.createReader(null); + restoredReader.addSplits(checkpoint); + assertThat(restoredReader.isAvailable()).isDone(); + + writeToTable(2, 2, 2); + TestingReaderOutput 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 =