[flink] Preserve log-only source restore mode

[luoyuxia]

Purpose

Linked issue: close #3354

Preserve the original Fluss-only (non-lake) startup semantics after a Flink source restore. When the source starts from a specified Fluss log position, it should not initialize lake snapshot splits after restore just because lakeSource is available again.

Brief change log

• Persist an empty remainingHybridLakeFlussSplits list when the enumerator checkpoints with lakeSource == null.
• Allow empty pending hybrid split lists to deserialize without lakeSource, while still requiring lakeSource for non-empty hybrid split state.
• Document the three-state meaning of pendingHybridLakeFlussSplits so null, empty, and non-empty states are explicit.
• Add a restore regression test that checkpoints a Fluss-only enumerator, restores it with a non-null lakeSource and a registered lake snapshot, and verifies no LakeSnapshotSplit is generated.
• Add a serializer round-trip test for empty pending hybrid split state with lakeSource == null.

Tests

• ./mvnw -pl fluss-flink/fluss-flink-common -Dtest=SourceEnumeratorStateSerializerTest,FlinkSourceEnumeratorTest#testRestoreFlussOnlySourceWithLakeSourceDoesNotGenerateLakeSplits test
• ./mvnw -pl fluss-flink/fluss-flink-common spotless:apply
• git diff --check

API and Format

No public API, storage format, or checkpoint serializer version change. The fix only changes the value stored in the existing enumerator state field for Fluss-only source checkpoints and permits that empty-state sentinel to deserialize without a lake source.

Documentation

No new feature or user-facing configuration change.

Generative AI disclosure

• Yes: OpenAI Codex was used to author this PR following the repository AGENTS.md guidance.

[luoyuxia]

@loserwang1024 Could you please help review

[loserwang1024]

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.

[luoyuxia]

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.

[luoyuxia]

@loserwang1024 Comments has been addressed.

[fresh-borzoni]

@luoyuxia Ty for the PR, I agree with the sync change and the serializer reorder, but I have comment about checkpointTriggeredBefore guard.

[fresh-borzoni]

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?