leonardBang commented on code in PR #3548:
URL: https://github.com/apache/fluss/pull/3548#discussion_r3499986097
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/SourceEnumeratorState.java:
##########
@@ -44,15 +46,62 @@ public class SourceEnumeratorState {
// lease context for restore.
private final String leaseId;
+ /**
+ * Whether the initial partition discovery has been completed. Following
FLIP-288, partitions
+ * discovered after the initial startup always use earliest offsets to
prevent data loss.
+ */
+ private final boolean initialDiscoveryFinished;
+
+ /**
+ * Splits that have been initialized (offsets resolved) but not yet
assigned to readers.
+ * Following Kafka's FLIP-288 pattern, these are persisted in checkpoint
state so that on
+ * restore they can be directly assigned without re-initialization,
preserving the original
+ * offset strategy.
+ */
+ private final Collection<SourceSplitBase> unassignedSplits;
+
public SourceEnumeratorState(
Set<TableBucket> assignedBuckets,
Map<Long, String> assignedPartitions,
@Nullable List<SourceSplitBase> remainingHybridLakeFlussSplits,
String leaseId) {
+ this(
+ assignedBuckets,
+ assignedPartitions,
+ remainingHybridLakeFlussSplits,
+ leaseId,
+ false,
Review Comment:
This default is used when restoring old V2 enumerator state, but it leaves
initialDiscoveryFinished=false. After an upgrade, partitions created after that
old checkpoint can be classified as initial partitions and use the user startup
mode (for example latest), which reintroduces the data-loss case this patch is
trying to avoid. Could we default old V2 state to true here, or otherwise make
the migration preserve post-initial discovery semantics?
--
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]
