fresh-borzoni commented on code in PR #3355:
URL: https://github.com/apache/fluss/pull/3355#discussion_r3280594558
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java:
##########
@@ -869,13 +881,19 @@ private List<SourceSplitBase> getLogSplit(
/** 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;
}
+ // Restored from checkpoint but pending lake split is null(e.g. the
source was
+ // originally started in Fluss-only mode without lake). Do not
generate lake
+ // splits for this restore; mark as initialized and return empty list.
+ if (checkpointTriggeredBefore) {
Review Comment:
I think the checkpointTriggeredBefore guard is solving the restore ambiguity
in the wrong layer.
Fresh null means “not initialized yet”.
Restored null should be normalized to “nothing pending”.
That normalization belongs in FlinkSource.restoreEnumerator(), not in
generateHybridLakeFlussSplits().
WDYT?
--
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]
