luoyuxia commented on code in PR #3355:
URL: https://github.com/apache/fluss/pull/3355#discussion_r3272820486
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java:
##########
@@ -1207,11 +1220,16 @@ public void addReader(int subtaskId) {
@Override
public SourceEnumeratorState snapshotState(long checkpointId) {
+ List<SourceSplitBase> remainingHybridLakeFlussSplits =
+ // Preserve Fluss-only (non-lake) startup across restore.
Otherwise a restored
+ // enumerator with a non-null lakeSource would treat null as
"not initialized yet"
+ // and generate lake snapshot splits.
+ lakeSource == null ? Collections.emptyList() :
pendingHybridLakeFlussSplits;
Review Comment:
I agree the original approach in snapshotState() is hard to understand
without sufficient context.
I've reworked the fix to use the checkpointTriggeredBefore flag in
generateHybridLakeFlussSplits() instead. While the restore-awareness logic
still requires some thought, this approach keeps all the complexity contained
within a single method rather than spreading it across snapshotState().
Additionally, I changed startInStreamModeForNonPartitionedTable to call
generateHybridLakeFlussSplits() synchronously, consistent with the
partitioned-table path in start(). This ensures lake split initialization
always completes before any checkpoint can be triggered, which is a
prerequisite for the checkpointTriggeredBefore guard to work correctly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
