Skip to content

[flink] Honor snapshot limit in exactly-once monitor source #9677

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jianguotian wants to merge 1 commit into
base: master
Choose a base branch
from
+188 −2
Open

[flink] Honor snapshot limit in exactly-once monitor source #9677

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 66 additions & 2 deletions ...mon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -136,15 +151,27 @@ private class Reader extends AbstractNonCoordinatedSourceReader<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
public void notifyCheckpointComplete(long checkpointId) {
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
Expand Down Expand Up @@ -190,6 +217,8 @@ public void addSplits(List<SimpleSourceSplit> list) {
for (Tuple2<Long, Long> tuple2 : nextSnapshotState.get()) {
nextSnapshotPerCheckpoint.put(tuple2.f0, tuple2.f1);
}
inFlightNextSnapshots.clear();
availableFuture.complete(null);
}

@Override
Expand All @@ -199,11 +228,20 @@ public CompletableFuture<Void> isAvailable() {

@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();
Expand Down Expand Up @@ -231,8 +269,18 @@ public InputStatus pollNext(ReaderOutput<Split> 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<RowData> buildSource(
Expand Down Expand Up @@ -307,8 +355,24 @@ public static DataStream<RowData> 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<Split, SimpleSourceSplit, NoOpEnumState> source = monitorSource;
if (table != null) {
source = new PaimonDataStreamSource<>(monitorSource, table);
Expand Down
122 changes: 122 additions & 0 deletions ...link-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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 {
Expand All @@ -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");
Expand Down Expand Up @@ -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<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 =
Expand Down
Loading