Skip to content

Commit 03d24f1

Browse files
luoyuxialuoyuxia
authored
[flink] Preserve log-only source restore mode (#3355)
1 parent b2c4a84 commit 03d24f1

5 files changed

Lines changed: 163 additions & 18 deletions

File tree

‎fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,9 @@
5252

5353
import javax.annotation.Nullable;
5454

55+
import java.util.Collections;
56+
import java.util.List;
57+
5558
import static org.apache.fluss.config.ConfigOptions.CLIENT_SCANNER_IO_TMP_DIR;
5659
import static org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils.getClientScannerIoTmpDir;
5760

@@ -236,6 +239,14 @@ public SplitEnumerator<SourceSplitBase, SourceEnumeratorState> createEnumerator(
236239
public SplitEnumerator<SourceSplitBase, SourceEnumeratorState> restoreEnumerator(
237240
SplitEnumeratorContext<SourceSplitBase> splitEnumeratorContext,
238241
SourceEnumeratorState sourceEnumeratorState) {
242+
List<SourceSplitBase> remainingHybridLakeFlussSplits =
243+
sourceEnumeratorState.getRemainingHybridLakeFlussSplits();
244+
// A fresh null means lake splits are not initialized yet. When restoring, null means
245+
// nothing is pending, so normalize it here to avoid generating lake splits later.
246+
if (remainingHybridLakeFlussSplits == null) {
247+
remainingHybridLakeFlussSplits = Collections.emptyList();
248+
}
249+
239250
return new FlinkSourceEnumerator(
240251
tablePath,
241252
flussConf,
@@ -244,7 +255,7 @@ public SplitEnumerator<SourceSplitBase, SourceEnumeratorState> restoreEnumerator
244255
splitEnumeratorContext,
245256
sourceEnumeratorState.getAssignedBuckets(),
246257
sourceEnumeratorState.getAssignedPartitions(),
247-
sourceEnumeratorState.getRemainingHybridLakeFlussSplits(),
258+
remainingHybridLakeFlussSplits,
248259
offsetsInitializer,
249260
scanPartitionDiscoveryIntervalMs,
250261
splitPerAssignmentBatchSize,

‎fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java‎

Lines changed: 25 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -133,6 +133,20 @@ public class FlinkSourceEnumerator
133133
/** Buckets that have been assigned to readers. */
134134
private final Set<TableBucket> assignedTableBuckets;
135135

136+
/**
137+
* Remaining lake snapshot and hybrid lake/Fluss splits to assign.
138+
*
139+
* <p>The field has three states:
140+
*
141+
* <ul>
142+
* <li>{@code null}: lake split initialization has not run yet, or the source has no lake
143+
* (non-lake table) so initialization will never run.
144+
* <li>empty list: lake split initialization has run, or this enumerator was started in
145+
* Fluss-only (non-lake) mode and must not initialize lake splits after restore.
146+
* <li>non-empty list: lake split initialization has run and these splits still need to be
147+
* assigned.
148+
* </ul>
149+
*/
136150
@Nullable private List<SourceSplitBase> pendingHybridLakeFlussSplits;
137151

138152
private final long scanPartitionDiscoveryIntervalMs;
@@ -513,18 +527,15 @@ private void startInBatchMode() {
513527

514528
private void startInStreamModeForNonPartitionedTable() {
515529
if (lakeSource != null) {
516-
context.callAsync(
517-
() -> {
518-
// firstly, try to generate hybrid lake splits,
519-
List<SourceSplitBase> splits = generateHybridLakeFlussSplits();
520-
// splits is null,
521-
// we'll fall back to normal fluss splits generation logic
522-
if (splits == null) {
523-
splits = this.initNonPartitionedSplits();
524-
}
525-
return splits;
526-
},
527-
this::handleSplitsAdd);
530+
// Generate lake splits synchronously so that they are available before the
531+
// first checkpoint. This is consistent with the partitioned-table path in
532+
// start().
533+
List<SourceSplitBase> splits = generateHybridLakeFlussSplits();
534+
if (splits == null) {
535+
// no lake snapshot, fall back to normal Fluss splits
536+
splits = this.initNonPartitionedSplits();
537+
}
538+
handleSplitsAdd(splits, null);
528539
} else {
529540
// init bucket splits and assign
530541
context.callAsync(this::initNonPartitionedSplits, this::handleSplitsAdd);
@@ -872,9 +883,8 @@ private List<SourceSplitBase> getLogSplit(
872883
/** Return the hybrid lake and fluss splits. Return null if no lake snapshot. */
873884
@Nullable
874885
private List<SourceSplitBase> generateHybridLakeFlussSplits() {
875-
// still have pending lake fluss splits,
876-
// should be restored from checkpoint, shouldn't
877-
// list splits again
886+
// Restored from checkpoint with pending lake splits — return them directly
887+
// without re-generating.
878888
if (pendingHybridLakeFlussSplits != null) {
879889
LOG.info("Still have pending lake fluss splits, shouldn't list splits again.");
880890
return pendingHybridLakeFlussSplits;

‎fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/FlussSourceEnumeratorStateSerializer.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -270,12 +270,15 @@ private List<SourceSplitBase> deserializeRemainingHybridLakeFlussSplits(
270270
if (in.readBoolean()) {
271271
int numSplits = in.readInt();
272272
List<SourceSplitBase> splits = new ArrayList<>(numSplits);
273+
int version = in.readInt();
274+
if (numSplits == 0) {
275+
return splits;
276+
}
273277
SourceSplitSerializer sourceSplitSerializer =
274278
new SourceSplitSerializer(
275279
checkNotNull(
276280
lakeSource,
277281
"lake source must not be null when there are hybrid lake splits."));
278-
int version = in.readInt();
279282
for (int i = 0; i < numSplits; i++) {
280283
int splitSizeInBytes = in.readInt();
281284
byte[] splitBytes = new byte[splitSizeInBytes];

‎fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java‎

Lines changed: 100 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,16 @@
2525
import org.apache.fluss.flink.FlinkConnectorOptions;
2626
import org.apache.fluss.flink.lake.split.LakeSnapshotAndFlussLogSplit;
2727
import org.apache.fluss.flink.lake.split.LakeSnapshotSplit;
28+
import org.apache.fluss.flink.source.FlinkSource;
29+
import org.apache.fluss.flink.source.deserializer.RowDataDeserializationSchema;
2830
import org.apache.fluss.flink.source.event.PartitionBucketsUnsubscribedEvent;
2931
import org.apache.fluss.flink.source.event.PartitionsRemovedEvent;
3032
import org.apache.fluss.flink.source.reader.LeaseContext;
3133
import org.apache.fluss.flink.source.split.HybridSnapshotLogSplit;
3234
import org.apache.fluss.flink.source.split.LogSplit;
3335
import org.apache.fluss.flink.source.split.SnapshotSplit;
3436
import org.apache.fluss.flink.source.split.SourceSplitBase;
37+
import org.apache.fluss.flink.source.state.SourceEnumeratorState;
3538
import org.apache.fluss.flink.utils.FlinkTestBase;
3639
import org.apache.fluss.lake.source.LakeSource;
3740
import org.apache.fluss.lake.source.LakeSplit;
@@ -53,8 +56,10 @@
5356

5457
import org.apache.flink.api.connector.source.ReaderInfo;
5558
import org.apache.flink.api.connector.source.SourceEvent;
59+
import org.apache.flink.api.connector.source.SplitEnumerator;
5660
import org.apache.flink.api.connector.source.SplitsAssignment;
5761
import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
62+
import org.apache.flink.table.data.RowData;
5863
import org.junit.jupiter.api.BeforeAll;
5964
import org.junit.jupiter.api.Test;
6065
import org.junit.jupiter.api.io.TempDir;
@@ -210,6 +215,100 @@ void testInvalidSplitAssignmentBatchSize() throws Exception {
210215
}
211216
}
212217

218+
@Test
219+
void testRestoreFlussOnlySourceWithLakeSourceDoesNotGenerateLakeSplits(@TempDir Path tempDir)
220+
throws Throwable {
221+
long tableId =
222+
createTable(DEFAULT_TABLE_PATH, DEFAULT_AUTO_PARTITIONED_LOG_TABLE_DESCRIPTOR);
223+
ZooKeeperClient zooKeeperClient = FLUSS_CLUSTER_EXTENSION.getZooKeeperClient();
224+
Map<Long, String> partitionNameByIds =
225+
waitUntilPartitions(zooKeeperClient, DEFAULT_TABLE_PATH);
226+
Long partitionId = partitionNameByIds.keySet().stream().sorted().findFirst().get();
227+
String partitionName = partitionNameByIds.get(partitionId);
228+
229+
LakeTableSnapshot lakeTableSnapshot =
230+
new LakeTableSnapshot(
231+
0,
232+
ImmutableMap.of(
233+
new TableBucket(tableId, partitionId, 0), 50L,
234+
new TableBucket(tableId, partitionId, 1), 50L,
235+
new TableBucket(tableId, partitionId, 2), 50L));
236+
LakeTableHelper lakeTableHelper = new LakeTableHelper(zooKeeperClient, tempDir.toString());
237+
lakeTableHelper.registerLakeTableSnapshotV1(tableId, lakeTableSnapshot);
238+
239+
ResolvedPartitionSpec partitionSpec =
240+
ResolvedPartitionSpec.fromPartitionName(
241+
Collections.singletonList("name"), partitionName);
242+
LakeSource<LakeSplit> lakeSource =
243+
new TestingLakeSource(
244+
DEFAULT_BUCKET_NUM,
245+
Collections.singletonList(
246+
new PartitionInfo(
247+
partitionId, partitionSpec, DEFAULT_REMOTE_DATA_DIR)));
248+
249+
SourceEnumeratorState checkpointState;
250+
try (MockSplitEnumeratorContext<SourceSplitBase> context =
251+
new MockSplitEnumeratorContext<>(1);
252+
SplitEnumerator<SourceSplitBase, SourceEnumeratorState> enumerator =
253+
new FlinkSource<RowData>(
254+
flussConf,
255+
DEFAULT_TABLE_PATH,
256+
false,
257+
true,
258+
DEFAULT_LOG_TABLE_SCHEMA.getRowType(),
259+
null,
260+
null,
261+
OffsetsInitializer.timestamp(1000L),
262+
0L,
263+
new RowDataDeserializationSchema(),
264+
streaming,
265+
null,
266+
LeaseContext.DEFAULT)
267+
.createEnumerator(context)) {
268+
checkpointState = enumerator.snapshotState(1L);
269+
assertThat(checkpointState.getRemainingHybridLakeFlussSplits()).isNull();
270+
}
271+
272+
try (MockSplitEnumeratorContext<SourceSplitBase> context =
273+
new MockSplitEnumeratorContext<>(DEFAULT_BUCKET_NUM);
274+
SplitEnumerator<SourceSplitBase, SourceEnumeratorState> restoredEnumerator =
275+
new FlinkSource<RowData>(
276+
flussConf,
277+
DEFAULT_TABLE_PATH,
278+
false,
279+
true,
280+
DEFAULT_LOG_TABLE_SCHEMA.getRowType(),
281+
null,
282+
null,
283+
OffsetsInitializer.full(),
284+
0L,
285+
new RowDataDeserializationSchema(),
286+
streaming,
287+
null,
288+
lakeSource,
289+
LeaseContext.DEFAULT)
290+
.restoreEnumerator(context, checkpointState)) {
291+
assertThat(restoredEnumerator.snapshotState(1L).getRemainingHybridLakeFlussSplits())
292+
.isEmpty();
293+
294+
restoredEnumerator.start();
295+
context.runNextOneTimeCallable();
296+
context.runNextOneTimeCallable();
297+
298+
for (int i = 0; i < DEFAULT_BUCKET_NUM; i++) {
299+
registerReader(context, restoredEnumerator, i);
300+
}
301+
302+
List<SourceSplitBase> assignedSplits =
303+
getReadersAssignments(context).values().stream()
304+
.flatMap(List::stream)
305+
.collect(Collectors.toList());
306+
assertThat(assignedSplits).isNotEmpty();
307+
assertThat(assignedSplits).allMatch(split -> split instanceof LogSplit);
308+
assertThat(assignedSplits).noneMatch(split -> split instanceof LakeSnapshotSplit);
309+
}
310+
}
311+
213312
@Test
214313
void testPkTableWithSnapshotSplits() throws Throwable {
215314
long tableId = createTable(DEFAULT_TABLE_PATH, DEFAULT_PK_TABLE_DESCRIPTOR);
@@ -852,7 +951,7 @@ void testPartitionsExpiredInFlussButExistInLake(
852951
// ---------------------
853952
private void registerReader(
854953
MockSplitEnumeratorContext<SourceSplitBase> context,
855-
FlinkSourceEnumerator enumerator,
954+
SplitEnumerator<SourceSplitBase, SourceEnumeratorState> enumerator,
856955
int readerId) {
857956
context.registerReader(new ReaderInfo(readerId, "location " + readerId));
858957
enumerator.addReader(readerId);

‎fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -169,4 +169,26 @@ void testInconsistentLakeSourceSerde() throws Exception {
169169
serializer.deserialize(serializer.getVersion(), serialized);
170170
assertThat(deserializedSourceEnumeratorState).isEqualTo(sourceEnumeratorState);
171171
}
172+
173+
@Test
174+
void testEmptyPendingSplitsCheckpointSerdeWithoutLakeSource() throws Exception {
175+
FlussSourceEnumeratorStateSerializer serializer =
176+
new FlussSourceEnumeratorStateSerializer(null);
177+
178+
SourceEnumeratorState sourceEnumeratorState =
179+
new SourceEnumeratorState(
180+
Collections.emptySet(),
181+
Collections.emptyMap(),
182+
Collections.emptyList(),
183+
LeaseContext.DEFAULT.getKvSnapshotLeaseId());
184+
185+
byte[] serialized = serializer.serialize(sourceEnumeratorState);
186+
SourceEnumeratorState deserializedSourceEnumeratorState =
187+
serializer.deserialize(serializer.getVersion(), serialized);
188+
189+
assertThat(deserializedSourceEnumeratorState).isEqualTo(sourceEnumeratorState);
190+
assertThat(deserializedSourceEnumeratorState.getRemainingHybridLakeFlussSplits())
191+
.isNotNull()
192+
.isEmpty();
193+
}
172194
}

0 commit comments

Comments
 (0)