loserwang1024 commented on code in PR #3355:
URL: https://github.com/apache/fluss/pull/3355#discussion_r3272468910
##########
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 understand what you mean, but the code seems hard to understand without
sufficient context. I've also thought about this:
- If the enumerator is created by FlinkSource#createEnumerator, it indicates
a stateless restart. Therefore, whether to generate lake splits depends on
whether it's a LakeSource.
- If the enumerator is created by FlinkSource#restoreEnumerator, there's no
need to generate lake splits again. This is because before the first checkpoint
is taken, FlinkSourceEnumerator#start →
FlinkSourceEnumerator#generateHybridLakeFlussSplits has already been executed.
Thus, upon restoration, the lake splits do not need to be regenerated.
Therefore, even if the job was previously started from a specified
timestamp, according to this logic, as long as a checkpoint has been taken,
upon stateful restart it will not read the lake splits again.
--
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]
