This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 7ed81de72 [flink]Newly discovered partition read from earliest rather
than scan.startup.mode (#3548)
7ed81de72 is described below
commit 7ed81de72a85253827d543c57417bb4c727c9a17
Author: Hongshun Wang <[email protected]>
AuthorDate: Thu Jul 2 22:08:58 2026 +0800
[flink]Newly discovered partition read from earliest rather than
scan.startup.mode (#3548)
---
.../org/apache/fluss/flink/source/FlinkSource.java | 4 +-
.../source/enumerator/FlinkSourceEnumerator.java | 386 +++++++++++++++++----
.../FlussSourceEnumeratorStateSerializer.java | 101 +++++-
.../flink/source/state/SourceEnumeratorState.java | 64 +++-
.../enumerator/FlinkSourceEnumeratorTest.java | 324 ++++++++++++++++-
.../state/SourceEnumeratorStateSerializerTest.java | 103 ++++++
6 files changed, 903 insertions(+), 79 deletions(-)
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
index 2ecfca09d..227db3fab 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
@@ -265,7 +265,9 @@ public class FlinkSource<OUT>
new LeaseContext(
sourceEnumeratorState.getLeaseId(),
leaseContext.getKvSnapshotLeaseDurationMs()),
- true);
+ true,
+ sourceEnumeratorState.isInitialDiscoveryFinished(),
+ sourceEnumeratorState.getUnassignedSplits());
}
@Override
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
index d107712ff..887eabf3b 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
@@ -155,6 +155,46 @@ public class FlinkSourceEnumerator
private final OffsetsInitializer startingOffsetsInitializer;
private final OffsetsInitializer stoppingOffsetsInitializer;
+ /**
+ * The offsets initializer used for partitions discovered after the
initial startup. Following
+ * <a
+ *
href="https://cwiki.apache.org/confluence/spaces/FLINK/pages/240881147/FLIP-288+Enable+Dynamic+Partition+Discovery+by+Default+in+Kafka+Source";>FLIP-288</a>)
+ * semantics, newly discovered partitions always start from earliest to
prevent data loss.
+ */
+ private final OffsetsInitializer newDiscoveryOffsetsInitializer;
+
+ /**
+ * Splits whose starting offsets have been initialized but that have not
yet been assigned to
+ * any reader. This map is persisted in checkpoint state (via {@link
#snapshotState}) so that on
+ * failover restore these splits are directly placed into {@link
#pendingSplitAssignment}
+ * without re-initialization, preserving the original offset strategy
determined at first
+ * discovery time (FLIP-288).
+ *
+ * <p>Lifecycle:
+ *
+ * <pre>
+ * ┌───────────────────────────────────┐
┌───────────────────────────────────┐
+ * │ Fluss splits (periodic discovery) │ │ Lake splits (one-time
generation) │
+ * └──────────────┬────────────────────┘
└────────────────┬──────────────────┘
+ * │ │
+ * ▼ ▼
+ * ┌───────────────————————————————————————————————————┐
+ * │unassignedSplits,(initialDiscoveryFinished = true) │◄────
addSplitsBack (reader failure)
+ * └───────┬──────————————————————————————————————————─┘
+ * │ addSplitToPendingAssignments, copy from
unassignedSplits.
+ * ▼
+ * ┌────────────────────────┐
+ * │ pendingSplitAssignment │
+ * └───────────┬────────────┘
+ * │ assignPendingSplits, remove from unassignedSplits
and pendingSplitAssignment.
+ * ▼
+ * ┌────────────────────────┐
+ * │ assignedTableBuckets │
+ * └────────────────────────┘
+ * </pre>
+ */
+ private final Collection<SourceSplitBase> unassignedSplits;
+
private final LeaseContext leaseContext;
/** checkpointId -> tableBuckets who finished consume kv snapshots. */
@@ -170,6 +210,14 @@ public class FlinkSourceEnumerator
// split initializing has finished.
private boolean noMoreNewSplits = false;
+ /**
+ * Whether the initial partition discovery has been completed. Following
FLIP-288, this flag
+ * alone determines offset strategy: partitions discovered before this
flag is set to {@code
+ * true} use the user-configured {@link #startingOffsetsInitializer},
while partitions
+ * discovered after use {@link #newDiscoveryOffsetsInitializer} (earliest)
to prevent data loss.
+ */
+ private boolean initialDiscoveryFinished;
+
private boolean lakeEnabled = false;
private volatile boolean closed = false;
@@ -260,7 +308,9 @@ public class FlinkSourceEnumerator
partitionFilters,
lakeSource,
leaseContext,
- checkpointTriggeredBefore);
+ checkpointTriggeredBefore,
+ false,
+ Collections.emptyList());
}
public FlinkSourceEnumerator(
@@ -278,7 +328,9 @@ public class FlinkSourceEnumerator
@Nullable Predicate partitionFilters,
@Nullable LakeSource<LakeSplit> lakeSource,
LeaseContext leaseContext,
- boolean checkpointTriggeredBefore) {
+ boolean checkpointTriggeredBefore,
+ boolean initialDiscoveryFinished,
+ List<SourceSplitBase> unassignedSplits) {
this(
tablePath,
flussConf,
@@ -295,7 +347,9 @@ public class FlinkSourceEnumerator
partitionFilters,
lakeSource,
leaseContext,
- checkpointTriggeredBefore);
+ checkpointTriggeredBefore,
+ initialDiscoveryFinished,
+ unassignedSplits);
}
public FlinkSourceEnumerator(
@@ -314,7 +368,9 @@ public class FlinkSourceEnumerator
@Nullable Predicate partitionFilters,
@Nullable LakeSource<LakeSplit> lakeSource,
LeaseContext leaseContext,
- boolean checkpointTriggeredBefore) {
+ boolean checkpointTriggeredBefore,
+ boolean initialDiscoveryFinished,
+ Collection<SourceSplitBase> unassignedSplits) {
this(
tablePath,
flussConf,
@@ -332,7 +388,9 @@ public class FlinkSourceEnumerator
lakeSource,
new WorkerExecutor(context),
leaseContext,
- checkpointTriggeredBefore);
+ checkpointTriggeredBefore,
+ initialDiscoveryFinished,
+ unassignedSplits);
}
FlinkSourceEnumerator(
@@ -369,7 +427,9 @@ public class FlinkSourceEnumerator
lakeSource,
workerExecutor,
leaseContext,
- checkpointTriggeredBefore);
+ checkpointTriggeredBefore,
+ false,
+ Collections.emptyList());
}
FlinkSourceEnumerator(
@@ -389,7 +449,9 @@ public class FlinkSourceEnumerator
@Nullable LakeSource<LakeSplit> lakeSource,
WorkerExecutor workerExecutor,
LeaseContext leaseContext,
- boolean checkpointTriggeredBefore) {
+ boolean checkpointTriggeredBefore,
+ boolean initialDiscoveryFinished,
+ Collection<SourceSplitBase> unassignedSplits) {
checkArgument(
splitPerAssignmentBatchSize > 0,
"Split assignment batch size must be positive, but was %s.",
@@ -407,6 +469,7 @@ public class FlinkSourceEnumerator
? null
: new LinkedList<>(pendingHybridLakeFlussSplits);
this.startingOffsetsInitializer = startingOffsetsInitializer;
+ this.newDiscoveryOffsetsInitializer = OffsetsInitializer.earliest();
this.scanPartitionDiscoveryIntervalMs =
scanPartitionDiscoveryIntervalMs;
this.streaming = streaming;
this.partitionFilters = partitionFilters;
@@ -417,10 +480,29 @@ public class FlinkSourceEnumerator
this.leaseContext = leaseContext;
this.checkpointTriggeredBefore = checkpointTriggeredBefore;
this.splitPerAssignmentBatchSize = splitPerAssignmentBatchSize;
+ this.initialDiscoveryFinished = initialDiscoveryFinished;
+ this.unassignedSplits = new ArrayList<>(unassignedSplits);
}
@Override
public void start() {
+ LOG.info(
+ "Starting FlinkSourceEnumerator for table {}: "
+ + "isPartitioned={}, hasPrimaryKey={}, streaming={},
lakeSource={}, "
+ + "initialDiscoveryFinished={},
restoredUnassignedSplits={}, "
+ + "restoredAssignedBuckets={},
restoredPendingLakeSplits={}",
+ tablePath,
+ isPartitioned,
+ hasPrimaryKey,
+ streaming,
+ lakeSource != null,
+ initialDiscoveryFinished,
+ unassignedSplits.size(),
+ assignedTableBuckets.size(),
+ pendingHybridLakeFlussSplits == null
+ ? "null"
+ : pendingHybridLakeFlussSplits.size());
+
// init admin client
connection = ConnectionFactory.createConnection(flussConf);
flussAdmin = connection.getAdmin();
@@ -434,6 +516,17 @@ public class FlinkSourceEnumerator
ExceptionUtils.stripCompletionException(e));
}
+ // Find splits where the start offset has been initialized but not yet
assigned to readers.
+ // These splits must not be reinitialized to keep offsets consistent
with first discovery.
+ if (!unassignedSplits.isEmpty()) {
+ LOG.info(
+ "Restoring {} unassigned splits from checkpoint state "
+ + "into pendingSplitAssignment for table {}.",
+ unassignedSplits.size(),
+ tablePath);
+ addSplitToPendingAssignments(unassignedSplits);
+ }
+
if (isPartitioned) {
if (streaming) {
if (lakeSource != null) {
@@ -509,7 +602,9 @@ public class FlinkSourceEnumerator
// Use log-only splits to avoid generating
mixed split
// types (HybridSnapshotLogSplit + LogSplit)
for
// primary-key tables, which is not supported.
- splits =
this.initLogTablePartitionSplits(partitions);
+ splits =
+ this.initLogTablePartitionSplits(
+ partitions,
startingOffsetsInitializer);
} else {
splits = this.getLogSplit(null, null);
}
@@ -526,6 +621,19 @@ public class FlinkSourceEnumerator
}
private void startInStreamModeForNonPartitionedTable() {
+ // If we have restored unassigned splits from checkpoint state, skip
re-initialization.
+ // These splits already have their offsets resolved and will be
assigned to readers
+ // when they register (via addReader -> assignPendingSplits).
+ if (!pendingSplitAssignment.isEmpty()) {
+ LOG.info(
+ "Skipping split re-initialization for non-partitioned
table {}: "
+ + "{} splits already restored from checkpoint
state.",
+ tablePath,
+
pendingSplitAssignment.values().stream().mapToInt(List::size).sum());
+ initialDiscoveryFinished = true;
+ return;
+ }
+
if (lakeSource != null) {
// Generate lake splits synchronously so that they are available
before the
// first checkpoint. This is consistent with the partitioned-table
path in
@@ -620,8 +728,16 @@ public class FlinkSourceEnumerator
tablePath,
partitionInfos.size());
- final PartitionChange partitionChange =
getPartitionChange(partitionInfos);
+ final PartitionChange partitionChange =
+ getPartitionChange(partitionInfos, !initialDiscoveryFinished);
+
if (partitionChange.isEmpty()) {
+ // No partition changes found. For the empty-table case (no
initial partitions
+ // to track), mark initial discovery as finished immediately since
there are
+ // no splits that need to be persisted in state first.
+ if (!initialDiscoveryFinished) {
+ initialDiscoveryFinished = true;
+ }
LOG.debug("No partition changes detected for table {}", tablePath);
return;
}
@@ -636,21 +752,32 @@ public class FlinkSourceEnumerator
handlePartitionsRemoved(partitionChange.removedPartitions);
}
- // handle new partitions
- if (!partitionChange.newPartitions.isEmpty()) {
+ // handle initial partitions and new partitions
+ boolean hasNewOrInitialPartitions =
+ !partitionChange.initialPartitions.isEmpty()
+ || !partitionChange.newPartitions.isEmpty();
+ if (hasNewOrInitialPartitions) {
+ Collection<Partition> allNewPartitions = new ArrayList<>();
+ allNewPartitions.addAll(partitionChange.initialPartitions);
+ allNewPartitions.addAll(partitionChange.newPartitions);
LOG.info(
- "Handling {} new partitions for table {}: {}",
- partitionChange.newPartitions.size(),
+ "Handling {} partitions for table {} (initial={}, new={}):
{}",
+ allNewPartitions.size(),
tablePath,
- partitionChange.newPartitions);
+ partitionChange.initialPartitions.size(),
+ partitionChange.newPartitions.size(),
+ allNewPartitions);
workerExecutor.callAsync(
- () -> initPartitionedSplits(partitionChange.newPartitions),
- this::handleSplitsAdd);
+ () -> initPartitionedSplits(partitionChange),
+ (splits, throwable) -> {
+ handleSplitsAdd(splits, throwable);
+ });
}
}
- private PartitionChange getPartitionChange(Set<PartitionInfo>
fetchedPartitionInfos) {
- final Set<Partition> newPartitions =
+ private PartitionChange getPartitionChange(
+ Set<PartitionInfo> fetchedPartitionInfos, boolean
initialDiscovery) {
+ final Set<Partition> allNewPartitions =
fetchedPartitionInfos.stream()
.map(p -> new Partition(p.getPartitionId(),
p.getPartitionName()))
.collect(Collectors.toSet());
@@ -679,7 +806,7 @@ public class FlinkSourceEnumerator
assignedOrPendingPartitions.forEach(
p -> {
- if (!newPartitions.remove(p)) {
+ if (!allNewPartitions.remove(p)) {
removedPartitions.add(p);
}
});
@@ -687,25 +814,63 @@ public class FlinkSourceEnumerator
if (!removedPartitions.isEmpty()) {
LOG.info("Discovered removed partitions: {}", removedPartitions);
}
- if (!newPartitions.isEmpty()) {
- LOG.info("Discovered new partitions: {}", newPartitions);
+ if (!allNewPartitions.isEmpty()) {
+ LOG.info("Discovered new partitions: {}", allNewPartitions);
+ }
+
+ // Following Kafka's FLIP-288 pattern: if this is the initial
discovery,
+ // all new partitions are classified as "initial partitions" and will
use
+ // the user-configured offset initializer. After initial discovery is
done,
+ // all new partitions are classified as "new partitions" and will use
earliest.
+ Set<Partition> initialPartitions = new HashSet<>();
+ Set<Partition> newPartitions;
+ if (initialDiscovery) {
+ initialPartitions.addAll(allNewPartitions);
+ newPartitions = Collections.emptySet();
+ } else {
+ newPartitions = allNewPartitions;
}
- return new PartitionChange(newPartitions, removedPartitions);
+ return new PartitionChange(initialPartitions, newPartitions,
removedPartitions);
}
- private List<SourceSplitBase> initPartitionedSplits(Collection<Partition>
newPartitions) {
+ private List<SourceSplitBase> initPartitionedSplits(PartitionChange
partitionChange) {
+ Collection<Partition> initialPartitions =
partitionChange.initialPartitions;
+ Collection<Partition> newPartitions = partitionChange.newPartitions;
+
if (hasPrimaryKey && startingOffsetsInitializer instanceof
SnapshotOffsetsInitializer) {
- return initPrimaryKeyTablePartitionSplits(newPartitions);
+ // Snapshot mode for PK tables is already safe: it reads the
snapshot or falls back
+ // to the earliest offsets when no snapshot is available.
+ List<Partition> allPartitions = new ArrayList<>();
+ allPartitions.addAll(initialPartitions);
+ allPartitions.addAll(newPartitions);
+ return initPrimaryKeyTablePartitionSplits(allPartitions);
} else {
- return initLogTablePartitionSplits(newPartitions);
+ // For log tables (or PK tables in non-snapshot mode), use
FLIP-288 semantics:
+ // - Initial partitions: use user-configured offset
+ // - New partitions: use earliest to prevent data loss
+ List<SourceSplitBase> splits = new ArrayList<>();
+ if (!initialPartitions.isEmpty()) {
+ splits.addAll(
+ initLogTablePartitionSplits(initialPartitions,
startingOffsetsInitializer));
+ }
+ if (!newPartitions.isEmpty()) {
+ splits.addAll(
+ initLogTablePartitionSplits(newPartitions,
newDiscoveryOffsetsInitializer));
+ }
+ return splits;
}
}
- private List<SourceSplitBase>
initLogTablePartitionSplits(Collection<Partition> newPartitions) {
+ private List<SourceSplitBase> initLogTablePartitionSplits(
+ Collection<Partition> newPartitions, OffsetsInitializer
effectiveOffsetsInitializer) {
List<SourceSplitBase> splits = new ArrayList<>();
for (Partition partition : newPartitions) {
- splits.addAll(getLogSplit(partition.getPartitionId(),
partition.getPartitionName()));
+ splits.addAll(
+ getLogSplit(
+ partition.getPartitionId(),
+ partition.getPartitionName(),
+ effectiveOffsetsInitializer));
}
return splits;
}
@@ -848,6 +1013,13 @@ public class FlinkSourceEnumerator
private List<SourceSplitBase> getLogSplit(
@Nullable Long partitionId, @Nullable String partitionName) {
+ return getLogSplit(partitionId, partitionName,
startingOffsetsInitializer);
+ }
+
+ private List<SourceSplitBase> getLogSplit(
+ @Nullable Long partitionId,
+ @Nullable String partitionName,
+ OffsetsInitializer effectiveStartingOffsetsInitializer) {
// always assume the bucket is from 0 to bucket num
List<SourceSplitBase> splits = new ArrayList<>();
List<Integer> bucketsNeedInitOffset = new ArrayList<>();
@@ -862,7 +1034,7 @@ public class FlinkSourceEnumerator
if (!bucketsNeedInitOffset.isEmpty()) {
Map<Integer, Long> startingOffsets =
- startingOffsetsInitializer.getBucketOffsets(
+ effectiveStartingOffsetsInitializer.getBucketOffsets(
partitionName, bucketsNeedInitOffset,
bucketOffsetsRetriever);
Map<Integer, Long> stoppingOffsets =
stoppingOffsetsInitializer.getBucketOffsets(
@@ -887,7 +1059,7 @@ public class FlinkSourceEnumerator
// without re-generating.
if (pendingHybridLakeFlussSplits != null) {
LOG.info("Still have pending lake fluss splits, shouldn't list
splits again.");
- return pendingHybridLakeFlussSplits;
+ return new ArrayList<>(pendingHybridLakeFlussSplits);
}
try {
LakeSplitGenerator lakeSplitGenerator =
@@ -907,7 +1079,7 @@ public class FlinkSourceEnumerator
return null;
} else {
pendingHybridLakeFlussSplits = generatedSplits;
- return generatedSplits;
+ return new ArrayList<>(generatedSplits);
}
} catch (Exception e) {
throw new FlinkRuntimeException("Failed to generate hybrid lake
fluss splits", e);
@@ -935,28 +1107,20 @@ public class FlinkSourceEnumerator
pendingSplitAssignment.forEach(
(reader, splits) ->
splits.removeIf(
- split -> {
- // Never remove LakeSnapshotSplit, because
during union reads,
- // data from the lake must still be read
even if the partition
- // has already expired in Fluss.
- if (split instanceof LakeSnapshotSplit) {
- return false;
- }
-
- // Similar to LakeSnapshotSplit, if it
contains any lake split,
- // never remove it; otherwise, it can be
removed when the Fluss
- // partition expires.
- if (split instanceof
LakeSnapshotAndFlussLogSplit) {
- LakeSnapshotAndFlussLogSplit
hybridSplit =
- (LakeSnapshotAndFlussLogSplit)
split;
- if
(!hybridSplit.isLakeSplitFinished()) {
- return false;
- }
- }
-
- return removedPartitionsMap.containsKey(
-
split.getTableBucket().getPartitionId());
- }));
+ split ->
+ shouldRemoveForDroppedPartition(
+ split, removedPartitionsMap)));
+
+ // remove from unassignedSplits to prevent stale splits from being
checkpointed
+ // and restored after failover for a deleted partition
+ unassignedSplits.removeIf(
+ split -> shouldRemoveForDroppedPartition(split,
removedPartitionsMap));
+
+ // remove from pendingHybridLakeFlussSplits as well
+ if (pendingHybridLakeFlussSplits != null) {
+ pendingHybridLakeFlussSplits.removeIf(
+ split -> shouldRemoveForDroppedPartition(split,
removedPartitionsMap));
+ }
// send partition removed event to all readers
PartitionsRemovedEvent event = new
PartitionsRemovedEvent(removedPartitionsMap);
@@ -965,6 +1129,34 @@ public class FlinkSourceEnumerator
}
}
+ /**
+ * Determines whether a split should be removed when its partition is
dropped.
+ *
+ * <p>Lake-related splits are preserved because lake data must still be
read even if the
+ * partition has expired in Fluss (union reads scenario).
+ */
+ private static boolean shouldRemoveForDroppedPartition(
+ SourceSplitBase split, Map<Long, String> removedPartitionsMap) {
+ // Never remove LakeSnapshotSplit, because during union reads,
+ // data from the lake must still be read even if the partition
+ // has already expired in Fluss.
+ if (split instanceof LakeSnapshotSplit) {
+ return false;
+ }
+
+ // Similar to LakeSnapshotSplit, if it contains any lake split,
+ // never remove it; otherwise, it can be removed when the Fluss
+ // partition expires.
+ if (split instanceof LakeSnapshotAndFlussLogSplit) {
+ LakeSnapshotAndFlussLogSplit hybridSplit =
(LakeSnapshotAndFlussLogSplit) split;
+ if (!hybridSplit.isLakeSplitFinished()) {
+ return false;
+ }
+ }
+
+ return
removedPartitionsMap.containsKey(split.getTableBucket().getPartitionId());
+ }
+
private void handleSplitsAdd(List<SourceSplitBase> splits, Throwable t) {
if (t != null) {
if (isPartitioned && streaming && scanPartitionDiscoveryIntervalMs
> 0) {
@@ -978,6 +1170,26 @@ public class FlinkSourceEnumerator
t);
}
}
+
+ initialDiscoveryFinished = true;
+ if (pendingHybridLakeFlussSplits != null) {
+ // removed from the pendingHybridLakeFlussSplits since this split
already be moved to
+ // unassignedSplits
+ pendingHybridLakeFlussSplits.removeAll(splits);
+ }
+ unassignedSplits.addAll(splits);
+ LOG.info(
+ "Added {} new splits to unassignedSplits for table {}: "
+ + "totalUnassigned={}, initialDiscoveryFinished={}, "
+ + "remainingLakeSplits={}",
+ splits.size(),
+ tablePath,
+ unassignedSplits.size(),
+ initialDiscoveryFinished,
+ pendingHybridLakeFlussSplits == null
+ ? "null"
+ : pendingHybridLakeFlussSplits.size());
+
if (isPartitioned) {
if (!streaming || scanPartitionDiscoveryIntervalMs <= 0) {
// if not streaming or partition discovery is disabled
@@ -1026,6 +1238,7 @@ public class FlinkSourceEnumerator
split -> {
TableBucket tableBucket = split.getTableBucket();
assignedTableBuckets.add(tableBucket);
+ unassignedSplits.remove(split);
if (isPartitioned) {
long partitionId =
@@ -1039,17 +1252,6 @@ public class FlinkSourceEnumerator
assignedPartitions.put(partitionId,
partitionName);
}
});
-
- if (pendingHybridLakeFlussSplits != null) {
- Set<String> splitIdsToRemove =
- pendingAssignmentForReader.stream()
- .map(SourceSplitBase::splitId)
- .collect(Collectors.toSet());
- // removed from the pendingHybridLakeFlussSplits
- // since this split already be assigned
- pendingHybridLakeFlussSplits.removeIf(
- split ->
splitIdsToRemove.contains(split.splitId()));
- }
}
}
@@ -1203,7 +1405,19 @@ public class FlinkSourceEnumerator
@Override
public void addSplitsBack(List<SourceSplitBase> splits, int subtaskId) {
- LOG.debug("Flink Source Enumerator adds splits back: {}", splits);
+ LOG.info(
+ "Adding {} splits back from failed reader {} for table {}: {}",
+ splits.size(),
+ subtaskId,
+ tablePath,
+ splits);
+ for (SourceSplitBase split : splits) {
+ unassignedSplits.add(split);
+ assignedTableBuckets.remove(split.getTableBucket());
+ if (isPartitioned) {
+
assignedPartitions.remove(split.getTableBucket().getPartitionId());
+ }
+ }
addSplitToPendingAssignments(splits);
// If the failed subtask has already restarted, we need to assign
pending splits to it
@@ -1214,7 +1428,14 @@ public class FlinkSourceEnumerator
@Override
public void addReader(int subtaskId) {
- LOG.debug("Adding reader: {} to Flink Source enumerator.", subtaskId);
+ LOG.info(
+ "Adding reader {} to FlinkSourceEnumerator for table {}, "
+ + "pendingSplitAssignment has {} splits for this
reader.",
+ subtaskId,
+ tablePath,
+ pendingSplitAssignment.containsKey(subtaskId)
+ ? pendingSplitAssignment.get(subtaskId).size()
+ : 0);
assignPendingSplits(Collections.singleton(subtaskId));
}
@@ -1225,8 +1446,21 @@ public class FlinkSourceEnumerator
assignedTableBuckets,
assignedPartitions,
pendingHybridLakeFlussSplits,
- leaseContext.getKvSnapshotLeaseId());
- LOG.debug("Source Checkpoint is {}", enumeratorState);
+ leaseContext.getKvSnapshotLeaseId(),
+ initialDiscoveryFinished,
+ unassignedSplits);
+ LOG.debug(
+ "Snapshot state for table {} at checkpoint {}: "
+ + "assignedBuckets={}, assignedPartitions={}, "
+ + "unassignedSplits={}, pendingLakeSplits={}, "
+ + "initialDiscoveryFinished={}",
+ tablePath,
+ checkpointId,
+ assignedTableBuckets.size(),
+ assignedPartitions.size(),
+ unassignedSplits.size(),
+ pendingHybridLakeFlussSplits == null ? "null" :
pendingHybridLakeFlussSplits.size(),
+ initialDiscoveryFinished);
return enumeratorState;
}
@@ -1293,6 +1527,14 @@ public class FlinkSourceEnumerator
@Override
public void close() throws IOException {
+ LOG.info(
+ "Closing FlinkSourceEnumerator for table {}: "
+ + "assignedBuckets={}, unassignedSplits={}, "
+ + "checkpointTriggeredBefore={}",
+ tablePath,
+ assignedTableBuckets.size(),
+ unassignedSplits.size(),
+ checkpointTriggeredBefore);
try {
maybeDropKvSnapshotLease();
@@ -1352,17 +1594,23 @@ public class FlinkSourceEnumerator
// --------------- private class ---------------
/** A container class to hold the newly added partitions and removed
partitions. */
private static class PartitionChange {
+ private final Collection<Partition> initialPartitions;
private final Collection<Partition> newPartitions;
private final Collection<Partition> removedPartitions;
PartitionChange(
- Collection<Partition> newPartitions, Collection<Partition>
removedPartitions) {
+ Collection<Partition> initialPartitions,
+ Collection<Partition> newPartitions,
+ Collection<Partition> removedPartitions) {
+ this.initialPartitions = initialPartitions;
this.newPartitions = newPartitions;
this.removedPartitions = removedPartitions;
}
public boolean isEmpty() {
- return newPartitions.isEmpty() && removedPartitions.isEmpty();
+ return initialPartitions.isEmpty()
+ && newPartitions.isEmpty()
+ && removedPartitions.isEmpty();
}
}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/FlussSourceEnumeratorStateSerializer.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/FlussSourceEnumeratorStateSerializer.java
index 996313e60..07aa1fb22 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/FlussSourceEnumeratorStateSerializer.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/FlussSourceEnumeratorStateSerializer.java
@@ -34,6 +34,7 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -53,14 +54,20 @@ import static
org.apache.fluss.utils.Preconditions.checkNotNull;
* <ul>
* <li><b>Version 0:</b> Initial version. Remaining hybrid lake splits are
only (de)serialized if
* the {@code lakeSource} is non-null.
- * <li><b>Version 1 (Current):</b> Decouples split serialization from the
{@code lakeSource}
- * presence. It always attempts to (de)serialize the splits, using an
internal boolean flag to
- * indicate presence. This ensures state consistency regardless of the
current runtime
- * configuration.
+ * <li><b>Version 1:</b> Decouples split serialization from the {@code
lakeSource} presence. It
+ * always attempts to (de)serialize the splits, using an internal
boolean flag to indicate
+ * presence. This ensures state consistency regardless of the current
runtime configuration.
+ * <li><b>Version 2:</b> Adds lease ID (UTF string) after the hybrid lake
splits section to
+ * support KV snapshot lease management.
+ * <li><b>Version 3 (Current):</b> Adds {@code initialDiscoveryFinished}
boolean flag after the
+ * lease ID, and appends {@code unassignedSplits} collection. The flag
tracks whether the
+ * first partition discovery round has completed (FLIP-288 offset
semantics); the unassigned
+ * splits are persisted so they survive failover without
re-initialization.
* </ul>
*
* <p><b>Compatibility Note:</b> This serializer is designed for backward
compatibility. It can
- * deserialize states from Version 0, but always produces Version 1 during
serialization.
+ * deserialize states from all previous versions, but always produces the
current version during
+ * serialization.
*/
public class FlussSourceEnumeratorStateSerializer
implements SimpleVersionedSerializer<SourceEnumeratorState> {
@@ -70,11 +77,12 @@ public class FlussSourceEnumeratorStateSerializer
private static final int VERSION_0 = 0;
private static final int VERSION_1 = 1;
private static final int VERSION_2 = 2;
+ private static final int VERSION_3 = 3;
private static final ThreadLocal<DataOutputSerializer> SERIALIZER_CACHE =
ThreadLocal.withInitial(() -> new DataOutputSerializer(64));
- private static final int CURRENT_VERSION = VERSION_2;
+ private static final int CURRENT_VERSION = VERSION_3;
public FlussSourceEnumeratorStateSerializer(LakeSource<LakeSplit>
lakeSource) {
this.lakeSource = lakeSource;
@@ -99,6 +107,12 @@ public class FlussSourceEnumeratorStateSerializer
// write lease context
serializeLeaseId(out, state);
+ // write initial discovery finished flag (VERSION_3+)
+ out.writeBoolean(state.isInitialDiscoveryFinished());
+
+ // write unassigned splits (VERSION_3+)
+ serializeUnassignedSplits(out, state.getUnassignedSplits());
+
final byte[] result = out.getCopyOfBuffer();
out.clear();
return result;
@@ -166,9 +180,24 @@ public class FlussSourceEnumeratorStateSerializer
return result;
}
+ /** Produces a real V2 payload: assignedBuckets + partitions +
hybridSplits + leaseId. */
+ @VisibleForTesting
+ protected byte[] serializeV2(SourceEnumeratorState state) throws
IOException {
+ final DataOutputSerializer out = SERIALIZER_CACHE.get();
+ serializeAssignBucketAndPartitions(
+ out, state.getAssignedBuckets(),
state.getAssignedPartitions());
+ serializeRemainingHybridLakeFlussSplits(out, state);
+ serializeLeaseId(out, state);
+ final byte[] result = out.getCopyOfBuffer();
+ out.clear();
+ return result;
+ }
+
@Override
public SourceEnumeratorState deserialize(int version, byte[] serialized)
throws IOException {
switch (version) {
+ case VERSION_3:
+ return deserializeV3(serialized);
case VERSION_2:
return deserializeV2(serialized);
case VERSION_1:
@@ -229,6 +258,8 @@ public class FlussSourceEnumeratorStateSerializer
// deserialize lease context
LeaseContext leaseContext = deserializeLeaseId(in);
+ // V2 does not have initialDiscoveryFinished flag; default to true
(safe choice
+ // to prevent data loss: treat all newly discovered partitions as
post-initial).
return new SourceEnumeratorState(
assignBucketAndPartitions.f0,
assignBucketAndPartitions.f1,
@@ -236,6 +267,31 @@ public class FlussSourceEnumeratorStateSerializer
leaseContext.getKvSnapshotLeaseId());
}
+ private SourceEnumeratorState deserializeV3(byte[] serialized) throws
IOException {
+ DataInputDeserializer in = new DataInputDeserializer(serialized);
+ Tuple2<Set<TableBucket>, Map<Long, String>> assignBucketAndPartitions =
+ deserializeAssignBucketAndPartitions(in);
+ List<SourceSplitBase> remainingHybridLakeFlussSplits =
+ deserializeRemainingHybridLakeFlussSplits(in);
+
+ // deserialize lease context
+ LeaseContext leaseContext = deserializeLeaseId(in);
+
+ // deserialize initial discovery finished flag
+ boolean initialDiscoveryFinished = in.readBoolean();
+
+ // deserialize unassigned splits
+ List<SourceSplitBase> unassignedSplits =
deserializeUnassignedSplits(in);
+
+ return new SourceEnumeratorState(
+ assignBucketAndPartitions.f0,
+ assignBucketAndPartitions.f1,
+ remainingHybridLakeFlussSplits,
+ leaseContext.getKvSnapshotLeaseId(),
+ initialDiscoveryFinished,
+ unassignedSplits);
+ }
+
private Tuple2<Set<TableBucket>, Map<Long, String>>
deserializeAssignBucketAndPartitions(
DataInputDeserializer in) throws IOException {
// deserialize assigned buckets
@@ -302,4 +358,37 @@ public class FlussSourceEnumeratorStateSerializer
return new LeaseContext(
kvSnapshotLeaseId,
LeaseContext.DEFAULT.getKvSnapshotLeaseDurationMs());
}
+
+ private void serializeUnassignedSplits(
+ final DataOutputSerializer out, Collection<SourceSplitBase>
unassignedSplits)
+ throws IOException {
+ out.writeInt(unassignedSplits.size());
+ if (!unassignedSplits.isEmpty()) {
+ SourceSplitSerializer sourceSplitSerializer = new
SourceSplitSerializer(lakeSource);
+ out.writeInt(sourceSplitSerializer.getVersion());
+ for (SourceSplitBase split : unassignedSplits) {
+ byte[] serializeBytes = sourceSplitSerializer.serialize(split);
+ out.writeInt(serializeBytes.length);
+ out.write(serializeBytes);
+ }
+ }
+ }
+
+ private List<SourceSplitBase> deserializeUnassignedSplits(final
DataInputDeserializer in)
+ throws IOException {
+ int numSplits = in.readInt();
+ if (numSplits == 0) {
+ return new ArrayList<>();
+ }
+ SourceSplitSerializer sourceSplitSerializer = new
SourceSplitSerializer(lakeSource);
+ int version = in.readInt();
+ List<SourceSplitBase> splits = new ArrayList<>(numSplits);
+ for (int i = 0; i < numSplits; i++) {
+ int splitSizeInBytes = in.readInt();
+ byte[] splitBytes = new byte[splitSizeInBytes];
+ in.readFully(splitBytes);
+ splits.add(sourceSplitSerializer.deserialize(version, splitBytes));
+ }
+ return splits;
+ }
}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/SourceEnumeratorState.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/SourceEnumeratorState.java
index 7ee021ff5..654b8c8df 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/SourceEnumeratorState.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/SourceEnumeratorState.java
@@ -22,6 +22,8 @@ import org.apache.fluss.metadata.TableBucket;
import javax.annotation.Nullable;
+import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -44,15 +46,54 @@ public class SourceEnumeratorState {
// lease context for restore.
private final String leaseId;
+ /** Whether the initial partition discovery has been completed. */
+ private final boolean initialDiscoveryFinished;
+
+ /** Splits that have been initialized (offsets resolved) but not yet
assigned to readers. */
+ 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,
+ true,
+ Collections.emptyList());
+ }
+
+ public SourceEnumeratorState(
+ Set<TableBucket> assignedBuckets,
+ Map<Long, String> assignedPartitions,
+ @Nullable List<SourceSplitBase> remainingHybridLakeFlussSplits,
+ String leaseId,
+ boolean initialDiscoveryFinished) {
+ this(
+ assignedBuckets,
+ assignedPartitions,
+ remainingHybridLakeFlussSplits,
+ leaseId,
+ initialDiscoveryFinished,
+ Collections.emptyList());
+ }
+
+ public SourceEnumeratorState(
+ Set<TableBucket> assignedBuckets,
+ Map<Long, String> assignedPartitions,
+ @Nullable List<SourceSplitBase> remainingHybridLakeFlussSplits,
+ String leaseId,
+ boolean initialDiscoveryFinished,
+ Collection<SourceSplitBase> unassignedSplits) {
this.assignedBuckets = assignedBuckets;
this.assignedPartitions = assignedPartitions;
this.remainingHybridLakeFlussSplits = remainingHybridLakeFlussSplits;
this.leaseId = leaseId;
+ this.initialDiscoveryFinished = initialDiscoveryFinished;
+ this.unassignedSplits = unassignedSplits;
}
public Set<TableBucket> getAssignedBuckets() {
@@ -72,6 +113,14 @@ public class SourceEnumeratorState {
return leaseId;
}
+ public boolean isInitialDiscoveryFinished() {
+ return initialDiscoveryFinished;
+ }
+
+ public Collection<SourceSplitBase> getUnassignedSplits() {
+ return unassignedSplits;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -85,12 +134,19 @@ public class SourceEnumeratorState {
&& Objects.equals(assignedPartitions, that.assignedPartitions)
&& Objects.equals(
remainingHybridLakeFlussSplits,
that.remainingHybridLakeFlussSplits)
- && Objects.equals(leaseId, that.leaseId);
+ && Objects.equals(leaseId, that.leaseId)
+ && initialDiscoveryFinished == that.initialDiscoveryFinished
+ && Objects.equals(unassignedSplits, that.unassignedSplits);
}
@Override
public int hashCode() {
- return Objects.hash(assignedBuckets, assignedPartitions,
remainingHybridLakeFlussSplits);
+ return Objects.hash(
+ assignedBuckets,
+ assignedPartitions,
+ remainingHybridLakeFlussSplits,
+ initialDiscoveryFinished,
+ unassignedSplits);
}
@Override
@@ -104,6 +160,10 @@ public class SourceEnumeratorState {
+ remainingHybridLakeFlussSplits
+ ", leaseId="
+ leaseId
+ + ", initialDiscoveryFinished="
+ + initialDiscoveryFinished
+ + ", unassignedSplits="
+ + unassignedSplits
+ '}';
}
}
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
index 184e1b6e7..3f90a4b1b 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
@@ -556,7 +556,9 @@ class FlinkSourceEnumeratorTest extends FlinkTestBase {
null,
null,
LeaseContext.DEFAULT,
- true);
+ true,
+ true,
+ Collections.emptyList());
enumerator.start();
assertThat(context.getSplitsAssignmentSequence()).isEmpty();
@@ -577,6 +579,210 @@ class FlinkSourceEnumeratorTest extends FlinkTestBase {
}
}
+ @Test
+ void testNewPartitionsUseEarliestOffset() throws Throwable {
+ int numSubtasks = 3;
+ createTable(DEFAULT_TABLE_PATH,
DEFAULT_AUTO_PARTITIONED_LOG_TABLE_DESCRIPTOR);
+ ZooKeeperClient zooKeeperClient =
FLUSS_CLUSTER_EXTENSION.getZooKeeperClient();
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(numSubtasks);
+ MockWorkExecutor workExecutor = new MockWorkExecutor(context);
+ FlinkSourceEnumerator enumerator =
+ new FlinkSourceEnumerator(
+ DEFAULT_TABLE_PATH,
+ flussConf,
+ false,
+ true,
+ context,
+ Collections.emptySet(),
+ Collections.emptyMap(),
+ null,
+ OffsetsInitializer.latest(),
+ DEFAULT_SCAN_PARTITION_DISCOVERY_INTERVAL_MS,
+ streaming,
+ null,
+ null,
+ workExecutor,
+ LeaseContext.DEFAULT,
+ false)) {
+
+ Map<Long, String> partitionNameByIds =
+ waitUntilPartitions(zooKeeperClient, DEFAULT_TABLE_PATH);
+ enumerator.start();
+
+ // register readers
+ for (int i = 0; i < numSubtasks; i++) {
+ registerReader(context, enumerator, i);
+ }
+
+ // First partition discovery: initial partitions should use latest
offset
+ runPeriodicPartitionDiscovery(workExecutor);
+ runPeriodicPartitionDiscovery(workExecutor);
+
+ // Snapshot state - initialDiscoveryFinished should be true
+ SourceEnumeratorState state = enumerator.snapshotState(1L);
+ assertThat(state.isInitialDiscoveryFinished()).isTrue();
+
+ // Verify initial partitions got latest offset (offset >= 0, not
EARLIEST_OFFSET)
+ List<SourceSplitBase> initialAssignedSplits =
+ getReadersAssignments(context).values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+ for (SourceSplitBase split : initialAssignedSplits) {
+ assertThat(split).isInstanceOf(LogSplit.class);
+ LogSplit logSplit = (LogSplit) split;
+ // latest offset for empty partition should be 0
+ assertThat(logSplit.getStartingOffset())
+ .as("Initial partitions should use latest offset
(>=0), not earliest (-2)")
+ .isGreaterThanOrEqualTo(0L);
+ }
+
+ // Create new partitions
+ List<String> newPartitions = Arrays.asList("newPartition1",
"newPartition2");
+ createPartitions(zooKeeperClient, DEFAULT_TABLE_PATH,
newPartitions);
+
+ // Second partition discovery: new partitions should use earliest
offset
+ int assignmentStart = context.getSplitsAssignmentSequence().size();
+ runPeriodicPartitionDiscovery(workExecutor);
+
+ List<SourceSplitBase> newAssignedSplits =
+ getReadersAssignments(context,
assignmentStart).values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+ assertThat(newAssignedSplits).isNotEmpty();
+ for (SourceSplitBase split : newAssignedSplits) {
+ assertThat(split).isInstanceOf(LogSplit.class);
+ LogSplit logSplit = (LogSplit) split;
+ assertThat(logSplit.getStartingOffset())
+ .as("Newly discovered partitions should use earliest
offset")
+ .isEqualTo(EARLIEST_OFFSET);
+ }
+
+ // Verify snapshot state is consistent after new partitions
+ SourceEnumeratorState state2 = enumerator.snapshotState(2L);
+ assertThat(state2.isInitialDiscoveryFinished()).isTrue();
+ }
+ }
+
+ /**
+ * Tests FLIP-288 restore: after restore with
initialDiscoveryFinished=true, unassigned splits
+ * from state preserve their original offset (latest), while newly
discovered partitions still
+ * use earliest offset.
+ */
+ @Test
+ void testRestorePreservesInitialDiscoveryFinished() throws Throwable {
+ int numSubtasks = 3;
+ long tableId =
+ createTable(DEFAULT_TABLE_PATH,
DEFAULT_AUTO_PARTITIONED_LOG_TABLE_DESCRIPTOR);
+ ZooKeeperClient zooKeeperClient =
FLUSS_CLUSTER_EXTENSION.getZooKeeperClient();
+ Map<Long, String> partitions = waitUntilPartitions(zooKeeperClient,
DEFAULT_TABLE_PATH);
+
+ // Pick one partition to simulate as an unassigned split from previous
checkpoint.
+ // This split was initialized with "latest" offset (resolved to 0 for
empty partition)
+ // before the failover, but had not been assigned to any reader yet.
+ Map.Entry<Long, String> unassignedPartitionEntry =
partitions.entrySet().iterator().next();
+ long unassignedPartitionId = unassignedPartitionEntry.getKey();
+ String unassignedPartitionName = unassignedPartitionEntry.getValue();
+ long latestResolvedOffset = 0L;
+ LogSplit unassignedSplit =
+ new LogSplit(
+ new TableBucket(tableId, unassignedPartitionId, 0),
+ unassignedPartitionName,
+ latestResolvedOffset);
+
+ // Simulate restore: initialDiscoveryFinished=true, one unassigned
split from state
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(numSubtasks);
+ MockWorkExecutor workExecutor = new MockWorkExecutor(context);
+ FlinkSourceEnumerator enumerator =
+ new FlinkSourceEnumerator(
+ DEFAULT_TABLE_PATH,
+ flussConf,
+ false,
+ true,
+ context,
+ Collections.emptySet(),
+ Collections.emptyMap(),
+ null,
+ OffsetsInitializer.latest(),
+ DEFAULT_SCAN_PARTITION_DISCOVERY_INTERVAL_MS,
+ Integer.MAX_VALUE,
+ streaming,
+ null,
+ null,
+ workExecutor,
+ LeaseContext.DEFAULT,
+ false,
+ true,
+ Collections.singletonList(unassignedSplit))) {
+
+ enumerator.start();
+
+ // register readers
+ for (int i = 0; i < numSubtasks; i++) {
+ registerReader(context, enumerator, i);
+ }
+
+ // Partition discovery after restore: since
initialDiscoveryFinished=true,
+ // newly discovered partitions should use earliest offset.
+ // The unassigned split's partition should be filtered out
(already in pending).
+ runPeriodicPartitionDiscovery(workExecutor);
+
+ // Collect all assigned splits across all readers
+ List<SourceSplitBase> assignedSplits =
+ getReadersAssignments(context).values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+
+ // Verify the restored unassigned split preserves its original
latest offset
+ List<SourceSplitBase> restoredSplits =
+ assignedSplits.stream()
+ .filter(
+ s ->
+ s.getTableBucket()
+ .equals(
+ new TableBucket(
+ tableId,
+
unassignedPartitionId,
+ 0)))
+ .collect(Collectors.toList());
+ assertThat(restoredSplits).hasSize(1);
+ LogSplit restoredLogSplit = (LogSplit) restoredSplits.get(0);
+ assertThat(restoredLogSplit.getStartingOffset())
+ .as(
+ "Unassigned split from state should preserve its
original "
+ + "latest offset, not be re-initialized")
+ .isEqualTo(latestResolvedOffset);
+
+ // Verify other partitions (truly new discoveries) use earliest
offset
+ List<SourceSplitBase> newDiscoverySplits =
+ assignedSplits.stream()
+ .filter(
+ s ->
+ !s.getTableBucket()
+ .equals(
+ new TableBucket(
+ tableId,
+
unassignedPartitionId,
+ 0)))
+ .collect(Collectors.toList());
+ assertThat(newDiscoverySplits).isNotEmpty();
+ for (SourceSplitBase split : newDiscoverySplits) {
+ assertThat(split).isInstanceOf(LogSplit.class);
+ LogSplit logSplit = (LogSplit) split;
+ assertThat(logSplit.getStartingOffset())
+ .as(
+ "After restore with
initialDiscoveryFinished=true, "
+ + "newly discovered partitions should
use earliest offset")
+ .isEqualTo(EARLIEST_OFFSET);
+ }
+
+ // Verify state after restore + discovery
+ SourceEnumeratorState state = enumerator.snapshotState(3L);
+ assertThat(state.isInitialDiscoveryFinished()).isTrue();
+ }
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
void testDiscoverPartitionsPeriodically(boolean isPrimaryKeyTable) throws
Throwable {
@@ -948,6 +1154,122 @@ class FlinkSourceEnumeratorTest extends FlinkTestBase {
}
}
+ /**
+ * Regression test: verifies that when a partition is dropped after splits
are restored into
+ * {@code unassignedSplits}, the stale splits are removed from both {@code
unassignedSplits} and
+ * {@code pendingSplitAssignment}. Without this fix, stale splits would
survive in checkpoint
+ * state and be assigned to readers for a deleted partition after failover.
+ */
+ @Test
+ void testDroppedPartitionSplitsRemovedFromUnassignedSplits() throws
Throwable {
+ int numSubtasks = 3;
+ long tableId =
+ createTable(DEFAULT_TABLE_PATH,
DEFAULT_AUTO_PARTITIONED_LOG_TABLE_DESCRIPTOR);
+ ZooKeeperClient zooKeeperClient =
FLUSS_CLUSTER_EXTENSION.getZooKeeperClient();
+ Map<Long, String> partitions = waitUntilPartitions(zooKeeperClient,
DEFAULT_TABLE_PATH);
+
+ // Pick one partition to simulate as "dropped after checkpoint"
+ Map.Entry<Long, String> partitionToDrop =
partitions.entrySet().iterator().next();
+ long droppedPartitionId = partitionToDrop.getKey();
+ String droppedPartitionName = partitionToDrop.getValue();
+
+ // Mock unassigned splits as if they were restored from a previous
checkpoint.
+ // These splits include one for the partition that will be dropped.
+ List<SourceSplitBase> unassignedSplits = new ArrayList<>();
+ for (Map.Entry<Long, String> entry : partitions.entrySet()) {
+ for (int bucket = 0; bucket < DEFAULT_BUCKET_NUM; bucket++) {
+ unassignedSplits.add(
+ new LogSplit(
+ new TableBucket(tableId, entry.getKey(),
bucket),
+ entry.getValue(),
+ EARLIEST_OFFSET));
+ }
+ }
+
+ // Drop the partition before restoring the enumerator
+ dropPartitions(
+ zooKeeperClient, DEFAULT_TABLE_PATH,
Collections.singleton(droppedPartitionName));
+
+ // Restore enumerator with mock unassigned splits (simulating failover
recovery)
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(numSubtasks);
+ MockWorkExecutor workExecutor = new MockWorkExecutor(context);
+ FlinkSourceEnumerator enumerator =
+ new FlinkSourceEnumerator(
+ DEFAULT_TABLE_PATH,
+ flussConf,
+ false,
+ true,
+ context,
+ Collections.emptySet(),
+ Collections.emptyMap(),
+ null,
+ OffsetsInitializer.earliest(),
+ DEFAULT_SCAN_PARTITION_DISCOVERY_INTERVAL_MS,
+ Integer.MAX_VALUE,
+ streaming,
+ null,
+ null,
+ workExecutor,
+ LeaseContext.DEFAULT,
+ false,
+ true,
+ unassignedSplits)) {
+
+ enumerator.start();
+
+ // Partition discovery detects the dropped partition and calls
handlePartitionsRemoved
+ runPeriodicPartitionDiscovery(workExecutor);
+
+ // Verify unassignedSplits no longer contains splits for the
dropped partition
+ SourceEnumeratorState state = enumerator.snapshotState(1L);
+ List<SourceSplitBase> staleSplits =
+ state.getUnassignedSplits().stream()
+ .filter(
+ split ->
+ Objects.equals(
+
split.getTableBucket().getPartitionId(),
+ droppedPartitionId))
+ .collect(Collectors.toList());
+ assertThat(staleSplits)
+ .as(
+ "Splits for dropped partition should be removed
from "
+ + "unassignedSplits to prevent stale
assignment after failover")
+ .isEmpty();
+
+ // Verify pendingSplitAssignment is also cleaned
+ long stalePendingSplits =
+ enumerator.getPendingSplitAssignment().values().stream()
+ .flatMap(List::stream)
+ .filter(
+ split ->
+ Objects.equals(
+
split.getTableBucket().getPartitionId(),
+ droppedPartitionId))
+ .count();
+ assertThat(stalePendingSplits)
+ .as(
+ "Splits for dropped partition should be removed
from pendingSplitAssignment")
+ .isZero();
+
+ // Register readers and verify no stale splits are assigned
+ for (int i = 0; i < numSubtasks; i++) {
+ registerReader(context, enumerator, i);
+ }
+ List<SourceSplitBase> allAssigned =
+ getReadersAssignments(context).values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+ assertThat(allAssigned)
+ .as("No assigned split should belong to the dropped
partition")
+ .noneMatch(
+ split ->
+ Objects.equals(
+
split.getTableBucket().getPartitionId(),
+ droppedPartitionId));
+ }
+ }
+
// ---------------------
private void registerReader(
MockSplitEnumeratorContext<SourceSplitBase> context,
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java
index fa90eda96..7394eddea 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java
@@ -144,6 +144,32 @@ class SourceEnumeratorStateSerializerTest {
assertThat(deserializedSourceEnumeratorState).isEqualTo(sourceEnumeratorState);
}
+ @Test
+ void testV2CompatibilityDefaultsInitialDiscoveryFinished() throws
Exception {
+ // Serialize a state using the real V2 format (without
initialDiscoveryFinished
+ // and unassignedSplits) and verify that deserialization defaults to
safe values.
+ FlussSourceEnumeratorStateSerializer serializer =
+ new FlussSourceEnumeratorStateSerializer(null);
+
+ Set<TableBucket> assignedBuckets =
+ new HashSet<>(Arrays.asList(new TableBucket(1, 0), new
TableBucket(1, 2L, 1)));
+ Map<Long, String> assignedPartitions = new HashMap<>();
+ assignedPartitions.put(2L, "partition2");
+
+ SourceEnumeratorState originalState =
+ new SourceEnumeratorState(assignedBuckets, assignedPartitions,
null, "v2LeaseId");
+
+ // Use serializeV2 to produce a real V2 payload
+ byte[] v2Bytes = serializer.serializeV2(originalState);
+
+ SourceEnumeratorState deserialized = serializer.deserialize(2,
v2Bytes);
+
assertThat(deserialized.getAssignedBuckets()).isEqualTo(assignedBuckets);
+
assertThat(deserialized.getAssignedPartitions()).isEqualTo(assignedPartitions);
+ assertThat(deserialized.getLeaseId()).isEqualTo("v2LeaseId");
+ assertThat(deserialized.isInitialDiscoveryFinished()).isTrue();
+ assertThat(deserialized.getUnassignedSplits()).isEmpty();
+ }
+
@Test
void testInconsistentLakeSourceSerde() throws Exception {
// test serialize with null lake source
@@ -191,4 +217,81 @@ class SourceEnumeratorStateSerializerTest {
.isNotNull()
.isEmpty();
}
+
+ @Test
+ void testInitialDiscoveryFinished() throws Exception {
+ FlussSourceEnumeratorStateSerializer serializer =
+ new FlussSourceEnumeratorStateSerializer(null);
+
+ Set<TableBucket> assignedBuckets =
+ new HashSet<>(Arrays.asList(new TableBucket(1, 0), new
TableBucket(1, 4L, 1)));
+ Map<Long, String> assignedPartitions = new HashMap<>();
+ assignedPartitions.put(1L, "partition1");
+ assignedPartitions.put(2L, "partition2");
+
+ List<SourceSplitBase> unassignedSplits = new ArrayList<>();
+ unassignedSplits.add(new LogSplit(new TableBucket(1, 0), null, 100L));
+ unassignedSplits.add(new LogSplit(new TableBucket(1, 5L, 1), "p5",
200L));
+ unassignedSplits.add(new HybridSnapshotLogSplit(new TableBucket(1, 2),
null, 300L, 50L));
+
+ SourceEnumeratorState sourceEnumeratorState =
+ new SourceEnumeratorState(
+ assignedBuckets,
+ assignedPartitions,
+ null,
+ "testLeaseId",
+ true,
+ unassignedSplits);
+
+ byte[] serialized = serializer.serialize(sourceEnumeratorState);
+ SourceEnumeratorState deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
+ assertThat(deserialized).isEqualTo(sourceEnumeratorState);
+ assertThat(deserialized.isInitialDiscoveryFinished()).isTrue();
+
assertThat(deserialized.getUnassignedSplits()).containsExactlyElementsOf(unassignedSplits);
+ }
+
+ @Test
+ void testInitialDiscoveryNotFinished() throws Exception {
+ FlussSourceEnumeratorStateSerializer serializer =
+ new FlussSourceEnumeratorStateSerializer(null);
+
+ Set<TableBucket> assignedBuckets =
+ new HashSet<>(Collections.singletonList(new TableBucket(1, 5L,
0)));
+ Map<Long, String> assignedPartitions = new HashMap<>();
+ assignedPartitions.put(5L, "p5");
+
+ SourceEnumeratorState sourceEnumeratorState =
+ new SourceEnumeratorState(
+ assignedBuckets,
+ assignedPartitions,
+ Collections.emptyList(),
+ "leaseId2",
+ false);
+
+ byte[] serialized = serializer.serialize(sourceEnumeratorState);
+ SourceEnumeratorState deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
+ assertThat(deserialized).isEqualTo(sourceEnumeratorState);
+ assertThat(deserialized.isInitialDiscoveryFinished()).isFalse();
+ }
+
+ @Test
+ void testEmptyState() throws Exception {
+ FlussSourceEnumeratorStateSerializer serializer =
+ new FlussSourceEnumeratorStateSerializer(null);
+
+ SourceEnumeratorState sourceEnumeratorState =
+ new SourceEnumeratorState(
+ Collections.emptySet(), Collections.emptyMap(), null,
"leaseEmpty", true);
+
+ byte[] serialized = serializer.serialize(sourceEnumeratorState);
+ SourceEnumeratorState deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
+ assertThat(deserialized).isEqualTo(sourceEnumeratorState);
+ assertThat(deserialized.isInitialDiscoveryFinished()).isTrue();
+ }
}
