From fb8325b5ddbd8a0cb6b019f5e42695bc5cdd88d8 Mon Sep 17 00:00:00 2001 From: HuangXiao Date: Wed, 19 Aug 2026 14:59:44 +0800 Subject: [PATCH 1/2] [flink] Mark lake and KV snapshot as finished in split state at EOF before processing log records --- .../flink/lake/LakeRecordRecordEmitter.java | 5 + .../LakeSnapshotAndFlussLogSplitState.java | 5 + .../source/emitter/FlinkRecordEmitter.java | 5 + .../reader/FlinkRecordsWithSplitIds.java | 3 + .../source/reader/FlinkSourceSplitReader.java | 35 ++- .../flink/source/reader/RecordAndPos.java | 26 +- .../split/HybridSnapshotLogSplitState.java | 5 + .../source/reader/FlinkSourceReaderTest.java | 285 +++++++++++++++++- .../reader/FlinkSourceSplitReaderTest.java | 27 +- .../flink/source/reader/RecordAndPosTest.java | 21 ++ 10 files changed, 399 insertions(+), 18 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeRecordRecordEmitter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeRecordRecordEmitter.java index ddd492fec6d..557c91e173c 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeRecordRecordEmitter.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeRecordRecordEmitter.java @@ -47,6 +47,11 @@ public void emitRecord( LakeSnapshotAndFlussLogSplitState lakeSnapshotAndFlussLogSplitState = (LakeSnapshotAndFlussLogSplitState) splitState; + if (recordAndPos.isSnapshotPhaseFinished()) { + lakeSnapshotAndFlussLogSplitState.markLakeSplitFinished(); + return; + } + // set current split index lakeSnapshotAndFlussLogSplitState.setCurrentLakeSplitIndex( recordAndPos.getCurrentSplitIndex()); diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/state/LakeSnapshotAndFlussLogSplitState.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/state/LakeSnapshotAndFlussLogSplitState.java index 4d36ad0eaa2..6a265e51c94 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/state/LakeSnapshotAndFlussLogSplitState.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/state/LakeSnapshotAndFlussLogSplitState.java @@ -49,6 +49,11 @@ public void setCurrentLakeSplitIndex(int currentLakeSplitIndex) { this.currentLakeSplitIndex = currentLakeSplitIndex; } + /** Marks the lake split as finished. */ + public void markLakeSplitFinished() { + isLakeSplitFinished = true; + } + public void setNextLogOffset(long nextOffset) { // if set offset, means lake splits is finished isLakeSplitFinished = true; diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitter.java index b659554949f..edf8a94f50e 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitter.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitter.java @@ -73,6 +73,11 @@ public void emitRecord( HybridSnapshotLogSplitState hybridSnapshotLogSplitState = splitState.asHybridSnapshotLogSplitState(); + if (recordAndPosition.isSnapshotPhaseFinished()) { + hybridSnapshotLogSplitState.markSnapshotFinished(); + return; + } + ScanRecord scanRecord = recordAndPosition.record(); if (scanRecord.logOffset() >= 0) { // record is with a valid offset, means it's in incremental phase, diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkRecordsWithSplitIds.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkRecordsWithSplitIds.java index 3b7babe1a6b..35c8f17ff19 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkRecordsWithSplitIds.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkRecordsWithSplitIds.java @@ -119,6 +119,9 @@ public RecordAndPos nextRecordFromSplit() { + "iterate over the records split."); if (currentRecordIterator.hasNext()) { RecordAndPos recordAndPos = currentRecordIterator.next(); + if (recordAndPos.isSnapshotPhaseFinished()) { + return recordAndPos; + } long offset = recordAndPos.record().logOffset(); // the record current offset is not less than the stopping offset, // shouldn't emit it diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java index 50e2143962c..271e267cd4e 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java @@ -547,19 +547,28 @@ private long getStoppingOffset(TableBucket tableBucket) { } private FlinkRecordsWithSplitIds finishCurrentBoundedSplit() throws IOException { - Set finishedSplits = - (currentBoundedSplit instanceof HybridSnapshotLogSplit - && !((HybridSnapshotLogSplit) currentBoundedSplit) - .isBatch()) - || (currentBoundedSplit instanceof LakeSnapshotAndFlussLogSplit - && ((LakeSnapshotAndFlussLogSplit) currentBoundedSplit) - .isStreaming()) - // is hybrid split, or is lakeAndFlussLog split in streaming mode, - // not to finish this split - // since it remains log to read - ? Collections.emptySet() - : Collections.singleton(currentBoundedSplit.splitId()); - final FlinkRecordsWithSplitIds finishRecords = new FlinkRecordsWithSplitIds(finishedSplits); + boolean isStreamingHybridSplit = + currentBoundedSplit instanceof HybridSnapshotLogSplit + && !((HybridSnapshotLogSplit) currentBoundedSplit).isBatch(); + boolean isStreamingLakeSplit = + currentBoundedSplit instanceof LakeSnapshotAndFlussLogSplit + && ((LakeSnapshotAndFlussLogSplit) currentBoundedSplit).isStreaming(); + final FlinkRecordsWithSplitIds finishRecords; + if (isStreamingHybridSplit || isStreamingLakeSplit) { + // is hybrid split, or is lakeAndFlussLog split in streaming mode, send an internal + // record + // to set snapshot phase finished in split state + finishRecords = + forBoundedSplitRecords( + currentBoundedSplit, + CloseableIterator.wrap( + Collections.singleton(RecordAndPos.snapshotPhaseFinished()) + .iterator())); + } else { + finishRecords = + new FlinkRecordsWithSplitIds( + Collections.singleton(currentBoundedSplit.splitId())); + } closeCurrentBoundedSplit(); return finishRecords; } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/RecordAndPos.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/RecordAndPos.java index 1cdf93dcee4..16c677789a0 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/RecordAndPos.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/RecordAndPos.java @@ -54,6 +54,8 @@ public class RecordAndPos { // the index for the current split that the record is from protected int currentSplitIndex; + private final boolean isSnapshotPhaseFinished; + public RecordAndPos(ScanRecord scanRecord) { this(scanRecord, NO_READ_RECORDS_COUNT, DEFAULT_SPLIT_INDEX); } @@ -63,9 +65,23 @@ public RecordAndPos(ScanRecord scanRecord, long readRecordsCount) { } public RecordAndPos(ScanRecord scanRecord, long readRecordsCount, int currentSplitIndex) { + this(scanRecord, readRecordsCount, currentSplitIndex, false); + } + + private RecordAndPos( + ScanRecord scanRecord, + long readRecordsCount, + int currentSplitIndex, + boolean isSnapshotPhaseFinished) { this.scanRecord = scanRecord; this.readRecordsCount = readRecordsCount; this.currentSplitIndex = currentSplitIndex; + this.isSnapshotPhaseFinished = isSnapshotPhaseFinished; + } + + /** Creates a marker indicating that the snapshot phase has been fully read. */ + public static RecordAndPos snapshotPhaseFinished() { + return new RecordAndPos(null, NO_READ_RECORDS_COUNT, DEFAULT_SPLIT_INDEX, true); } public long readRecordsCount() { @@ -80,6 +96,11 @@ public ScanRecord record() { return scanRecord; } + /** Returns whether this is a marker indicating that the snapshot phase has been fully read. */ + public boolean isSnapshotPhaseFinished() { + return isSnapshotPhaseFinished; + } + @Override public boolean equals(Object o) { if (this == o) { @@ -91,12 +112,14 @@ public boolean equals(Object o) { RecordAndPos that = (RecordAndPos) o; return readRecordsCount == that.readRecordsCount && currentSplitIndex == that.currentSplitIndex + && isSnapshotPhaseFinished == that.isSnapshotPhaseFinished && Objects.equals(scanRecord, that.scanRecord); } @Override public int hashCode() { - return Objects.hash(scanRecord, readRecordsCount, currentSplitIndex); + return Objects.hash( + scanRecord, readRecordsCount, currentSplitIndex, isSnapshotPhaseFinished); } @Override @@ -108,6 +131,7 @@ public String toString() { + readRecordsCount + ", currentSplitIndex=" + currentSplitIndex + + (isSnapshotPhaseFinished ? ", isSnapshotPhaseFinished=true" : "") + '}'; } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplitState.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplitState.java index a28d45a042d..515be9090f3 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplitState.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplitState.java @@ -53,6 +53,11 @@ public void setRecordsToSkip(long recordsToSkip) { this.recordsToSkip = recordsToSkip; } + /** Marks the snapshot phase as finished. */ + public void markSnapshotFinished() { + snapshotFinished = true; + } + public void setNextOffset(long nextOffset) { // if set offset, means snapshot is finished snapshotFinished = true; diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java index 556a81e69d7..91b685deacd 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java @@ -17,22 +17,38 @@ package org.apache.fluss.flink.source.reader; +import org.apache.fluss.client.metadata.KvSnapshots; +import org.apache.fluss.client.table.Table; +import org.apache.fluss.client.table.writer.UpsertResult; +import org.apache.fluss.client.table.writer.UpsertWriter; import org.apache.fluss.config.Configuration; +import org.apache.fluss.flink.lake.split.LakeSnapshotAndFlussLogSplit; import org.apache.fluss.flink.source.deserializer.DeserializerInitContextImpl; import org.apache.fluss.flink.source.deserializer.RowDataDeserializationSchema; import org.apache.fluss.flink.source.emitter.FlinkRecordEmitter; +import org.apache.fluss.flink.source.event.FinishedKvSnapshotConsumeEvent; import org.apache.fluss.flink.source.event.PartitionBucketsUnsubscribedEvent; import org.apache.fluss.flink.source.event.PartitionsRemovedEvent; import org.apache.fluss.flink.source.metrics.FlinkSourceReaderMetrics; +import org.apache.fluss.flink.source.split.HybridSnapshotLogSplit; import org.apache.fluss.flink.source.split.LogSplit; +import org.apache.fluss.flink.source.split.SourceSplitBase; import org.apache.fluss.flink.utils.FlinkTestBase; import org.apache.fluss.lake.source.LakeSource; import org.apache.fluss.lake.source.LakeSplit; +import org.apache.fluss.lake.source.RecordReader; +import org.apache.fluss.lake.source.TestingLakeSource; +import org.apache.fluss.lake.source.TestingLakeSplit; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.ChangeType; +import org.apache.fluss.record.GenericRecord; +import org.apache.fluss.record.LogRecord; +import org.apache.fluss.row.InternalRow; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.types.RowType; +import org.apache.fluss.utils.CloseableIterator; import org.apache.flink.api.connector.source.SourceEvent; import org.apache.flink.api.connector.source.SourceReaderContext; @@ -43,6 +59,7 @@ import org.apache.flink.table.data.RowData; import org.junit.jupiter.api.Test; +import java.io.IOException; import java.time.Duration; import java.util.ArrayList; import java.util.Collections; @@ -52,14 +69,248 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; +import static org.apache.fluss.testutils.DataTestUtils.row; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.assertj.core.api.Assertions.assertThat; /** Test for {@link FlinkSourceReader}. */ class FlinkSourceReaderTest extends FlinkTestBase { + @Test + void testCheckpointHybridSnapshotFinishedBeforeFirstLogRecord() throws Exception { + TablePath tablePath = TablePath.of(DEFAULT_DB, "test_checkpoint_hybrid_snapshot_finished"); + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema(DEFAULT_PK_TABLE_SCHEMA) + .distributedBy(1, "id") + .build(); + long tableId = createTable(tablePath, tableDescriptor); + + UpsertResult seedResult = upsert(tablePath, row(1, "snapshot")); + TableBucket tableBucket = new TableBucket(tableId, 0); + assertThat(seedResult.getBucket()).isEqualTo(tableBucket); + FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(tablePath); + + KvSnapshots snapshots = admin.getLatestKvSnapshots(tablePath).get(); + long snapshotId = snapshots.getSnapshotId(0).getAsLong(); + long logStartingOffset = snapshots.getLogOffset(0).getAsLong(); + assertThat(logStartingOffset).isEqualTo(seedResult.getLogEndOffset()); + HybridSnapshotLogSplit split = + new HybridSnapshotLogSplit( + tableBucket, + null, + snapshotId, + 0, + false, + logStartingOffset, + LogSplit.NO_STOPPING_OFFSET, + false); + + TestingReaderContext readerContext = new TestingReaderContext(); + TestingReaderOutput snapshotOutput = new TestingReaderOutput<>(); + HybridSnapshotLogSplit checkpointSplit; + try (FlinkSourceReader reader = + createReader( + clientConf, + tablePath, + tableDescriptor.getSchema().getRowType(), + readerContext, + null)) { + reader.addSplits(Collections.singletonList(split)); + + // Poll until the snapshot EOF marker reaches the split state. There is no incremental + // record yet, so this checkpoint must preserve the snapshot-to-log boundary itself. + retry( + Duration.ofMinutes(1), + () -> { + reader.pollNext(snapshotOutput); + List checkpoint = reader.snapshotState(1L); + assertThat(checkpoint).hasSize(1); + assertThat(checkpoint.get(0)).isInstanceOf(HybridSnapshotLogSplit.class); + assertThat( + ((HybridSnapshotLogSplit) checkpoint.get(0)) + .isSnapshotFinished()) + .isTrue(); + }); + + List checkpoint = reader.snapshotState(2L); + assertThat(checkpoint).hasSize(1); + checkpointSplit = (HybridSnapshotLogSplit) checkpoint.get(0); + } + + assertThat(checkpointSplit.isSnapshotFinished()).isTrue(); + assertThat(checkpointSplit.getSnapshotId()).isEqualTo(snapshotId); + assertThat(checkpointSplit.getLogStartingOffset()).isEqualTo(logStartingOffset); + assertThat(checkpointSplit.recordsToSkip()).isEqualTo(1L); + assertThat(snapshotOutput.getEmittedRecords()).hasSize(1); + assertThat(snapshotOutput.getEmittedRecords().get(0).getInt(0)).isEqualTo(1); + assertThat(snapshotOutput.getEmittedRecords().get(0).getString(1).toString()) + .isEqualTo("snapshot"); + assertThat(readerContext.getSentEvents()) + .containsExactly( + new FinishedKvSnapshotConsumeEvent(1L, Collections.singleton(tableBucket))); + + // Restore with an invalid snapshot ID to model a snapshot released after the checkpoint. + // Since the checkpoint marks the snapshot complete, recovery must subscribe only to log. + UpsertResult newLogResult = upsert(tablePath, row(2, "new-log")); + assertThat(newLogResult.getBucket()).isEqualTo(tableBucket); + HybridSnapshotLogSplit restoringSplit = + new HybridSnapshotLogSplit( + tableBucket, + null, + Long.MAX_VALUE, + checkpointSplit.recordsToSkip(), + checkpointSplit.isSnapshotFinished(), + checkpointSplit.getLogStartingOffset(), + checkpointSplit.getLogStoppingOffset().orElse(LogSplit.NO_STOPPING_OFFSET), + checkpointSplit.isBatch()); + TestingReaderOutput restoredOutput = new TestingReaderOutput<>(); + try (FlinkSourceReader restoredReader = + createReader( + clientConf, + tablePath, + tableDescriptor.getSchema().getRowType(), + new TestingReaderContext(), + null)) { + restoredReader.addSplits(Collections.singletonList(restoringSplit)); + + retry( + Duration.ofMinutes(1), + () -> { + restoredReader.pollNext(restoredOutput); + assertThat(restoredOutput.getEmittedRecords()).hasSize(1); + }); + + assertThat(restoredOutput.getEmittedRecords()).hasSize(1); + assertThat(restoredOutput.getEmittedRecords().get(0).getInt(0)).isEqualTo(2); + assertThat(restoredOutput.getEmittedRecords().get(0).getString(1).toString()) + .isEqualTo("new-log"); + + List restoredCheckpoint = restoredReader.snapshotState(2L); + assertThat(restoredCheckpoint).hasSize(1); + HybridSnapshotLogSplit restoredCheckpointSplit = + (HybridSnapshotLogSplit) restoredCheckpoint.get(0); + assertThat(restoredCheckpointSplit.isSnapshotFinished()).isTrue(); + assertThat(restoredCheckpointSplit.getSnapshotId()).isEqualTo(Long.MAX_VALUE); + assertThat(restoredCheckpointSplit.recordsToSkip()).isEqualTo(1L); + assertThat(restoredCheckpointSplit.getLogStartingOffset()) + .isEqualTo(newLogResult.getLogEndOffset()); + } + } + + @Test + void testCheckpointLakeSplitFinishedBeforeFirstLogRecord() throws Exception { + TablePath tablePath = TablePath.of(DEFAULT_DB, "test_checkpoint_lake_split_finished"); + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema(DEFAULT_PK_TABLE_SCHEMA) + .distributedBy(1, "id") + .build(); + long tableId = createTable(tablePath, tableDescriptor); + + UpsertResult seedResult = upsert(tablePath, row(1, "lake")); + TableBucket tableBucket = new TableBucket(tableId, 0); + assertThat(seedResult.getBucket()).isEqualTo(tableBucket); + long startingOffset = seedResult.getLogEndOffset(); + + // Mirror the seed row in the lake snapshot. Since the log starts at its LEO, this row must + // be emitted only from the lake. + TrackingLakeSource lakeSource = + new TrackingLakeSource( + Collections.singletonList( + new GenericRecord(-1L, -1L, ChangeType.INSERT, row(1, "lake")))); + LakeSnapshotAndFlussLogSplit split = + new LakeSnapshotAndFlussLogSplit( + tableBucket, + null, + Collections.singletonList(new TestingLakeSplit(0, Collections.emptyList())), + startingOffset, + LogSplit.NO_STOPPING_OFFSET); + + TestingReaderOutput lakeOutput = new TestingReaderOutput<>(); + LakeSnapshotAndFlussLogSplit checkpointSplit; + try (FlinkSourceReader reader = + createReader( + clientConf, + tablePath, + tableDescriptor.getSchema().getRowType(), + new TestingReaderContext(), + lakeSource)) { + reader.addSplits(Collections.singletonList(split)); + + // Poll until the lake EOF marker reaches the split state. No incremental record exists + // yet, so the checkpoint captures the lake-to-log boundary. + retry( + Duration.ofMinutes(1), + () -> { + reader.pollNext(lakeOutput); + List checkpoint = reader.snapshotState(1L); + assertThat(checkpoint).hasSize(1); + assertThat(checkpoint.get(0)) + .isInstanceOf(LakeSnapshotAndFlussLogSplit.class); + assertThat( + ((LakeSnapshotAndFlussLogSplit) checkpoint.get(0)) + .isLakeSplitFinished()) + .isTrue(); + }); + + List checkpoint = reader.snapshotState(2L); + assertThat(checkpoint).hasSize(1); + checkpointSplit = (LakeSnapshotAndFlussLogSplit) checkpoint.get(0); + assertThat(checkpointSplit.isLakeSplitFinished()).isTrue(); + assertThat(checkpointSplit.getStartingOffset()).isEqualTo(startingOffset); + assertThat(checkpointSplit.getRecordsToSkip()).isEqualTo(1L); + assertThat(checkpointSplit.getCurrentLakeSplitIndex()).isZero(); + assertThat(lakeOutput.getEmittedRecords()).hasSize(1); + assertThat(lakeOutput.getEmittedRecords().get(0).getInt(0)).isEqualTo(1); + assertThat(lakeOutput.getEmittedRecords().get(0).getString(1).toString()) + .isEqualTo("lake"); + } + assertThat(lakeSource.getCreateLakeRecordReaderCount()).isEqualTo(1); + + // Append the first incremental record after the boundary checkpoint. Recovery must read it + // through LogScanner without reopening the finished lake RecordReader. + UpsertResult newLogResult = upsert(tablePath, row(2, "new-log")); + assertThat(newLogResult.getBucket()).isEqualTo(tableBucket); + TrackingLakeSource restoringLakeSource = new TrackingLakeSource(Collections.emptyList()); + TestingReaderOutput restoredOutput = new TestingReaderOutput<>(); + try (FlinkSourceReader restoredReader = + createReader( + clientConf, + tablePath, + tableDescriptor.getSchema().getRowType(), + new TestingReaderContext(), + restoringLakeSource)) { + restoredReader.addSplits(Collections.singletonList(checkpointSplit)); + + retry( + Duration.ofMinutes(1), + () -> { + restoredReader.pollNext(restoredOutput); + assertThat(restoredOutput.getEmittedRecords()).hasSize(1); + }); + + assertThat(restoringLakeSource.getCreateLakeRecordReaderCount()).isZero(); + assertThat(restoredOutput.getEmittedRecords().get(0).getInt(0)).isEqualTo(2); + assertThat(restoredOutput.getEmittedRecords().get(0).getString(1).toString()) + .isEqualTo("new-log"); + + List restoredCheckpoint = restoredReader.snapshotState(3L); + assertThat(restoredCheckpoint).hasSize(1); + LakeSnapshotAndFlussLogSplit restoredCheckpointSplit = + (LakeSnapshotAndFlussLogSplit) restoredCheckpoint.get(0); + assertThat(restoredCheckpointSplit.isLakeSplitFinished()).isTrue(); + assertThat(restoredCheckpointSplit.getStartingOffset()) + .isEqualTo(newLogResult.getLogEndOffset()); + assertThat(restoredCheckpointSplit.getRecordsToSkip()).isEqualTo(1L); + assertThat(restoredCheckpointSplit.getCurrentLakeSplitIndex()).isZero(); + } + } + @Test void testHandlePartitionsRemovedEvent() throws Exception { TablePath tablePath = TablePath.of(DEFAULT_DB, "test_partitioned_table"); @@ -82,7 +333,7 @@ void testHandlePartitionsRemovedEvent() throws Exception { // try to write some rows to the table TestingReaderContext readerContext = new TestingReaderContext(); - try (final FlinkSourceReader reader = + try (final FlinkSourceReader reader = createReader( clientConf, tablePath, @@ -158,7 +409,16 @@ void testHandlePartitionsRemovedEvent() throws Exception { } } - private FlinkSourceReader createReader( + private UpsertResult upsert(TablePath tablePath, InternalRow row) throws Exception { + try (Table table = conn.getTable(tablePath)) { + UpsertWriter writer = table.newUpsert().createWriter(); + CompletableFuture result = writer.upsert(row); + writer.flush(); + return result.get(); + } + } + + private FlinkSourceReader createReader( Configuration flussConf, TablePath tablePath, RowType sourceOutputType, @@ -188,4 +448,25 @@ private FlinkSourceReader createReader( recordEmitter, lakeSource); } + + private static final class TrackingLakeSource extends TestingLakeSource { + + private final List records; + private final AtomicInteger createLakeRecordReaderCount = new AtomicInteger(); + + private TrackingLakeSource(List records) { + this.records = records; + } + + @Override + public RecordReader createRecordReader(ReaderContext context) + throws IOException { + createLakeRecordReaderCount.incrementAndGet(); + return () -> CloseableIterator.wrap(records.iterator()); + } + + private int getCreateLakeRecordReaderCount() { + return createLakeRecordReaderCount.get(); + } + } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java index 5b0117a0f73..bef07bab3d0 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java @@ -371,6 +371,15 @@ private void assignSplitsAndFetchUntilRetrieveRecords( assignSplits(reader, splits); Map> splitConsumedRecords = new HashMap<>(); + Set expectedSnapshotPhaseFinishedSplits = new HashSet<>(); + for (SourceSplitBase split : splits) { + if (split.isHybridSnapshotLogSplit() + && !split.asHybridSnapshotLogSplit().isBatch() + && !split.asHybridSnapshotLogSplit().isSnapshotFinished()) { + expectedSnapshotPhaseFinishedSplits.add(split.splitId()); + } + } + Set snapshotPhaseFinishedSplits = new HashSet<>(); Set finishedSplits = new HashSet<>(); while (finishedSplits.size() < splits.size()) { @@ -381,7 +390,16 @@ private void assignSplitsAndFetchUntilRetrieveRecords( List splitFetch = new ArrayList<>(); RecordAndPos record; while ((record = recordsBySplitIds.nextRecordFromSplit()) != null) { - splitFetch.add(new RecordAndPos(record.record(), record.readRecordsCount())); + if (record.isSnapshotPhaseFinished()) { + assertThat(record.record()).isNull(); + assertThat(expectedSnapshotPhaseFinishedSplits).contains(splitId); + assertThat(snapshotPhaseFinishedSplits.add(splitId)) + .as("only one snapshot phase finished marker per split") + .isTrue(); + } else { + splitFetch.add( + new RecordAndPos(record.record(), record.readRecordsCount())); + } } splitConsumedRecords @@ -391,7 +409,9 @@ private void assignSplitsAndFetchUntilRetrieveRecords( // if records retrieved from this split is greater or equal to expected records, // it means we should stop read if (splitConsumedRecords.getOrDefault(splitId, Collections.emptyList()).size() - >= expectedRecords.get(splitId).size()) { + >= expectedRecords.get(splitId).size() + && (!expectedSnapshotPhaseFinishedSplits.contains(splitId) + || snapshotPhaseFinishedSplits.contains(splitId))) { finishedSplits.add(splitId); } splitId = recordsBySplitIds.nextSplit(); @@ -399,6 +419,9 @@ private void assignSplitsAndFetchUntilRetrieveRecords( recordsBySplitIds.recycle(); } + assertThat(snapshotPhaseFinishedSplits) + .containsExactlyInAnyOrderElementsOf(expectedSnapshotPhaseFinishedSplits); + // now, verify the records consumed from each split. verifyConsumedRecords(splitConsumedRecords, expectedRecords, rowType); } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/RecordAndPosTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/RecordAndPosTest.java index 3a975b3e48b..c539c07ec18 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/RecordAndPosTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/RecordAndPosTest.java @@ -55,4 +55,25 @@ void testRecordAndPos() { .isEqualTo( "RecordAndPos{scanRecord=+A(1,null,3)@0, readRecordsCount=3, currentSplitIndex=0}"); } + + @Test + void testSnapshotPhaseFinishedMarker() { + RecordAndPos marker = RecordAndPos.snapshotPhaseFinished(); + RecordAndPos equalMarker = RecordAndPos.snapshotPhaseFinished(); + RecordAndPos regularNullRecord = new RecordAndPos(null); + + assertThat(marker.record()).isNull(); + assertThat(marker.isSnapshotPhaseFinished()).isTrue(); + assertThat(marker).isEqualTo(equalMarker).isNotEqualTo(regularNullRecord); + assertThat(marker.hashCode()).isEqualTo(equalMarker.hashCode()); + assertThat(marker.toString()) + .isEqualTo( + "RecordAndPos{scanRecord=null, readRecordsCount=-1, currentSplitIndex=0, " + + "isSnapshotPhaseFinished=true}"); + + assertThat(regularNullRecord.isSnapshotPhaseFinished()).isFalse(); + assertThat(regularNullRecord.toString()) + .isEqualTo( + "RecordAndPos{scanRecord=null, readRecordsCount=-1, currentSplitIndex=0}"); + } } From 5427859a340029be8e3a4852bbd052b831f2edb8 Mon Sep 17 00:00:00 2001 From: HuangXiao Date: Wed, 19 Aug 2026 20:12:09 +0800 Subject: [PATCH 2/2] address comments --- .../source/reader/FlinkSourceSplitReaderTest.java | 13 ++++++++++--- .../fluss/flink/source/reader/RecordAndPosTest.java | 7 ------- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java index bef07bab3d0..c340fad347e 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java @@ -23,6 +23,7 @@ import org.apache.fluss.client.table.writer.AppendWriter; import org.apache.fluss.client.table.writer.UpsertWriter; import org.apache.fluss.client.write.HashBucketAssigner; +import org.apache.fluss.flink.lake.split.LakeSnapshotAndFlussLogSplit; import org.apache.fluss.flink.source.metrics.FlinkSourceReaderMetrics; import org.apache.fluss.flink.source.split.HybridSnapshotLogSplit; import org.apache.fluss.flink.source.split.LogSplit; @@ -373,9 +374,15 @@ private void assignSplitsAndFetchUntilRetrieveRecords( Map> splitConsumedRecords = new HashMap<>(); Set expectedSnapshotPhaseFinishedSplits = new HashSet<>(); for (SourceSplitBase split : splits) { - if (split.isHybridSnapshotLogSplit() - && !split.asHybridSnapshotLogSplit().isBatch() - && !split.asHybridSnapshotLogSplit().isSnapshotFinished()) { + boolean isUnfinishedStreamingHybridSplit = + split.isHybridSnapshotLogSplit() + && !split.asHybridSnapshotLogSplit().isBatch() + && !split.asHybridSnapshotLogSplit().isSnapshotFinished(); + boolean isUnfinishedStreamingLakeSplit = + split instanceof LakeSnapshotAndFlussLogSplit + && ((LakeSnapshotAndFlussLogSplit) split).isStreaming() + && !((LakeSnapshotAndFlussLogSplit) split).isLakeSplitFinished(); + if (isUnfinishedStreamingHybridSplit || isUnfinishedStreamingLakeSplit) { expectedSnapshotPhaseFinishedSplits.add(split.splitId()); } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/RecordAndPosTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/RecordAndPosTest.java index c539c07ec18..b49b454c967 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/RecordAndPosTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/RecordAndPosTest.java @@ -66,14 +66,7 @@ void testSnapshotPhaseFinishedMarker() { assertThat(marker.isSnapshotPhaseFinished()).isTrue(); assertThat(marker).isEqualTo(equalMarker).isNotEqualTo(regularNullRecord); assertThat(marker.hashCode()).isEqualTo(equalMarker.hashCode()); - assertThat(marker.toString()) - .isEqualTo( - "RecordAndPos{scanRecord=null, readRecordsCount=-1, currentSplitIndex=0, " - + "isSnapshotPhaseFinished=true}"); assertThat(regularNullRecord.isSnapshotPhaseFinished()).isFalse(); - assertThat(regularNullRecord.toString()) - .isEqualTo( - "RecordAndPos{scanRecord=null, readRecordsCount=-1, currentSplitIndex=0}"); } }