This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 03d24f121 [flink] Preserve log-only source restore mode (#3355)
03d24f121 is described below
commit 03d24f12165a9e5451c727e2de4ed9d434b637c5
Author: yuxia Luo <[email protected]>
AuthorDate: Thu May 28 11:42:02 2026 +0800
[flink] Preserve log-only source restore mode (#3355)
---
.../org/apache/fluss/flink/source/FlinkSource.java | 13 ++-
.../source/enumerator/FlinkSourceEnumerator.java | 40 +++++---
.../FlussSourceEnumeratorStateSerializer.java | 5 +-
.../enumerator/FlinkSourceEnumeratorTest.java | 101 ++++++++++++++++++++-
.../state/SourceEnumeratorStateSerializerTest.java | 22 +++++
5 files changed, 163 insertions(+), 18 deletions(-)
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
index 8535a881a..2ecfca09d 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
@@ -52,6 +52,9 @@ import org.apache.flink.core.io.SimpleVersionedSerializer;
import javax.annotation.Nullable;
+import java.util.Collections;
+import java.util.List;
+
import static org.apache.fluss.config.ConfigOptions.CLIENT_SCANNER_IO_TMP_DIR;
import static
org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils.getClientScannerIoTmpDir;
@@ -236,6 +239,14 @@ public class FlinkSource<OUT>
public SplitEnumerator<SourceSplitBase, SourceEnumeratorState>
restoreEnumerator(
SplitEnumeratorContext<SourceSplitBase> splitEnumeratorContext,
SourceEnumeratorState sourceEnumeratorState) {
+ List<SourceSplitBase> remainingHybridLakeFlussSplits =
+ sourceEnumeratorState.getRemainingHybridLakeFlussSplits();
+ // A fresh null means lake splits are not initialized yet. When
restoring, null means
+ // nothing is pending, so normalize it here to avoid generating lake
splits later.
+ if (remainingHybridLakeFlussSplits == null) {
+ remainingHybridLakeFlussSplits = Collections.emptyList();
+ }
+
return new FlinkSourceEnumerator(
tablePath,
flussConf,
@@ -244,7 +255,7 @@ public class FlinkSource<OUT>
splitEnumeratorContext,
sourceEnumeratorState.getAssignedBuckets(),
sourceEnumeratorState.getAssignedPartitions(),
- sourceEnumeratorState.getRemainingHybridLakeFlussSplits(),
+ remainingHybridLakeFlussSplits,
offsetsInitializer,
scanPartitionDiscoveryIntervalMs,
splitPerAssignmentBatchSize,
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
index f58c4adfa..d107712ff 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
@@ -133,6 +133,20 @@ public class FlinkSourceEnumerator
/** Buckets that have been assigned to readers. */
private final Set<TableBucket> assignedTableBuckets;
+ /**
+ * Remaining lake snapshot and hybrid lake/Fluss splits to assign.
+ *
+ * <p>The field has three states:
+ *
+ * <ul>
+ * <li>{@code null}: lake split initialization has not run yet, or the
source has no lake
+ * (non-lake table) so initialization will never run.
+ * <li>empty list: lake split initialization has run, or this enumerator
was started in
+ * Fluss-only (non-lake) mode and must not initialize lake splits
after restore.
+ * <li>non-empty list: lake split initialization has run and these
splits still need to be
+ * assigned.
+ * </ul>
+ */
@Nullable private List<SourceSplitBase> pendingHybridLakeFlussSplits;
private final long scanPartitionDiscoveryIntervalMs;
@@ -513,18 +527,15 @@ public class FlinkSourceEnumerator
private void startInStreamModeForNonPartitionedTable() {
if (lakeSource != null) {
- context.callAsync(
- () -> {
- // firstly, try to generate hybrid lake splits,
- List<SourceSplitBase> splits =
generateHybridLakeFlussSplits();
- // splits is null,
- // we'll fall back to normal fluss splits generation
logic
- if (splits == null) {
- splits = this.initNonPartitionedSplits();
- }
- return splits;
- },
- this::handleSplitsAdd);
+ // Generate lake splits synchronously so that they are available
before the
+ // first checkpoint. This is consistent with the partitioned-table
path in
+ // start().
+ List<SourceSplitBase> splits = generateHybridLakeFlussSplits();
+ if (splits == null) {
+ // no lake snapshot, fall back to normal Fluss splits
+ splits = this.initNonPartitionedSplits();
+ }
+ handleSplitsAdd(splits, null);
} else {
// init bucket splits and assign
context.callAsync(this::initNonPartitionedSplits,
this::handleSplitsAdd);
@@ -872,9 +883,8 @@ public class FlinkSourceEnumerator
/** Return the hybrid lake and fluss splits. Return null if no lake
snapshot. */
@Nullable
private List<SourceSplitBase> generateHybridLakeFlussSplits() {
- // still have pending lake fluss splits,
- // should be restored from checkpoint, shouldn't
- // list splits again
+ // Restored from checkpoint with pending lake splits — return them
directly
+ // without re-generating.
if (pendingHybridLakeFlussSplits != null) {
LOG.info("Still have pending lake fluss splits, shouldn't list
splits again.");
return pendingHybridLakeFlussSplits;
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/FlussSourceEnumeratorStateSerializer.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/FlussSourceEnumeratorStateSerializer.java
index 121041df1..996313e60 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/FlussSourceEnumeratorStateSerializer.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/FlussSourceEnumeratorStateSerializer.java
@@ -270,12 +270,15 @@ public class FlussSourceEnumeratorStateSerializer
if (in.readBoolean()) {
int numSplits = in.readInt();
List<SourceSplitBase> splits = new ArrayList<>(numSplits);
+ int version = in.readInt();
+ if (numSplits == 0) {
+ return splits;
+ }
SourceSplitSerializer sourceSplitSerializer =
new SourceSplitSerializer(
checkNotNull(
lakeSource,
"lake source must not be null when there
are hybrid lake splits."));
- int version = in.readInt();
for (int i = 0; i < numSplits; i++) {
int splitSizeInBytes = in.readInt();
byte[] splitBytes = new byte[splitSizeInBytes];
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
index 6d239f965..184e1b6e7 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
@@ -25,6 +25,8 @@ import org.apache.fluss.config.Configuration;
import org.apache.fluss.flink.FlinkConnectorOptions;
import org.apache.fluss.flink.lake.split.LakeSnapshotAndFlussLogSplit;
import org.apache.fluss.flink.lake.split.LakeSnapshotSplit;
+import org.apache.fluss.flink.source.FlinkSource;
+import org.apache.fluss.flink.source.deserializer.RowDataDeserializationSchema;
import org.apache.fluss.flink.source.event.PartitionBucketsUnsubscribedEvent;
import org.apache.fluss.flink.source.event.PartitionsRemovedEvent;
import org.apache.fluss.flink.source.reader.LeaseContext;
@@ -32,6 +34,7 @@ import
org.apache.fluss.flink.source.split.HybridSnapshotLogSplit;
import org.apache.fluss.flink.source.split.LogSplit;
import org.apache.fluss.flink.source.split.SnapshotSplit;
import org.apache.fluss.flink.source.split.SourceSplitBase;
+import org.apache.fluss.flink.source.state.SourceEnumeratorState;
import org.apache.fluss.flink.utils.FlinkTestBase;
import org.apache.fluss.lake.source.LakeSource;
import org.apache.fluss.lake.source.LakeSplit;
@@ -53,8 +56,10 @@ import org.apache.fluss.types.DataTypes;
import org.apache.flink.api.connector.source.ReaderInfo;
import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitsAssignment;
import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
+import org.apache.flink.table.data.RowData;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -210,6 +215,100 @@ class FlinkSourceEnumeratorTest extends FlinkTestBase {
}
}
+ @Test
+ void
testRestoreFlussOnlySourceWithLakeSourceDoesNotGenerateLakeSplits(@TempDir Path
tempDir)
+ throws Throwable {
+ long tableId =
+ createTable(DEFAULT_TABLE_PATH,
DEFAULT_AUTO_PARTITIONED_LOG_TABLE_DESCRIPTOR);
+ ZooKeeperClient zooKeeperClient =
FLUSS_CLUSTER_EXTENSION.getZooKeeperClient();
+ Map<Long, String> partitionNameByIds =
+ waitUntilPartitions(zooKeeperClient, DEFAULT_TABLE_PATH);
+ Long partitionId =
partitionNameByIds.keySet().stream().sorted().findFirst().get();
+ String partitionName = partitionNameByIds.get(partitionId);
+
+ LakeTableSnapshot lakeTableSnapshot =
+ new LakeTableSnapshot(
+ 0,
+ ImmutableMap.of(
+ new TableBucket(tableId, partitionId, 0), 50L,
+ new TableBucket(tableId, partitionId, 1), 50L,
+ new TableBucket(tableId, partitionId, 2),
50L));
+ LakeTableHelper lakeTableHelper = new LakeTableHelper(zooKeeperClient,
tempDir.toString());
+ lakeTableHelper.registerLakeTableSnapshotV1(tableId,
lakeTableSnapshot);
+
+ ResolvedPartitionSpec partitionSpec =
+ ResolvedPartitionSpec.fromPartitionName(
+ Collections.singletonList("name"), partitionName);
+ LakeSource<LakeSplit> lakeSource =
+ new TestingLakeSource(
+ DEFAULT_BUCKET_NUM,
+ Collections.singletonList(
+ new PartitionInfo(
+ partitionId, partitionSpec,
DEFAULT_REMOTE_DATA_DIR)));
+
+ SourceEnumeratorState checkpointState;
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(1);
+ SplitEnumerator<SourceSplitBase, SourceEnumeratorState>
enumerator =
+ new FlinkSource<RowData>(
+ flussConf,
+ DEFAULT_TABLE_PATH,
+ false,
+ true,
+ DEFAULT_LOG_TABLE_SCHEMA.getRowType(),
+ null,
+ null,
+ OffsetsInitializer.timestamp(1000L),
+ 0L,
+ new RowDataDeserializationSchema(),
+ streaming,
+ null,
+ LeaseContext.DEFAULT)
+ .createEnumerator(context)) {
+ checkpointState = enumerator.snapshotState(1L);
+
assertThat(checkpointState.getRemainingHybridLakeFlussSplits()).isNull();
+ }
+
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(DEFAULT_BUCKET_NUM);
+ SplitEnumerator<SourceSplitBase, SourceEnumeratorState>
restoredEnumerator =
+ new FlinkSource<RowData>(
+ flussConf,
+ DEFAULT_TABLE_PATH,
+ false,
+ true,
+ DEFAULT_LOG_TABLE_SCHEMA.getRowType(),
+ null,
+ null,
+ OffsetsInitializer.full(),
+ 0L,
+ new RowDataDeserializationSchema(),
+ streaming,
+ null,
+ lakeSource,
+ LeaseContext.DEFAULT)
+ .restoreEnumerator(context, checkpointState)) {
+
assertThat(restoredEnumerator.snapshotState(1L).getRemainingHybridLakeFlussSplits())
+ .isEmpty();
+
+ restoredEnumerator.start();
+ context.runNextOneTimeCallable();
+ context.runNextOneTimeCallable();
+
+ for (int i = 0; i < DEFAULT_BUCKET_NUM; i++) {
+ registerReader(context, restoredEnumerator, i);
+ }
+
+ List<SourceSplitBase> assignedSplits =
+ getReadersAssignments(context).values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+ assertThat(assignedSplits).isNotEmpty();
+ assertThat(assignedSplits).allMatch(split -> split instanceof
LogSplit);
+ assertThat(assignedSplits).noneMatch(split -> split instanceof
LakeSnapshotSplit);
+ }
+ }
+
@Test
void testPkTableWithSnapshotSplits() throws Throwable {
long tableId = createTable(DEFAULT_TABLE_PATH,
DEFAULT_PK_TABLE_DESCRIPTOR);
@@ -852,7 +951,7 @@ class FlinkSourceEnumeratorTest extends FlinkTestBase {
// ---------------------
private void registerReader(
MockSplitEnumeratorContext<SourceSplitBase> context,
- FlinkSourceEnumerator enumerator,
+ SplitEnumerator<SourceSplitBase, SourceEnumeratorState> enumerator,
int readerId) {
context.registerReader(new ReaderInfo(readerId, "location " +
readerId));
enumerator.addReader(readerId);
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java
index 5273d12d1..fa90eda96 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java
@@ -169,4 +169,26 @@ class SourceEnumeratorStateSerializerTest {
serializer.deserialize(serializer.getVersion(), serialized);
assertThat(deserializedSourceEnumeratorState).isEqualTo(sourceEnumeratorState);
}
+
+ @Test
+ void testEmptyPendingSplitsCheckpointSerdeWithoutLakeSource() throws
Exception {
+ FlussSourceEnumeratorStateSerializer serializer =
+ new FlussSourceEnumeratorStateSerializer(null);
+
+ SourceEnumeratorState sourceEnumeratorState =
+ new SourceEnumeratorState(
+ Collections.emptySet(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ LeaseContext.DEFAULT.getKvSnapshotLeaseId());
+
+ byte[] serialized = serializer.serialize(sourceEnumeratorState);
+ SourceEnumeratorState deserializedSourceEnumeratorState =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
+
assertThat(deserializedSourceEnumeratorState).isEqualTo(sourceEnumeratorState);
+
assertThat(deserializedSourceEnumeratorState.getRemainingHybridLakeFlussSplits())
+ .isNotNull()
+ .isEmpty();
+ }
}
