diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/state/WriterState.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/state/WriterState.java
index cc4f3088a34..fff74e0ff54 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/state/WriterState.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/state/WriterState.java
@@ -32,12 +32,30 @@
* writer is responsible for. During failover recovery, these offsets are used to generate undo logs
* and rollback to the checkpoint state.
*
+ * V2 state is a complete but sparse recovery baseline. It stores only positive offsets; a live
+ * bucket absent from the map has an explicit baseline of zero. The state element itself, including
+ * an element with an empty map, proves that the checkpoint was produced with this completeness
+ * guarantee.
+ *
*
This class uses {@link TableBucket} as the key to support partitioned tables. Each bucket is
* uniquely identified by its table ID, partition ID (if applicable), and bucket ID.
*/
public class WriterState implements Serializable {
private static final long serialVersionUID = 1L;
+ private static final long NO_TABLE_ID = -1L;
+
+ /** The semantic guarantees carried by the serialized writer state. */
+ public enum StateFormat {
+ /** Legacy state whose bucket map may be sparse. */
+ V1_LEGACY,
+
+ /** Complete state for the table identified by {@link WriterState#getTableId()}. */
+ V2_COMPLETE
+ }
+
+ private final StateFormat stateFormat;
+ private final long tableId;
/**
* Map from TableBucket to the last successfully written changelog offset.
@@ -50,13 +68,22 @@ public class WriterState implements Serializable {
*/
private final Map bucketOffsets;
+ /** Creates legacy V1 state whose bucket map must not be treated as complete. */
public WriterState(Map bucketOffsets) {
+ this(StateFormat.V1_LEGACY, NO_TABLE_ID, bucketOffsets);
+ }
+
+ private WriterState(
+ StateFormat stateFormat, long tableId, Map bucketOffsets) {
if (bucketOffsets == null) {
throw new IllegalArgumentException("bucketOffsets must not be null");
}
- // Validate no null or negative offsets
- // Note: offset=0 is valid and means the bucket was empty at checkpoint time
- // (no records have been written to this bucket yet)
+ if (stateFormat == null) {
+ throw new IllegalArgumentException("stateFormat must not be null");
+ }
+ if (stateFormat == StateFormat.V2_COMPLETE && tableId < 0) {
+ throw new IllegalArgumentException("tableId must not be negative: " + tableId);
+ }
for (Map.Entry entry : bucketOffsets.entrySet()) {
if (entry.getKey() == null) {
throw new IllegalArgumentException("TableBucket in bucketOffsets must not be null");
@@ -65,14 +92,47 @@ public WriterState(Map bucketOffsets) {
throw new IllegalArgumentException(
"Invalid offset for bucket " + entry.getKey() + ": " + entry.getValue());
}
+ if (stateFormat == StateFormat.V2_COMPLETE && entry.getKey().getTableId() != tableId) {
+ throw new IllegalArgumentException(
+ "Bucket table ID "
+ + entry.getKey().getTableId()
+ + " does not match complete state table ID "
+ + tableId);
+ }
}
+ this.stateFormat = stateFormat;
+ this.tableId = tableId;
this.bucketOffsets = Collections.unmodifiableMap(new HashMap<>(bucketOffsets));
}
+ /** Creates complete V2 state for the given table. */
+ public static WriterState complete(long tableId, Map bucketOffsets) {
+ return new WriterState(StateFormat.V2_COMPLETE, tableId, bucketOffsets);
+ }
+
+ /** Returns the state format and its completeness guarantee. */
+ public StateFormat getStateFormat() {
+ return stateFormat;
+ }
+
+ /**
+ * Returns the table ID carried by complete V2 state.
+ *
+ * @throws IllegalStateException if this is legacy state
+ */
+ public long getTableId() {
+ if (stateFormat != StateFormat.V2_COMPLETE) {
+ throw new IllegalStateException("Legacy writer state does not carry a table ID");
+ }
+ return tableId;
+ }
+
+ /** Returns the immutable bucket-to-offset map. */
public Map getBucketOffsets() {
return bucketOffsets;
}
+ /** Returns the offset for the bucket, or {@code null} when it is absent. */
public Long getOffsetForBucket(TableBucket tableBucket) {
return bucketOffsets.get(tableBucket);
}
@@ -95,16 +155,25 @@ public boolean equals(Object o) {
return false;
}
WriterState that = (WriterState) o;
- return Objects.equals(bucketOffsets, that.bucketOffsets);
+ return tableId == that.tableId
+ && stateFormat == that.stateFormat
+ && Objects.equals(bucketOffsets, that.bucketOffsets);
}
@Override
public int hashCode() {
- return Objects.hash(bucketOffsets);
+ return Objects.hash(stateFormat, tableId, bucketOffsets);
}
@Override
public String toString() {
- return "WriterState{" + "bucketOffsets=" + bucketOffsets + '}';
+ return "WriterState{"
+ + "stateFormat="
+ + stateFormat
+ + ", tableId="
+ + tableId
+ + ", bucketOffsets="
+ + bucketOffsets
+ + '}';
}
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/state/WriterStateSerializer.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/state/WriterStateSerializer.java
index 710ef52b219..2739a0c15e0 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/state/WriterStateSerializer.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/state/WriterStateSerializer.java
@@ -35,9 +35,10 @@
* This serializer extends {@link TypeSerializer} for use with Flink's {@code
* ListStateDescriptor} in Union List State.
*
- *
Serialization format:
+ *
V1 serialization format:
*
*
+ * - int: version (1)
*
- int: number of bucket offsets
*
- For each bucket offset:
*
*
*
- * This serializer uses a stable binary format that supports both partitioned and non-partitioned
- * tables via {@link TableBucket}.
+ *
V2 serialization format:
+ *
+ *
+ * - int: version (2)
+ *
- long: table ID
+ *
- int: number of bucket offsets
+ *
- For each bucket offset:
+ *
+ * - boolean: has partition ID
+ *
- long: partition ID (if has partition ID is true)
+ *
- int: bucket ID
+ *
- long: offset
+ *
+ *
*/
public class WriterStateSerializer extends TypeSerializer {
private static final long serialVersionUID = 1L;
- /** The current version of the serialization format. */
- private static final int CURRENT_VERSION = 1;
+ private static final int V1_VERSION = 1;
+ private static final int V2_VERSION = 2;
+
// -------------------------------------------------------------------------
// TypeSerializer methods
// -------------------------------------------------------------------------
@@ -99,39 +113,46 @@ public int getLength() {
@Override
public void serialize(WriterState record, DataOutputView target) throws IOException {
- target.writeInt(CURRENT_VERSION);
+ if (record.getStateFormat() == WriterState.StateFormat.V1_LEGACY) {
+ serializeV1(record, target);
+ } else if (record.getStateFormat() == WriterState.StateFormat.V2_COMPLETE) {
+ serializeV2(record, target);
+ } else {
+ throw new IOException("Unsupported writer state format: " + record.getStateFormat());
+ }
+ }
+
+ private void serializeV1(WriterState record, DataOutputView target) throws IOException {
+ target.writeInt(V1_VERSION);
Map bucketOffsets = record.getBucketOffsets();
target.writeInt(bucketOffsets.size());
for (Map.Entry entry : bucketOffsets.entrySet()) {
TableBucket bucket = entry.getKey();
target.writeLong(bucket.getTableId());
- target.writeBoolean(bucket.getPartitionId() != null);
- if (bucket.getPartitionId() != null) {
- target.writeLong(bucket.getPartitionId());
- }
- target.writeInt(bucket.getBucket());
- target.writeLong(entry.getValue());
+ writeBucketOffset(target, bucket, entry.getValue());
+ }
+ }
+
+ private void serializeV2(WriterState record, DataOutputView target) throws IOException {
+ target.writeInt(V2_VERSION);
+ target.writeLong(record.getTableId());
+ Map bucketOffsets = record.getBucketOffsets();
+ target.writeInt(bucketOffsets.size());
+ for (Map.Entry entry : bucketOffsets.entrySet()) {
+ writeBucketOffset(target, entry.getKey(), entry.getValue());
}
}
@Override
public WriterState deserialize(DataInputView source) throws IOException {
int version = source.readInt();
- if (version != CURRENT_VERSION) {
- throw new IOException(
- "Unsupported version: " + version + ". Expected version: " + CURRENT_VERSION);
- }
- int size = source.readInt();
- Map bucketOffsets = new HashMap<>(size);
- for (int i = 0; i < size; i++) {
- long tableId = source.readLong();
- boolean hasPartitionId = source.readBoolean();
- Long partitionId = hasPartitionId ? source.readLong() : null;
- int bucketId = source.readInt();
- long offset = source.readLong();
- bucketOffsets.put(new TableBucket(tableId, partitionId, bucketId), offset);
+ if (version == V1_VERSION) {
+ return deserializeV1(source);
+ } else if (version == V2_VERSION) {
+ return deserializeV2(source);
+ } else {
+ throw new IOException("Unsupported writer state version: " + version);
}
- return new WriterState(bucketOffsets);
}
@Override
@@ -143,28 +164,60 @@ public WriterState deserialize(WriterState reuse, DataInputView source) throws I
@Override
public void copy(DataInputView source, DataOutputView target) throws IOException {
int version = source.readInt();
- if (version != CURRENT_VERSION) {
- throw new IOException(
- "Unsupported version: " + version + ". Expected version: " + CURRENT_VERSION);
+ WriterState state;
+ if (version == V1_VERSION) {
+ state = deserializeV1(source);
+ } else if (version == V2_VERSION) {
+ state = deserializeV2(source);
+ } else {
+ throw new IOException("Unsupported writer state version: " + version);
}
- target.writeInt(version);
- // Copy bucket offsets size
+ serialize(state, target);
+ }
+
+ private WriterState deserializeV1(DataInputView source) throws IOException {
+ int size = source.readInt();
+ Map bucketOffsets = readBucketOffsets(source, size, null);
+ return new WriterState(bucketOffsets);
+ }
+
+ private WriterState deserializeV2(DataInputView source) throws IOException {
+ long tableId = source.readLong();
+ validateTableId(tableId);
int size = source.readInt();
- target.writeInt(size);
- // Copy each bucket offset entry
+ Map bucketOffsets = readBucketOffsets(source, size, tableId);
+ return WriterState.complete(tableId, bucketOffsets);
+ }
+
+ private static Map readBucketOffsets(
+ DataInputView source, int size, Long tableIdFromStateHeader) throws IOException {
+ Map bucketOffsets = new HashMap<>(size);
for (int i = 0; i < size; i++) {
- // Copy table ID
- target.writeLong(source.readLong());
- // Copy has partition ID flag and partition ID if present
+ long tableId =
+ tableIdFromStateHeader == null ? source.readLong() : tableIdFromStateHeader;
boolean hasPartitionId = source.readBoolean();
- target.writeBoolean(hasPartitionId);
- if (hasPartitionId) {
- target.writeLong(source.readLong());
- }
- // Copy bucket ID
- target.writeInt(source.readInt());
- // Copy offset
- target.writeLong(source.readLong());
+ Long partitionId = hasPartitionId ? source.readLong() : null;
+ int bucketId = source.readInt();
+ long offset = source.readLong();
+ TableBucket bucket = new TableBucket(tableId, partitionId, bucketId);
+ bucketOffsets.put(bucket, offset);
+ }
+ return bucketOffsets;
+ }
+
+ private static void writeBucketOffset(DataOutputView target, TableBucket bucket, long offset)
+ throws IOException {
+ target.writeBoolean(bucket.getPartitionId() != null);
+ if (bucket.getPartitionId() != null) {
+ target.writeLong(bucket.getPartitionId());
+ }
+ target.writeInt(bucket.getBucket());
+ target.writeLong(offset);
+ }
+
+ private static void validateTableId(long tableId) throws IOException {
+ if (tableId < 0) {
+ throw new IOException("Invalid complete writer state table ID: " + tableId);
}
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManager.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManager.java
index 947a06630b0..38e22957c81 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManager.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManager.java
@@ -42,23 +42,35 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
/**
* Manages recovery offset determination for undo recovery in aggregation tables.
*
+ * State semantics:
+ *
+ *
+ * - {@code null} means that no Flink checkpoint is being restored, so the producer offset
+ * snapshot provides the initial baseline.
+ *
- V2 state is a complete sparse baseline; an assigned live bucket absent from it has baseline
+ * zero.
+ *
- Legacy or empty restored state is rejected because it cannot prove the same completeness
+ * guarantee.
+ *
+ *
+ * The complete baseline is preserved independently from the Undo work set. Unchanged buckets
+ * must remain in the next checkpoint even though only buckets whose current offsets exceed their
+ * baselines require Undo.
+ *
*
Recovery flow:
*
*
- * - Get recovery offsets (from checkpoint or producer offsets)
- *
- Fetch partition info once and cache it (partitioned tables only), create TableBuckets for
- * partitions not in recovery offsets
- *
- Filter buckets by sharding to current subtask
- *
- Fetch current offsets for filtered buckets (one RPC per partition due to Admin API
- * limitation)
- *
- Filter buckets with changed offsets (recovery < current, new partition buckets use 0 as
- * original offset)
- *
- Return recovery decision
+ *
- Classify Flink state and reject legacy V1 state before external reads.
+ *
- Load the current partition metadata through the existing Admin API.
+ *
- Merge state fragments without guessing and enumerate every assigned live bucket.
+ *
- Resolve each bucket's baseline and fetch its strict current log end offset.
+ *
- Return the complete non-zero baseline separately from the bounded Undo work set.
*
*/
public class RecoveryOffsetManager {
@@ -92,16 +104,22 @@ public enum RecoveryStrategy {
PRODUCER_OFFSET_RECOVERY
}
+ private enum RecoveryStateKind {
+ NO_FLINK_STATE,
+ V1_LEGACY,
+ V2_COMPLETE
+ }
+
/** Result of recovery strategy determination. */
public static class RecoveryDecision {
private final RecoveryStrategy strategy;
- @Nullable private final Map recoveryOffsets;
- @Nullable private final Map undoOffsets;
+ private final Map recoveryOffsets;
+ private final Map undoOffsets;
private RecoveryDecision(
RecoveryStrategy strategy,
- @Nullable Map recoveryOffsets,
- @Nullable Map undoOffsets) {
+ Map recoveryOffsets,
+ Map undoOffsets) {
this.strategy = strategy;
this.recoveryOffsets = recoveryOffsets;
this.undoOffsets = undoOffsets;
@@ -111,7 +129,7 @@ public RecoveryStrategy getStrategy() {
return strategy;
}
- @Nullable
+ /** Returns all non-zero recovery offsets that the next checkpoint must preserve. */
public Map getRecoveryOffsets() {
return recoveryOffsets;
}
@@ -122,30 +140,28 @@ public Map getRecoveryOffsets() {
* This is used by UndoRecoveryManager to perform undo recovery without needing to call
* listOffset again.
*
- * @return map of bucket to UndoOffsets, or null if no recovery needed
+ * @return map of bucket to UndoOffsets, empty if no recovery is needed
*/
- @Nullable
public Map getUndoOffsets() {
return undoOffsets;
}
public boolean needsUndoRecovery() {
- return strategy != RecoveryStrategy.FRESH_START
- && undoOffsets != null
- && !undoOffsets.isEmpty();
+ return !undoOffsets.isEmpty();
}
static RecoveryDecision of(
RecoveryStrategy strategy,
- @Nullable Map recoveryOffsets,
- @Nullable Map undoOffsets) {
+ Map recoveryOffsets,
+ Map undoOffsets) {
return new RecoveryDecision(strategy, recoveryOffsets, undoOffsets);
}
@Override
public String toString() {
- int size = undoOffsets != null ? undoOffsets.size() : 0;
- return String.format("RecoveryDecision{strategy=%s, buckets=%d}", strategy, size);
+ return String.format(
+ "RecoveryDecision{strategy=%s, recoveryBuckets=%d, undoBuckets=%d}",
+ strategy, recoveryOffsets.size(), undoOffsets.size());
}
}
@@ -206,104 +222,95 @@ public RecoveryDecision determineRecoveryStrategy(
parallelism,
producerId);
- // Step 1: Get recovery offsets (checkpoint or producer offsets)
- // Note: recoveredState == null means no checkpoint exists (fresh start or pre-checkpoint
- // failure)
- // recoveredState != null but empty means checkpoint exists but no data was written
- // OR the checkpoint was taken before UndoRecoveryOperator was added to the topology
- // In both cases, we should use producer offsets for recovery decision
- boolean hasCheckpoint = recoveredState != null && !recoveredState.isEmpty();
+ RecoveryStateKind stateKind = classifyRecoveredState(recoveredState);
+ Map partitionNames = getPartitionNameMap();
Map recoveryOffsets =
- hasCheckpoint ? mergeCheckpointState(recoveredState) : getProducerOffsets();
-
- // Validate that checkpoint state refers to the same table (detect table re-creation)
- if (hasCheckpoint) {
- validateTableId(recoveryOffsets);
- }
+ stateKind == RecoveryStateKind.NO_FLINK_STATE
+ ? getProducerOffsets()
+ : mergeCheckpointState(recoveredState, partitionNames);
LOG.info(
"Recovery offsets for subtask {} (source={}): {}",
subtaskIndex,
- hasCheckpoint ? "checkpoint" : "producer",
+ stateKind,
recoveryOffsets);
- // Step 2: Get all buckets (to ensure no bucket is missed during listOffset)
Set allBuckets = getAllBuckets();
-
- // Step 3: Filter by sharding (both allBuckets and recoveryOffsets)
- Map partitionNames = getPartitionNameMap();
Set filteredBuckets = filterBucketsBySharding(allBuckets, partitionNames);
- Map filteredRecoveryOffsets =
- filterRecoveryOffsetsBySharding(recoveryOffsets, partitionNames);
LOG.info(
- "Subtask {}: filteredBuckets={}, filteredRecoveryOffsets={}",
+ "Subtask {}: filteredBuckets={}, recoveryOffsets={}",
subtaskIndex,
filteredBuckets,
- filteredRecoveryOffsets);
-
- if (filteredBuckets.isEmpty()) {
- LOG.info("No buckets assigned to subtask {} after filtering", subtaskIndex);
- return RecoveryDecision.of(RecoveryStrategy.FRESH_START, null, null);
- }
+ recoveryOffsets);
- // Step 4: Fetch current offsets for all filtered buckets
Map currentOffsets =
fetchCurrentOffsets(filteredBuckets, partitionNames);
LOG.info("Subtask {}: currentOffsets={}", subtaskIndex, currentOffsets);
- // Step 5: Filter changed buckets and build UndoOffsets in one pass
- // For buckets not in filteredRecoveryOffsets, use 0 as recovery offset
- Map changedOffsets = new HashMap<>();
+ Map retainedRecoveryOffsets = new HashMap<>();
Map undoOffsets = new HashMap<>();
+ List legacyStateGaps = new ArrayList<>();
for (TableBucket bucket : filteredBuckets) {
- long recovery = filteredRecoveryOffsets.getOrDefault(bucket, 0L);
- long current = currentOffsets.getOrDefault(bucket, 0L);
- boolean inRecoveryOffsets = filteredRecoveryOffsets.containsKey(bucket);
+ Long sourceOffset = recoveryOffsets.get(bucket);
+ long baseline = sourceOffset == null ? 0L : sourceOffset;
+ Long currentOffset = currentOffsets.get(bucket);
+ if (currentOffset == null) {
+ throw new IllegalStateException("missing latest offset for live bucket " + bucket);
+ }
+ long current = currentOffset;
+
+ if (stateKind == RecoveryStateKind.V1_LEGACY && sourceOffset == null && current > 0) {
+ legacyStateGaps.add(bucket);
+ }
LOG.info(
- "Subtask {}: bucket={}, recovery={} (inCheckpoint={}), current={}",
+ "Subtask {}: bucket={}, baseline={} (explicit={}), current={}",
subtaskIndex,
bucket,
- recovery,
- inRecoveryOffsets,
+ baseline,
+ sourceOffset != null,
current);
- if (recovery > current) {
+ if (baseline > current) {
throw new IllegalStateException(
String.format(
- "Data inconsistency: bucket %s recovery=%d > current=%d",
- bucket, recovery, current));
+ "Data inconsistency: bucket %s baseline=%d > current=%d",
+ bucket, baseline, current));
}
- if (recovery < current) {
- changedOffsets.put(bucket, recovery);
- // Build UndoOffsets with checkpointOffset and logEndOffset (current offset)
- undoOffsets.put(bucket, new UndoOffsets(recovery, current));
+ if (baseline > 0) {
+ retainedRecoveryOffsets.put(bucket, baseline);
+ }
+ if (baseline < current) {
+ undoOffsets.put(bucket, new UndoOffsets(baseline, current));
}
- // recovery == current: no change, skip
}
- // Only return FRESH_START when changedOffsets is empty (after all checks)
- if (changedOffsets.isEmpty()) {
- LOG.info(
- "No buckets with changed offsets, fresh start (hasCheckpointState={})",
- hasCheckpoint);
- return RecoveryDecision.of(RecoveryStrategy.FRESH_START, null, null);
+ if (!legacyStateGaps.isEmpty()) {
+ LOG.warn(
+ "Restoring legacy V1 Undo Recovery state with {} assigned live buckets "
+ + "missing from the checkpoint state. V1 cannot distinguish buckets "
+ + "that had offset zero at checkpoint time from buckets whose state "
+ + "was previously lost. Missing buckets use recovery offset zero, "
+ + "so Undo Recovery may scan excessive history.",
+ legacyStateGaps.size());
+ LOG.debug("Legacy V1 state gaps for subtask {}: {}", subtaskIndex, legacyStateGaps);
}
- // Step 6: Return decision with both recoveryOffsets and undoOffsets
RecoveryStrategy strategy =
- hasCheckpoint
- ? RecoveryStrategy.CHECKPOINT_RECOVERY
- : RecoveryStrategy.PRODUCER_OFFSET_RECOVERY;
+ undoOffsets.isEmpty()
+ ? RecoveryStrategy.FRESH_START
+ : stateKind == RecoveryStateKind.NO_FLINK_STATE
+ ? RecoveryStrategy.PRODUCER_OFFSET_RECOVERY
+ : RecoveryStrategy.CHECKPOINT_RECOVERY;
LOG.info(
"{}: {} buckets need recovery for subtask {}",
strategy,
- changedOffsets.size(),
+ undoOffsets.size(),
subtaskIndex);
- return RecoveryDecision.of(strategy, changedOffsets, undoOffsets);
+ return RecoveryDecision.of(strategy, retainedRecoveryOffsets, undoOffsets);
}
/** Cleans up registered producer offsets. Should only be called by Task0. */
@@ -321,34 +328,102 @@ public void cleanupOffsets() {
// ==================== Step 1: Get Recovery Offsets ====================
- private Map mergeCheckpointState(Collection states) {
+ private RecoveryStateKind classifyRecoveredState(
+ @Nullable Collection recoveredState) {
+ if (recoveredState == null) {
+ return RecoveryStateKind.NO_FLINK_STATE;
+ }
+ if (recoveredState.isEmpty()) {
+ // A checkpoint produced by V2 always contains at least one state element, even when
+ // its baseline is empty. An empty restored collection therefore cannot prove a
+ // complete baseline.
+ throw new IllegalStateException(
+ "The job was restored but Undo Recovery has no state fragments. "
+ + "Cannot distinguish a legacy empty state from a topology that did "
+ + "not contain Undo Recovery; perform a controlled stateless restart.");
+ }
+
+ WriterState.StateFormat stateFormat = null;
+ for (WriterState state : recoveredState) {
+ if (state == null) {
+ throw new IllegalStateException("Undo Recovery state contains a null fragment.");
+ }
+ if (stateFormat != null && state.getStateFormat() != stateFormat) {
+ throw new IllegalStateException(
+ "Undo Recovery state contains mixed V1 and V2 fragments.");
+ }
+ stateFormat = state.getStateFormat();
+ }
+ if (stateFormat == WriterState.StateFormat.V1_LEGACY) {
+ return RecoveryStateKind.V1_LEGACY;
+ }
+ return RecoveryStateKind.V2_COMPLETE;
+ }
+
+ private Map mergeCheckpointState(
+ Collection states, Map partitionNames) {
Map merged = new HashMap<>();
for (WriterState state : states) {
- state.getBucketOffsets()
- .forEach((bucket, offset) -> merged.merge(bucket, offset, Math::max));
+ if (state.getStateFormat() == WriterState.StateFormat.V2_COMPLETE
+ && state.getTableId() != tableId) {
+ throw new IllegalStateException(
+ String.format(
+ "V2 state table ID %d does not match current table ID %d for %s.",
+ state.getTableId(), tableId, tablePath));
+ }
+ for (Map.Entry entry : state.getBucketOffsets().entrySet()) {
+ TableBucket bucket = entry.getKey();
+ validateTableId(bucket);
+ validateBaselineOffset(bucket, entry.getValue());
+ if (!isLiveStateBucket(bucket, partitionNames)) {
+ continue;
+ }
+ putMergedOffset(merged, bucket, entry.getValue());
+ }
}
return merged;
}
- /**
- * Validates that all buckets in the recovery offsets belong to the current table.
- *
- * If the table was dropped and re-created, the checkpoint state will contain buckets with
- * the old table ID, which won't match the current table ID. In this case, restoring from the
- * checkpoint is not safe and should fail explicitly.
- *
- * @param recoveryOffsets the merged recovery offsets from checkpoint state
- * @throws IllegalStateException if any bucket has a mismatched table ID
- */
- private void validateTableId(Map recoveryOffsets) {
- for (TableBucket bucket : recoveryOffsets.keySet()) {
- if (bucket.getTableId() != tableId) {
+ private void putMergedOffset(Map merged, TableBucket bucket, Long offset) {
+ Long previous = merged.putIfAbsent(bucket, offset);
+ if (previous != null && !previous.equals(offset)) {
+ throw new IllegalStateException(
+ String.format(
+ "Conflicting checkpoint offsets for %s: %d and %d.",
+ bucket, previous, offset));
+ }
+ }
+
+ private void validateTableId(TableBucket bucket) {
+ if (bucket.getTableId() != tableId) {
+ throw new IllegalStateException(
+ String.format(
+ "Table '%s' has been re-created (state tableId=%d, current tableId=%d). "
+ + "Cannot restore from checkpoint/savepoint after table re-creation.",
+ tablePath, bucket.getTableId(), tableId));
+ }
+ }
+
+ private boolean isLiveStateBucket(TableBucket bucket, Map partitionNames) {
+ Long partitionId = bucket.getPartitionId();
+ if (isPartitioned) {
+ if (partitionId == null) {
throw new IllegalStateException(
- String.format(
- "Table '%s' has been re-created (state tableId=%d, current tableId=%d). "
- + "Cannot restore from checkpoint/savepoint after table re-creation.",
- tablePath, bucket.getTableId(), tableId));
+ "State bucket " + bucket + " has no partition ID for a partitioned table.");
}
+ return partitionNames.containsKey(partitionId);
+ }
+ if (partitionId != null) {
+ throw new IllegalStateException(
+ "State bucket " + bucket + " has a partition ID for a non-partitioned table.");
+ }
+ return true;
+ }
+
+ private void validateBaselineOffset(TableBucket bucket, Long offset) {
+ if (offset == null || offset < 0) {
+ throw new IllegalStateException(
+ "Invalid checkpoint baseline offset for " + bucket + ": " + offset);
}
}
@@ -439,17 +514,6 @@ private Set filterBucketsBySharding(
return filtered;
}
- private Map filterRecoveryOffsetsBySharding(
- Map recoveryOffsets, Map partitionNames) {
- Map filtered = new HashMap<>();
- for (Map.Entry entry : recoveryOffsets.entrySet()) {
- if (isAssignedToSubtask(entry.getKey(), partitionNames)) {
- filtered.put(entry.getKey(), entry.getValue());
- }
- }
- return filtered;
- }
-
/**
* Determines if a bucket is assigned to the current subtask.
*
@@ -612,7 +676,21 @@ private ListOffsetsResult listOffsets(@Nullable String partitionName, List bucketResult = result.bucketResult(bucketId);
+ if (bucketResult == null) {
+ throw new IllegalStateException("missing latest offset for bucket ID " + bucketId);
+ }
+ Long offset = bucketResult.get();
+ if (offset == null) {
+ throw new IllegalStateException("null latest offset for bucket ID " + bucketId);
+ }
+ if (offset < 0) {
+ throw new IllegalStateException(
+ "negative latest offset for bucket ID " + bucketId + ": " + offset);
+ }
+ return offset;
}
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/UndoRecoveryOperator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/UndoRecoveryOperator.java
index 8e09fb27acd..6b3154bc467 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/UndoRecoveryOperator.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/undo/UndoRecoveryOperator.java
@@ -27,6 +27,7 @@
import org.apache.fluss.flink.sink.state.WriterStateSerializer;
import org.apache.fluss.flink.sink.undo.UndoRecoveryManager.UndoOffsets;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.types.RowType;
@@ -152,6 +153,9 @@ public class UndoRecoveryOperator extends AbstractStreamOperator
/** Union List State for storing bucket offsets across checkpoints. */
private transient ListState undoStateList;
+ /** Table ID whose complete baseline is held in {@link #bucketOffsets}. */
+ @Nullable private transient Long resolvedTableId;
+
/**
* Map from TableBucket to the latest written offset.
*
@@ -259,6 +263,7 @@ public void initializeState(StateInitializationContext context) throws Exception
subtaskIndex = RuntimeContextAdapter.getIndexOfThisSubtask(runtimeContext);
producerOffsetsDeleted = false;
restoredFromCheckpoint = context.isRestored();
+ resolvedTableId = null;
// Resolve producerId: use configured value or default to Flink job ID
resolvedProducerId = configuredProducerId;
@@ -312,6 +317,7 @@ public void initializeState(StateInitializationContext context) throws Exception
table = connection.getTable(tablePath);
}
+ TableInfo tableInfo = table.getTableInfo();
RecoveryOffsetManager offsetManager =
new RecoveryOffsetManager(
connection.getAdmin(),
@@ -321,14 +327,20 @@ public void initializeState(StateInitializationContext context) throws Exception
producerOffsetsPollIntervalMs,
maxPollTimeoutMs,
tablePath,
- table.getTableInfo());
+ tableInfo);
RecoveryOffsetManager.RecoveryDecision decision =
offsetManager.determineRecoveryStrategy(recoveredState);
+ // The recovery decision was built for this table identity and its current live buckets.
+ resolvedTableId = tableInfo.getTableId();
LOG.info("Recovery decision for subtask {}: {}", subtaskIndex, decision);
- // Step 4: Execute undo recovery if needed
+ // Step 4: Retain the complete positive-offset baseline regardless of whether any bucket
+ // currently needs Undo. Missing entries in V2 represent an explicit zero baseline.
+ Map recoveryOffsets = new HashMap<>(decision.getRecoveryOffsets());
+
+ // Step 5: Execute undo recovery if needed.
if (decision.needsUndoRecovery()) {
Map undoOffsets = decision.getUndoOffsets();
LOG.info(
@@ -338,21 +350,17 @@ public void initializeState(StateInitializationContext context) throws Exception
LOG.debug("Subtask {} undoOffsets details: {}", subtaskIndex, undoOffsets);
performUndoRecovery(undoOffsets);
-
- // Initialize bucket offsets with recovery offsets (checkpoint offsets)
- Map recoveryOffsets = decision.getRecoveryOffsets();
- LOG.info(
- "Subtask {} initializing bucketOffsets from recovery: {} buckets",
- subtaskIndex,
- recoveryOffsets.size());
- LOG.debug("Subtask {} recovery offsets details: {}", subtaskIndex, recoveryOffsets);
- initializeBucketOffsets(recoveryOffsets);
} else {
LOG.info("No undo recovery needed for subtask {}", subtaskIndex);
- // Initialize empty bucket offsets
- initializeBucketOffsets(new HashMap<>());
}
+ LOG.info(
+ "Subtask {} initializing complete baseline with {} positive offsets",
+ subtaskIndex,
+ recoveryOffsets.size());
+ LOG.debug("Subtask {} complete baseline details: {}", subtaskIndex, recoveryOffsets);
+ initializeBucketOffsets(recoveryOffsets);
+
LOG.info(
"UndoRecoveryOperator initialized for subtask {} with {} bucket offsets",
subtaskIndex,
@@ -410,7 +418,7 @@ private void performUndoRecovery(Map undoOffsets) thro
* Snapshots the current state during checkpoint.
*
* This method is called by Flink during checkpoint processing. It clears the existing state
- * list and adds a new {@link WriterState} with the current bucket offsets if the map is not
+ * list and adds exactly one complete V2 {@link WriterState}, including when the baseline map is
* empty.
*
*
Note: Producer offset cleanup is NOT done here. It is done in {@link
@@ -425,31 +433,24 @@ private void performUndoRecovery(Map undoOffsets) thro
public void snapshotState(StateSnapshotContext context) throws Exception {
super.snapshotState(context);
- // Clear existing state
+ checkState(bucketOffsets != null, "bucketOffsets must be initialized before snapshot");
+ checkState(resolvedTableId != null, "table ID must be resolved before snapshot");
+
undoStateList.clear();
+ // Persist one element even for an empty baseline so that V2 completeness survives restore.
+ undoStateList.add(WriterState.complete(resolvedTableId, new HashMap<>(bucketOffsets)));
- // Add new state if bucket offsets is not empty
- if (bucketOffsets != null) {
- if (!bucketOffsets.isEmpty()) {
- WriterState state = new WriterState(new HashMap<>(bucketOffsets));
- undoStateList.add(state);
- LOG.info(
- "Subtask {} snapshot state at checkpoint {}: {} buckets",
- subtaskIndex,
- context.getCheckpointId(),
- bucketOffsets.size());
- LOG.debug(
- "Subtask {} checkpoint {} bucketOffsets details: {}",
- subtaskIndex,
- context.getCheckpointId(),
- bucketOffsets);
- } else {
- LOG.debug(
- "Subtask {} snapshot state at checkpoint {}: bucketOffsets is EMPTY",
- subtaskIndex,
- context.getCheckpointId());
- }
- }
+ LOG.info(
+ "Subtask {} snapshot complete V2 state at checkpoint {}: tableId={}, {} positive offsets",
+ subtaskIndex,
+ context.getCheckpointId(),
+ resolvedTableId,
+ bucketOffsets.size());
+ LOG.debug(
+ "Subtask {} checkpoint {} complete baseline details: {}",
+ subtaskIndex,
+ context.getCheckpointId(),
+ bucketOffsets);
}
/**
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/state/WriterStateSerializerTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/state/WriterStateSerializerTest.java
index c386190a81b..2876015a3e7 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/state/WriterStateSerializerTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/state/WriterStateSerializerTest.java
@@ -28,10 +28,12 @@
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Unit tests for {@link WriterStateSerializer}'s TypeSerializer implementation. */
class WriterStateSerializerTest {
@@ -43,61 +45,16 @@ void setUp() {
serializer = new WriterStateSerializer();
}
- /** Tests serialize/deserialize with normal state, empty state, and reuse scenarios. */
+ /** Tests that binary copy preserves both V1 legacy and V2 complete state formats. */
@Test
- void testSerializeDeserialize() throws IOException {
- // Test with multiple bucket offsets (including partitioned table)
- Map bucketOffsets = new HashMap<>();
- bucketOffsets.put(new TableBucket(1L, null, 0), 100L);
- bucketOffsets.put(new TableBucket(1L, null, 1), 200L);
- bucketOffsets.put(new TableBucket(2L, 10L, 0), 300L);
- WriterState original = new WriterState(bucketOffsets);
-
- DataOutputSerializer output = new DataOutputSerializer(256);
- serializer.serialize(original, output);
-
- DataInputDeserializer input = new DataInputDeserializer(output.getCopyOfBuffer());
- WriterState deserialized = serializer.deserialize(input);
-
- assertThat(deserialized).isEqualTo(original);
- assertThat(deserialized.getBucketOffsets()).hasSize(3);
-
- // Test empty state
- WriterState emptyState = WriterState.empty();
- output = new DataOutputSerializer(64);
- serializer.serialize(emptyState, output);
- input = new DataInputDeserializer(output.getCopyOfBuffer());
- assertThat(serializer.deserialize(input)).isEqualTo(emptyState);
-
- // Test deserialize with reuse (should be ignored for immutable types)
- output = new DataOutputSerializer(256);
- serializer.serialize(original, output);
- input = new DataInputDeserializer(output.getCopyOfBuffer());
- WriterState reuse = WriterState.empty();
- deserialized = serializer.deserialize(reuse, input);
- assertThat(deserialized).isEqualTo(original);
- assertThat(deserialized).isNotSameAs(reuse);
- }
-
- /** Tests copy between DataInputView and DataOutputView. */
- @Test
- void testCopyBetweenViews() throws IOException {
- Map bucketOffsets = new HashMap<>();
- bucketOffsets.put(new TableBucket(1L, null, 0), 100L);
- bucketOffsets.put(new TableBucket(2L, 10L, 1), 200L);
- WriterState original = new WriterState(bucketOffsets);
-
- DataOutputSerializer sourceOutput = new DataOutputSerializer(256);
- serializer.serialize(original, sourceOutput);
-
- DataInputDeserializer sourceInput =
- new DataInputDeserializer(sourceOutput.getCopyOfBuffer());
- DataOutputSerializer targetOutput = new DataOutputSerializer(256);
- serializer.copy(sourceInput, targetOutput);
-
- DataInputDeserializer targetInput =
- new DataInputDeserializer(targetOutput.getCopyOfBuffer());
- assertThat(serializer.deserialize(targetInput)).isEqualTo(original);
+ void testCopyPreservesV1AndV2State() throws IOException {
+ byte[] v1Payload = createV1Payload();
+ WriterState v1State = serializer.deserialize(new DataInputDeserializer(v1Payload));
+ assertCopiedState(v1Payload, v1State);
+
+ byte[] v2Payload = createV2Payload();
+ WriterState v2State = serializer.deserialize(new DataInputDeserializer(v2Payload));
+ assertCopiedState(v2Payload, v2State);
}
/** Tests TypeSerializerSnapshot creation and restoration. */
@@ -141,8 +98,7 @@ void testSerializerProperties() {
assertThat(serializer.hashCode()).isEqualTo(other.hashCode());
// CreateInstance returns empty state
- WriterState instance = serializer.createInstance();
- assertThat(instance.getBucketOffsets()).isEmpty();
+ assertThat(serializer.createInstance()).isEqualTo(WriterState.empty());
// Copy returns same instance for immutable types
Map bucketOffsets = new HashMap<>();
@@ -152,23 +108,109 @@ void testSerializerProperties() {
assertThat(serializer.copy(original, WriterState.empty())).isSameAs(original);
}
- /** Tests that TypeSerializer and SimpleVersionedSerializer produce compatible bytes. */
@Test
- void testSimpleVersionedSerializerCompatibility() throws IOException {
- Map bucketOffsets = new HashMap<>();
- bucketOffsets.put(new TableBucket(1L, null, 0), 100L);
- bucketOffsets.put(new TableBucket(2L, 10L, 1), 200L);
- WriterState original = new WriterState(bucketOffsets);
+ void testDeserializeLiteralV1PayloadRemainsLegacy() throws IOException {
+ Map expectedOffsets = new HashMap<>();
+ expectedOffsets.put(new TableBucket(1L, null, 0), 100L);
+ expectedOffsets.put(new TableBucket(1L, 10L, 1), 200L);
+
+ WriterState reuse = WriterState.empty();
+ WriterState state =
+ serializer.deserialize(reuse, new DataInputDeserializer(createV1Payload()));
- // Serialize using TypeSerializer
- DataOutputSerializer output = new DataOutputSerializer(256);
+ assertThat(state.getStateFormat()).isEqualTo(WriterState.StateFormat.V1_LEGACY);
+ assertThat(state.getBucketOffsets()).isEqualTo(expectedOffsets);
+ assertThat(state).isNotSameAs(reuse);
+ }
+
+ @Test
+ void testV2GoldenPayload() throws IOException {
+ long tableId = 7L;
+ Map offsets =
+ Collections.singletonMap(new TableBucket(tableId, 12L, 1), 20L);
+ WriterState original = WriterState.complete(tableId, offsets);
+
+ byte[] expectedPayload = createV2Payload();
+ WriterState restored = serializer.deserialize(new DataInputDeserializer(expectedPayload));
+ DataOutputSerializer output = new DataOutputSerializer(128);
+ serializer.serialize(original, output);
+
+ assertThat(restored).isEqualTo(original);
+ assertThat(output.getCopyOfBuffer()).containsExactly(expectedPayload);
+ }
+
+ @Test
+ void testV2RoundTripWithEmptyCompleteMarker() throws IOException {
+ WriterState original = WriterState.complete(7L, Collections.emptyMap());
+
+ DataOutputSerializer output = new DataOutputSerializer(32);
serializer.serialize(original, output);
- byte[] typeSerializerBytes = output.getCopyOfBuffer();
+ WriterState restored =
+ serializer.deserialize(new DataInputDeserializer(output.getCopyOfBuffer()));
+
+ assertThat(restored.getStateFormat()).isEqualTo(WriterState.StateFormat.V2_COMPLETE);
+ assertThat(restored.getTableId()).isEqualTo(7L);
+ assertThat(restored.getBucketOffsets()).isEmpty();
+ assertThat(restored).isNotEqualTo(WriterState.empty());
+ }
+
+ @Test
+ void testRejectsUnsupportedVersion() throws IOException {
+ assertMalformedPayload(payloadWithVersion(99), "version");
+ }
+
+ private void assertCopiedState(byte[] sourceBytes, WriterState expectedState)
+ throws IOException {
+ DataOutputSerializer copiedOutput = new DataOutputSerializer(128);
+ serializer.copy(new DataInputDeserializer(sourceBytes), copiedOutput);
+
+ WriterState copied =
+ serializer.deserialize(new DataInputDeserializer(copiedOutput.getCopyOfBuffer()));
+ assertThat(copied).isEqualTo(expectedState);
+ }
+
+ private void assertMalformedPayload(byte[] payload, String expectedMessage) {
+ assertThatThrownBy(() -> serializer.deserialize(new DataInputDeserializer(payload)))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining(expectedMessage);
+ }
+
+ private static byte[] createV1Payload() throws IOException {
+ DataOutputSerializer output = new DataOutputSerializer(128);
+ output.writeInt(1);
+ output.writeInt(2);
+ writeV1Entry(output, 1L, null, 0, 100L);
+ writeV1Entry(output, 1L, 10L, 1, 200L);
+ return output.getCopyOfBuffer();
+ }
- // Both should deserialize to equal results
- DataInputDeserializer input = new DataInputDeserializer(typeSerializerBytes);
- WriterState fromTypeSerializer = serializer.deserialize(input);
+ private static byte[] createV2Payload() throws IOException {
+ DataOutputSerializer output = new DataOutputSerializer(64);
+ output.writeInt(2);
+ output.writeLong(7L);
+ output.writeInt(1);
+ output.writeBoolean(true);
+ output.writeLong(12L);
+ output.writeInt(1);
+ output.writeLong(20L);
+ return output.getCopyOfBuffer();
+ }
+
+ private static byte[] payloadWithVersion(int version) throws IOException {
+ DataOutputSerializer output = new DataOutputSerializer(8);
+ output.writeInt(version);
+ return output.getCopyOfBuffer();
+ }
- assertThat(fromTypeSerializer).isEqualTo(original);
+ private static void writeV1Entry(
+ DataOutputSerializer output, long tableId, Long partitionId, int bucketId, long offset)
+ throws IOException {
+ output.writeLong(tableId);
+ output.writeBoolean(partitionId != null);
+ if (partitionId != null) {
+ output.writeLong(partitionId);
+ }
+ output.writeInt(bucketId);
+ output.writeLong(offset);
}
}
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManagerTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManagerTest.java
index 2e072ef757a..a9e3ec66f41 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManagerTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/undo/RecoveryOffsetManagerTest.java
@@ -63,6 +63,10 @@ public class RecoveryOffsetManagerTest {
// ==================== Test Data Helpers ====================
private static TableInfo createTableInfo(int numBuckets, boolean isPartitioned) {
+ return createTableInfo(TABLE_ID, numBuckets, isPartitioned);
+ }
+
+ private static TableInfo createTableInfo(long tableId, int numBuckets, boolean isPartitioned) {
Schema schema =
Schema.newBuilder()
.column("id", DataTypes.INT())
@@ -75,7 +79,7 @@ private static TableInfo createTableInfo(int numBuckets, boolean isPartitioned)
return new TableInfo(
TABLE_PATH,
- TABLE_ID,
+ tableId,
0, // schemaId
schema,
Collections.emptyList(), // bucketKeys
@@ -89,6 +93,12 @@ private static TableInfo createTableInfo(int numBuckets, boolean isPartitioned)
System.currentTimeMillis());
}
+ private static RecoveryOffsetManager createManager(
+ RecoveryTestAdmin admin, int subtaskIndex, int parallelism, TableInfo tableInfo) {
+ return new RecoveryOffsetManager(
+ admin, PRODUCER_ID, subtaskIndex, parallelism, 10L, 5000L, TABLE_PATH, tableInfo);
+ }
+
private static PartitionInfo createPartitionInfo(long partitionId, String partitionName) {
ResolvedPartitionSpec spec = ResolvedPartitionSpec.fromPartitionValue("pt", partitionName);
return new PartitionInfo(partitionId, spec, DEFAULT_REMOTE_DATA_DIR);
@@ -97,31 +107,24 @@ private static PartitionInfo createPartitionInfo(long partitionId, String partit
// ==================== FRESH_START Tests ====================
@Test
- void testFreshStartWithEmptyCheckpoint() throws Exception {
- // Setup: current offsets are all 0 (no data written)
+ void testEmptyRestoredStateIsRejectedBecauseCompletenessIsUnknown() {
Map currentOffsets = new HashMap<>();
currentOffsets.put(new TableBucket(TABLE_ID, 0), 0L);
currentOffsets.put(new TableBucket(TABLE_ID, 1), 0L);
RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
TableInfo tableInfo = createTableInfo(2, false);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
-
- // Execute: empty checkpoint (checkpoint exists but no data written)
- RecoveryOffsetManager.RecoveryDecision decision =
- manager.determineRecoveryStrategy(new ArrayList<>());
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
- // Verify
- assertThat(decision.getStrategy())
- .isEqualTo(RecoveryOffsetManager.RecoveryStrategy.FRESH_START);
- assertThat(decision.needsUndoRecovery()).isFalse();
- assertThat(decision.getRecoveryOffsets()).isNull();
+ assertThatThrownBy(() -> manager.determineRecoveryStrategy(new ArrayList<>()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("restored")
+ .hasMessageContaining("no state fragments");
+ assertThat(admin.wasRegisterCalled()).isFalse();
}
@Test
- void testFreshStartWhenOffsetsMatch() throws Exception {
+ void testNoUndoWhenCheckpointOffsetsMatchCurrent() throws Exception {
// Setup: checkpoint offsets match current offsets
Map currentOffsets = new HashMap<>();
currentOffsets.put(new TableBucket(TABLE_ID, 0), 100L);
@@ -129,13 +132,11 @@ void testFreshStartWhenOffsetsMatch() throws Exception {
RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
TableInfo tableInfo = createTableInfo(2, false);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
// Checkpoint offsets match current
Map checkpointOffsets = new HashMap<>(currentOffsets);
- WriterState state = new WriterState(checkpointOffsets);
+ WriterState state = WriterState.complete(TABLE_ID, checkpointOffsets);
// Execute
RecoveryOffsetManager.RecoveryDecision decision =
@@ -145,13 +146,14 @@ void testFreshStartWhenOffsetsMatch() throws Exception {
assertThat(decision.getStrategy())
.isEqualTo(RecoveryOffsetManager.RecoveryStrategy.FRESH_START);
assertThat(decision.needsUndoRecovery()).isFalse();
- assertThat(decision.getRecoveryOffsets()).isNull();
+ assertThat(decision.getRecoveryOffsets())
+ .containsExactlyInAnyOrderEntriesOf(checkpointOffsets);
}
// ==================== CHECKPOINT_RECOVERY Tests ====================
@Test
- void testCheckpointRecoveryWithChangedOffsets() throws Exception {
+ void testCheckpointOffsetsBehindCurrentRequireUndo() throws Exception {
// Setup: current offsets are ahead of checkpoint
Map currentOffsets = new HashMap<>();
currentOffsets.put(new TableBucket(TABLE_ID, 0), 150L);
@@ -159,15 +161,13 @@ void testCheckpointRecoveryWithChangedOffsets() throws Exception {
RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
TableInfo tableInfo = createTableInfo(2, false);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
// Checkpoint offsets are behind current
Map checkpointOffsets = new HashMap<>();
checkpointOffsets.put(new TableBucket(TABLE_ID, 0), 100L);
checkpointOffsets.put(new TableBucket(TABLE_ID, 1), 200L);
- WriterState state = new WriterState(checkpointOffsets);
+ WriterState state = WriterState.complete(TABLE_ID, checkpointOffsets);
// Execute
RecoveryOffsetManager.RecoveryDecision decision =
@@ -183,7 +183,7 @@ void testCheckpointRecoveryWithChangedOffsets() throws Exception {
}
@Test
- void testCheckpointRecoveryWithRescaling() throws Exception {
+ void testUnionStateIsFilteredBySubtaskAssignment() throws Exception {
// Simulates rescaling from parallelism=4 to parallelism=4 with state redistribution.
// Previously each subtask had one unique bucket. After rescaling, Flink may redistribute
// all 4 states to subtask 0. The manager should filter by bucket assignment and only
@@ -199,26 +199,24 @@ void testCheckpointRecoveryWithRescaling() throws Exception {
RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
TableInfo tableInfo = createTableInfo(4, false);
// subtaskIndex=0, parallelism=4: only bucket 0 assigned (0 % 4 = 0)
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 4, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 4, tableInfo);
// 4 distinct states from previous subtasks, each with a unique bucket
Map offsets0 = new HashMap<>();
offsets0.put(new TableBucket(TABLE_ID, 0), 100L);
- WriterState state0 = new WriterState(offsets0);
+ WriterState state0 = WriterState.complete(TABLE_ID, offsets0);
Map offsets1 = new HashMap<>();
offsets1.put(new TableBucket(TABLE_ID, 1), 150L);
- WriterState state1 = new WriterState(offsets1);
+ WriterState state1 = WriterState.complete(TABLE_ID, offsets1);
Map offsets2 = new HashMap<>();
offsets2.put(new TableBucket(TABLE_ID, 2), 200L);
- WriterState state2 = new WriterState(offsets2);
+ WriterState state2 = WriterState.complete(TABLE_ID, offsets2);
Map offsets3 = new HashMap<>();
offsets3.put(new TableBucket(TABLE_ID, 3), 250L);
- WriterState state3 = new WriterState(offsets3);
+ WriterState state3 = WriterState.complete(TABLE_ID, offsets3);
// Execute: subtask 0 receives all 4 states after rescaling
RecoveryOffsetManager.RecoveryDecision decision =
@@ -234,7 +232,7 @@ void testCheckpointRecoveryWithRescaling() throws Exception {
}
@Test
- void testNewBucketUsesZeroAsRecoveryOffset() throws Exception {
+ void testV2MissingLiveBucketUsesZeroBaseline() throws Exception {
// Setup: bucket 1 has data but not in checkpoint
Map currentOffsets = new HashMap<>();
currentOffsets.put(new TableBucket(TABLE_ID, 0), 100L);
@@ -242,14 +240,12 @@ void testNewBucketUsesZeroAsRecoveryOffset() throws Exception {
RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
TableInfo tableInfo = createTableInfo(2, false);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
// Checkpoint only has bucket 0
Map checkpointOffsets = new HashMap<>();
checkpointOffsets.put(new TableBucket(TABLE_ID, 0), 100L);
- WriterState state = new WriterState(checkpointOffsets);
+ WriterState state = WriterState.complete(TABLE_ID, checkpointOffsets);
// Execute
RecoveryOffsetManager.RecoveryDecision decision =
@@ -259,8 +255,12 @@ void testNewBucketUsesZeroAsRecoveryOffset() throws Exception {
assertThat(decision.getStrategy())
.isEqualTo(RecoveryOffsetManager.RecoveryStrategy.CHECKPOINT_RECOVERY);
assertThat(decision.needsUndoRecovery()).isTrue();
- assertThat(decision.getRecoveryOffsets()).hasSize(1);
- assertThat(decision.getRecoveryOffsets().get(new TableBucket(TABLE_ID, 1))).isEqualTo(0L);
+ assertThat(decision.getRecoveryOffsets()).containsOnlyKeys(new TableBucket(TABLE_ID, 0));
+ assertThat(
+ decision.getUndoOffsets()
+ .get(new TableBucket(TABLE_ID, 1))
+ .getCheckpointOffset())
+ .isZero();
}
// ==================== PRODUCER_OFFSET_RECOVERY Tests ====================
@@ -278,9 +278,7 @@ void testProducerOffsetRecoveryTask0() throws Exception {
admin.setInitialOffsetsForRegistration(initialOffsets);
TableInfo tableInfo = createTableInfo(1, false);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
// Execute: null means no checkpoint
RecoveryOffsetManager.RecoveryDecision decision = manager.determineRecoveryStrategy(null);
@@ -313,9 +311,7 @@ void testProducerOffsetRecoveryNonTask0() throws Exception {
TableInfo tableInfo = createTableInfo(2, false);
// subtaskIndex=1, parallelism=2: bucket 1 assigned
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 1, 2, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 1, 2, tableInfo);
// Execute
RecoveryOffsetManager.RecoveryDecision decision = manager.determineRecoveryStrategy(null);
@@ -331,18 +327,17 @@ void testProducerOffsetRecoveryNonTask0() throws Exception {
}
@Test
- void testNoUndoNeededWhenProducerOffsetsMatchCurrent() throws Exception {
+ void testProducerOffsetsAreRetainedWhenNoUndoIsNeeded() throws Exception {
// Setup: registered offsets match current (no writes after registration)
Map currentOffsets = new HashMap<>();
currentOffsets.put(new TableBucket(TABLE_ID, 0), 100L);
+ currentOffsets.put(new TableBucket(TABLE_ID, 1), 200L);
RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
admin.setInitialOffsetsForRegistration(currentOffsets);
- TableInfo tableInfo = createTableInfo(1, false);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ TableInfo tableInfo = createTableInfo(2, false);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
// Execute
RecoveryOffsetManager.RecoveryDecision decision = manager.determineRecoveryStrategy(null);
@@ -351,12 +346,160 @@ void testNoUndoNeededWhenProducerOffsetsMatchCurrent() throws Exception {
assertThat(decision.getStrategy())
.isEqualTo(RecoveryOffsetManager.RecoveryStrategy.FRESH_START);
assertThat(decision.needsUndoRecovery()).isFalse();
+ assertThat(decision.getRecoveryOffsets())
+ .as("the complete producer baseline must survive even when no Undo is needed")
+ .containsExactlyInAnyOrderEntriesOf(currentOffsets);
+ }
+
+ @Test
+ void testRecoveryOffsetsRetainBucketsOutsideUndoWorkSet() throws Exception {
+ TableBucket unchangedBucket = new TableBucket(TABLE_ID, 0);
+ TableBucket changedBucket = new TableBucket(TABLE_ID, 1);
+
+ Map checkpointOffsets = new HashMap<>();
+ checkpointOffsets.put(unchangedBucket, 100L);
+ checkpointOffsets.put(changedBucket, 200L);
+
+ Map currentOffsets = new HashMap<>();
+ currentOffsets.put(unchangedBucket, 100L);
+ currentOffsets.put(changedBucket, 260L);
+
+ RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, createTableInfo(2, false));
+
+ RecoveryOffsetManager.RecoveryDecision decision =
+ manager.determineRecoveryStrategy(
+ Collections.singletonList(
+ WriterState.complete(TABLE_ID, checkpointOffsets)));
+
+ assertThat(decision.getRecoveryOffsets())
+ .as("the next checkpoint baseline must retain changed and unchanged buckets")
+ .containsExactlyInAnyOrderEntriesOf(checkpointOffsets);
+ assertThat(decision.getUndoOffsets()).containsOnlyKeys(changedBucket);
+ assertThat(decision.getUndoOffsets().get(changedBucket).getCheckpointOffset())
+ .isEqualTo(200L);
+ assertThat(decision.getUndoOffsets().get(changedBucket).getLogEndOffset()).isEqualTo(260L);
+ }
+
+ @Test
+ void testSparseV1StateUsesZeroForMissingBucket() throws Exception {
+ TableBucket missingBucket = new TableBucket(TABLE_ID, 0);
+ TableBucket knownBucket = new TableBucket(TABLE_ID, 1);
+
+ Map currentOffsets = new HashMap<>();
+ currentOffsets.put(missingBucket, 1_000_000L);
+ currentOffsets.put(knownBucket, 260L);
+
+ RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, createTableInfo(2, false));
+
+ RecoveryOffsetManager.RecoveryDecision decision =
+ manager.determineRecoveryStrategy(
+ Collections.singletonList(
+ new WriterState(Collections.singletonMap(knownBucket, 260L))));
+
+ assertThat(decision.getRecoveryOffsets()).containsOnlyKeys(knownBucket);
+ assertThat(decision.getRecoveryOffsets().get(knownBucket)).isEqualTo(260L);
+ assertThat(decision.getUndoOffsets()).containsOnlyKeys(missingBucket);
+ assertThat(decision.getUndoOffsets().get(missingBucket).getCheckpointOffset()).isZero();
+ assertThat(decision.getUndoOffsets().get(missingBucket).getLogEndOffset())
+ .isEqualTo(1_000_000L);
+ assertThat(admin.wasRegisterCalled()).isFalse();
+ }
+
+ @Test
+ void testV1OffsetsAreRetainedWhenNoUndoIsNeeded() throws Exception {
+ TableBucket bucket = new TableBucket(TABLE_ID, 0);
+ Map offsets = Collections.singletonMap(bucket, 260L);
+ RecoveryTestAdmin admin = new RecoveryTestAdmin(offsets);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, createTableInfo(1, false));
+
+ RecoveryOffsetManager.RecoveryDecision decision =
+ manager.determineRecoveryStrategy(
+ Collections.singletonList(new WriterState(offsets)));
+
+ assertThat(decision.getRecoveryOffsets()).containsOnlyKeys(bucket);
+ assertThat(decision.getRecoveryOffsets().get(bucket)).isEqualTo(260L);
+ assertThat(decision.needsUndoRecovery()).isFalse();
+ }
+
+ @Test
+ void testMixedV1AndV2FragmentsFail() {
+ TableBucket bucket = new TableBucket(TABLE_ID, 0);
+ Map offsets = Collections.singletonMap(bucket, 10L);
+ RecoveryTestAdmin admin = new RecoveryTestAdmin(offsets);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, createTableInfo(1, false));
+
+ assertThatThrownBy(
+ () ->
+ manager.determineRecoveryStrategy(
+ Arrays.asList(
+ new WriterState(offsets),
+ WriterState.complete(TABLE_ID, offsets))))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("mixed")
+ .hasMessageContaining("V1")
+ .hasMessageContaining("V2");
+ }
+
+ @Test
+ void testConflictingFragmentOffsetsFail() {
+ TableBucket bucket = new TableBucket(TABLE_ID, 0);
+ Map first = Collections.singletonMap(bucket, 10L);
+ Map second = Collections.singletonMap(bucket, 20L);
+ RecoveryTestAdmin admin = new RecoveryTestAdmin(second);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, createTableInfo(1, false));
+
+ assertThatThrownBy(
+ () ->
+ manager.determineRecoveryStrategy(
+ Arrays.asList(
+ WriterState.complete(TABLE_ID, first),
+ WriterState.complete(TABLE_ID, second))))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("Conflicting")
+ .hasMessageContaining(bucket.toString())
+ .hasMessageContaining("10")
+ .hasMessageContaining("20");
+ }
+
+ @Test
+ void testV2EmptyMarkerWithMismatchedTableIdFails() {
+ TableBucket bucket = new TableBucket(TABLE_ID, 0);
+ RecoveryTestAdmin admin = new RecoveryTestAdmin(Collections.singletonMap(bucket, 0L));
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, createTableInfo(1, false));
+
+ assertThatThrownBy(
+ () ->
+ manager.determineRecoveryStrategy(
+ Collections.singletonList(
+ WriterState.complete(
+ TABLE_ID + 1, Collections.emptyMap()))))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("table ID")
+ .hasMessageContaining(String.valueOf(TABLE_ID + 1));
+ }
+
+ @Test
+ void testMissingLatestOffsetFailsRecovery() {
+ RecoveryTestAdmin admin = new RecoveryTestAdmin(new HashMap<>());
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, createTableInfo(1, false));
+
+ assertThatThrownBy(
+ () ->
+ manager.determineRecoveryStrategy(
+ Collections.singletonList(
+ WriterState.complete(
+ TABLE_ID, Collections.emptyMap()))))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("latest offset")
+ .hasMessageContaining("missing");
}
// ==================== Sharding Tests ====================
@Test
- void testShardingFiltersBuckets() throws Exception {
+ void testRecoveryOffsetsAreFilteredBySubtaskAssignment() throws Exception {
// Setup: 3 buckets, parallelism 3
Map currentOffsets = new HashMap<>();
currentOffsets.put(new TableBucket(TABLE_ID, 0), 150L);
@@ -366,15 +509,13 @@ void testShardingFiltersBuckets() throws Exception {
RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
TableInfo tableInfo = createTableInfo(3, false);
// subtask 0 with parallelism 3: only bucket 0 assigned (0 % 3 = 0)
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 3, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 3, tableInfo);
Map checkpointOffsets = new HashMap<>();
checkpointOffsets.put(new TableBucket(TABLE_ID, 0), 100L);
checkpointOffsets.put(new TableBucket(TABLE_ID, 1), 200L);
checkpointOffsets.put(new TableBucket(TABLE_ID, 2), 300L);
- WriterState state = new WriterState(checkpointOffsets);
+ WriterState state = WriterState.complete(TABLE_ID, checkpointOffsets);
// Execute
RecoveryOffsetManager.RecoveryDecision decision =
@@ -390,7 +531,7 @@ void testShardingFiltersBuckets() throws Exception {
}
@Test
- void testSubtaskWithNoBuckets() throws Exception {
+ void testSubtaskWithNoAssignedBucketsReturnsEmptyDecision() throws Exception {
// Setup: 2 buckets, parallelism 4 -> subtask 2 and 3 have no buckets
Map currentOffsets = new HashMap<>();
currentOffsets.put(new TableBucket(TABLE_ID, 0), 100L);
@@ -399,14 +540,12 @@ void testSubtaskWithNoBuckets() throws Exception {
RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
TableInfo tableInfo = createTableInfo(2, false);
// subtask 2 with parallelism 4: no buckets assigned
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 2, 4, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 2, 4, tableInfo);
Map checkpointOffsets = new HashMap<>();
checkpointOffsets.put(new TableBucket(TABLE_ID, 0), 50L);
checkpointOffsets.put(new TableBucket(TABLE_ID, 1), 100L);
- WriterState state = new WriterState(checkpointOffsets);
+ WriterState state = WriterState.complete(TABLE_ID, checkpointOffsets);
// Execute
RecoveryOffsetManager.RecoveryDecision decision =
@@ -416,13 +555,13 @@ void testSubtaskWithNoBuckets() throws Exception {
assertThat(decision.getStrategy())
.isEqualTo(RecoveryOffsetManager.RecoveryStrategy.FRESH_START);
assertThat(decision.needsUndoRecovery()).isFalse();
- assertThat(decision.getRecoveryOffsets()).isNull();
+ assertThat(decision.getRecoveryOffsets()).isEmpty();
}
// ==================== Partitioned Table Tests ====================
@Test
- void testPartitionedTableRecovery() throws Exception {
+ void testPartitionedRecoveryRetainsOffsetsForLivePartitions() throws Exception {
// Setup: partitioned table with 2 partitions
Map currentOffsets = new HashMap<>();
currentOffsets.put(new TableBucket(TABLE_ID, 1L, 0), 150L);
@@ -435,14 +574,12 @@ void testPartitionedTableRecovery() throws Exception {
admin.setPartitions(partitions);
TableInfo tableInfo = createTableInfo(1, true);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
Map checkpointOffsets = new HashMap<>();
checkpointOffsets.put(new TableBucket(TABLE_ID, 1L, 0), 100L);
checkpointOffsets.put(new TableBucket(TABLE_ID, 2L, 0), 200L);
- WriterState state = new WriterState(checkpointOffsets);
+ WriterState state = WriterState.complete(TABLE_ID, checkpointOffsets);
// Execute
RecoveryOffsetManager.RecoveryDecision decision =
@@ -473,14 +610,12 @@ void testNewPartitionUsesZeroAsRecoveryOffset() throws Exception {
admin.setPartitions(partitions);
TableInfo tableInfo = createTableInfo(1, true);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
// Checkpoint only has partition 1
Map checkpointOffsets = new HashMap<>();
checkpointOffsets.put(new TableBucket(TABLE_ID, 1L, 0), 100L);
- WriterState state = new WriterState(checkpointOffsets);
+ WriterState state = WriterState.complete(TABLE_ID, checkpointOffsets);
// Execute
RecoveryOffsetManager.RecoveryDecision decision =
@@ -490,9 +625,13 @@ void testNewPartitionUsesZeroAsRecoveryOffset() throws Exception {
assertThat(decision.getStrategy())
.isEqualTo(RecoveryOffsetManager.RecoveryStrategy.CHECKPOINT_RECOVERY);
assertThat(decision.needsUndoRecovery()).isTrue();
- assertThat(decision.getRecoveryOffsets()).hasSize(1);
- assertThat(decision.getRecoveryOffsets().get(new TableBucket(TABLE_ID, 2L, 0)))
- .isEqualTo(0L);
+ assertThat(decision.getRecoveryOffsets())
+ .containsOnlyKeys(new TableBucket(TABLE_ID, 1L, 0));
+ assertThat(
+ decision.getUndoOffsets()
+ .get(new TableBucket(TABLE_ID, 2L, 0))
+ .getCheckpointOffset())
+ .isZero();
}
@Test
@@ -515,9 +654,7 @@ void testPartitionedTableProducerOffsetRecoveryTask0() throws Exception {
admin.setInitialOffsetsForRegistration(initialOffsets);
TableInfo tableInfo = createTableInfo(1, true);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
// Execute: null means no checkpoint (producer offset recovery)
RecoveryOffsetManager.RecoveryDecision decision = manager.determineRecoveryStrategy(null);
@@ -539,7 +676,7 @@ void testPartitionedTableProducerOffsetRecoveryTask0() throws Exception {
// ==================== Error Handling Tests ====================
@Test
- void testTableRecreatedThrowsException() throws Exception {
+ void testCheckpointFromRecreatedTableIsRejected() throws Exception {
// Simulates: savepoint → drop table → re-create table → restore from savepoint.
// The checkpoint state contains buckets with the old table ID (TABLE_ID=1),
// but the re-created table has a new table ID (NEW_TABLE_ID=999).
@@ -553,38 +690,35 @@ void testTableRecreatedThrowsException() throws Exception {
RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
// Use the package-private constructor to set the new tableId directly
RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, newTableId, 1, false);
+ createManager(admin, 0, 1, createTableInfo(newTableId, 1, false));
// Checkpoint state from before table was dropped (old tableId)
Map checkpointOffsets = new HashMap<>();
checkpointOffsets.put(new TableBucket(oldTableId, 0), 100L);
- WriterState state = new WriterState(checkpointOffsets);
+ WriterState state = WriterState.complete(oldTableId, checkpointOffsets);
// Execute & Verify: should detect table re-creation and throw
assertThatThrownBy(
() -> manager.determineRecoveryStrategy(Collections.singletonList(state)))
.isInstanceOf(IllegalStateException.class)
- .hasMessageContaining("re-created")
+ .hasMessageContaining("table ID")
.hasMessageContaining(String.valueOf(oldTableId))
.hasMessageContaining(String.valueOf(newTableId));
}
@Test
- void testDataInconsistencyThrowsException() throws Exception {
+ void testRecoveryOffsetAheadOfCurrentOffsetFails() throws Exception {
// Setup: checkpoint offset > current offset (data loss)
Map currentOffsets = new HashMap<>();
currentOffsets.put(new TableBucket(TABLE_ID, 0), 50L); // current < checkpoint
RecoveryTestAdmin admin = new RecoveryTestAdmin(currentOffsets);
TableInfo tableInfo = createTableInfo(1, false);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
Map checkpointOffsets = new HashMap<>();
checkpointOffsets.put(new TableBucket(TABLE_ID, 0), 100L); // checkpoint > current
- WriterState state = new WriterState(checkpointOffsets);
+ WriterState state = WriterState.complete(TABLE_ID, checkpointOffsets);
// Execute & Verify
assertThatThrownBy(
@@ -600,13 +734,11 @@ void testListOffsetsFailureThrowsException() throws Exception {
admin.setListOffsetsFailure(new RuntimeException("Connection failed"));
TableInfo tableInfo = createTableInfo(1, false);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
Map checkpointOffsets = new HashMap<>();
checkpointOffsets.put(new TableBucket(TABLE_ID, 0), 100L);
- WriterState state = new WriterState(checkpointOffsets);
+ WriterState state = WriterState.complete(TABLE_ID, checkpointOffsets);
// Execute & Verify
assertThatThrownBy(
@@ -626,13 +758,11 @@ void testListPartitionInfosFailureThrowsException() throws Exception {
admin.setListPartitionInfosFailure(new RuntimeException("Partition lookup failed"));
TableInfo tableInfo = createTableInfo(1, true);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
Map checkpointOffsets = new HashMap<>();
checkpointOffsets.put(new TableBucket(TABLE_ID, 1L, 0), 50L);
- WriterState state = new WriterState(checkpointOffsets);
+ WriterState state = WriterState.complete(TABLE_ID, checkpointOffsets);
// Execute & Verify
assertThatThrownBy(
@@ -649,9 +779,7 @@ void testCleanupOffsetsTask0() {
// Setup
RecoveryTestAdmin admin = new RecoveryTestAdmin(new HashMap<>());
TableInfo tableInfo = createTableInfo(1, false);
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 0, 1, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 0, 1, tableInfo);
// Execute
manager.cleanupOffsets();
@@ -667,9 +795,7 @@ void testCleanupOffsetsNonTask0() {
RecoveryTestAdmin admin = new RecoveryTestAdmin(new HashMap<>());
TableInfo tableInfo = createTableInfo(1, false);
// subtaskIndex=1 (non-Task0)
- RecoveryOffsetManager manager =
- new RecoveryOffsetManager(
- admin, PRODUCER_ID, 1, 2, 10L, 5000L, TABLE_PATH, tableInfo);
+ RecoveryOffsetManager manager = createManager(admin, 1, 2, tableInfo);
// Execute
manager.cleanupOffsets();
@@ -786,8 +912,10 @@ private ListOffsetsResult createListOffsetsResult(
partitionId != null
? new TableBucket(TABLE_ID, partitionId, bucketId)
: new TableBucket(TABLE_ID, bucketId);
- long offset = currentOffsets.getOrDefault(tb, 0L);
- futures.put(bucketId, CompletableFuture.completedFuture(offset));
+ if (currentOffsets.containsKey(tb)) {
+ futures.put(
+ bucketId, CompletableFuture.completedFuture(currentOffsets.get(tb)));
+ }
}
return new ListOffsetsResult(futures);
}
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/undo/UndoRecoveryOperatorStateTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/undo/UndoRecoveryOperatorStateTest.java
new file mode 100644
index 00000000000..97a07887d83
--- /dev/null
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/undo/UndoRecoveryOperatorStateTest.java
@@ -0,0 +1,340 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.sink.undo;
+
+import org.apache.fluss.client.admin.OffsetSpec;
+import org.apache.fluss.client.lookup.Lookuper;
+import org.apache.fluss.client.table.Table;
+import org.apache.fluss.client.table.writer.UpsertResult;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.sink.state.WriterState;
+import org.apache.fluss.flink.sink.state.WriterStateSerializer;
+import org.apache.fluss.flink.utils.FlinkTestBase;
+import org.apache.fluss.metadata.AggFunctions;
+import org.apache.fluss.metadata.MergeEngineType;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.DataTypes;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Integration tests for checkpoint state maintained by {@link UndoRecoveryOperator}. */
+public class UndoRecoveryOperatorStateTest extends FlinkTestBase {
+
+ private static final AtomicInteger TABLE_SEQUENCE = new AtomicInteger();
+ private static final String UNDO_RECOVERY_STATE_NAME = "undo_recovery_state";
+
+ private static final Schema AGG_SCHEMA =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("value", DataTypes.BIGINT(), AggFunctions.SUM())
+ .primaryKey("id")
+ .build();
+
+ @Test
+ void testV1StateMigratesToV2OnNextCheckpoint() throws Exception {
+ AggTable table = createAggTable("v1_migration", 1);
+ UpsertResult write = upsert(table.tablePath, 1, 10L);
+ TableBucket bucket = write.getBucket();
+ long offset = write.getLogEndOffset();
+ WriterState legacyState = new WriterState(Collections.singletonMap(bucket, offset));
+
+ OperatorSubtaskState v1Snapshot;
+ try (StateOperatorHarness seed =
+ new StateOperatorHarness(new StateSeedOperator(legacyState))) {
+ seed.open();
+ v1Snapshot = seed.snapshot(1L, 1L);
+ }
+
+ OperatorSubtaskState migratedSnapshot;
+ try (UndoOperatorHarness migrating =
+ createHarness(table, uniqueProducerId("v1-migration"))) {
+ migrating.initializeState(v1Snapshot);
+ migrating.open();
+ assertThat(migrating.getBucketOffsets()).containsEntry(bucket, offset);
+ migratedSnapshot = migrating.snapshot(2L, 2L);
+ }
+
+ StateInspectOperator inspector = new StateInspectOperator();
+ try (StateOperatorHarness inspect = new StateOperatorHarness(inspector)) {
+ inspect.initializeState(migratedSnapshot);
+ inspect.open();
+ }
+
+ assertThat(inspector.getRestoredStates()).hasSize(1);
+ WriterState migratedState = inspector.getRestoredStates().get(0);
+ assertThat(migratedState.getStateFormat()).isEqualTo(WriterState.StateFormat.V2_COMPLETE);
+ assertThat(migratedState.getTableId()).isEqualTo(table.tableId);
+ assertThat(migratedState.getBucketOffsets()).containsOnlyKeys(bucket);
+ assertThat(migratedState.getBucketOffsets().get(bucket)).isEqualTo(offset);
+ }
+
+ @Test
+ void testSnapshotPreservesOffsetsForUnchangedBuckets() throws Exception {
+ AggTable table = createAggTable("complete_baseline", 2);
+ List writtenRows = writeRowsToDistinctBuckets(table);
+ WrittenRow unchangedRow = writtenRows.get(0);
+ WrittenRow updatedRow = writtenRows.get(1);
+ long unchangedLeo = latestOffset(table.tablePath, unchangedRow.bucket);
+ String producerId = uniqueProducerId("complete-baseline");
+ OperatorSubtaskState snapshot;
+ long reportedLeo;
+
+ UndoRecoveryOperatorFactory initialFactory = createFactory(table, producerId);
+ try (UndoOperatorHarness initial = createHarness(initialFactory)) {
+ initial.open();
+
+ UpsertResult reportedWrite = upsert(table.tablePath, updatedRow.key, 7L);
+ assertThat(reportedWrite.getBucket()).isEqualTo(updatedRow.bucket);
+ reportedLeo = reportedWrite.getLogEndOffset();
+ initialFactory
+ .createProducerOffsetReporter(0)
+ .reportOffset(reportedWrite.getBucket(), reportedLeo);
+ snapshot = initial.snapshot(2L, 2L);
+ }
+
+ try (UndoOperatorHarness restored = createHarness(table, producerId)) {
+ restored.initializeState(snapshot);
+ restored.open();
+
+ assertThat(restored.getBucketOffsets())
+ .hasSize(2)
+ .containsEntry(unchangedRow.bucket, unchangedLeo)
+ .containsEntry(updatedRow.bucket, reportedLeo);
+ assertThat(lookupValue(table.tablePath, unchangedRow.key))
+ .isEqualTo(unchangedRow.value);
+ }
+ }
+
+ private AggTable createAggTable(String prefix, int numBuckets) throws Exception {
+ TablePath tablePath =
+ TablePath.of(DEFAULT_DB, prefix + "_" + TABLE_SEQUENCE.incrementAndGet());
+ long tableId =
+ createTable(
+ tablePath,
+ TableDescriptor.builder()
+ .schema(AGG_SCHEMA)
+ .distributedBy(numBuckets, "id")
+ .property(
+ ConfigOptions.TABLE_MERGE_ENGINE,
+ MergeEngineType.AGGREGATION)
+ .build());
+ FLUSS_CLUSTER_EXTENSION.waitUntilTableReady(tableId);
+ return new AggTable(tablePath, tableId, numBuckets);
+ }
+
+ private static String uniqueProducerId(String prefix) {
+ return "undo-state-" + prefix + "-" + TABLE_SEQUENCE.incrementAndGet();
+ }
+
+ private List writeRowsToDistinctBuckets(AggTable table) throws Exception {
+ List rows = new ArrayList<>();
+ try (Table flussTable = conn.getTable(table.tablePath)) {
+ UpsertWriter writer = flussTable.newUpsert().createWriter();
+ for (int key = 0; key < 100 && rows.size() < table.numBuckets; key++) {
+ long value = 100L + key;
+ CompletableFuture future = writer.upsert(row(key, value));
+ writer.flush();
+ UpsertResult result = future.get();
+ assertThat(result.getBucket()).isNotNull();
+ assertThat(result.getBucket().getTableId()).isEqualTo(table.tableId);
+ if (!containsBucket(rows, result.getBucket())) {
+ rows.add(new WrittenRow(key, value, result.getBucket()));
+ }
+ }
+ }
+ assertThat(rows).hasSize(table.numBuckets);
+ return rows;
+ }
+
+ private static boolean containsBucket(List rows, TableBucket bucket) {
+ for (WrittenRow row : rows) {
+ if (row.bucket.equals(bucket)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private UpsertResult upsert(TablePath tablePath, int key, long value) throws Exception {
+ try (Table table = conn.getTable(tablePath)) {
+ UpsertWriter writer = table.newUpsert().createWriter();
+ CompletableFuture result = writer.upsert(row(key, value));
+ writer.flush();
+ return result.get();
+ }
+ }
+
+ private long lookupValue(TablePath tablePath, int key) throws Exception {
+ try (Table table = conn.getTable(tablePath)) {
+ Lookuper lookuper = table.newLookup().createLookuper();
+ InternalRow result = lookuper.lookup(row(key)).get().getSingletonRow();
+ assertThat(result).isNotNull();
+ return result.getLong(1);
+ }
+ }
+
+ private long latestOffset(TablePath tablePath, TableBucket bucket) throws Exception {
+ Long offset =
+ admin.listOffsets(
+ tablePath,
+ Collections.singletonList(bucket.getBucket()),
+ new OffsetSpec.LatestSpec())
+ .bucketResult(bucket.getBucket())
+ .get();
+ assertThat(offset).isNotNull();
+ return offset;
+ }
+
+ private UndoOperatorHarness createHarness(AggTable table, String producerId) throws Exception {
+ return createHarness(createFactory(table, producerId));
+ }
+
+ private UndoOperatorHarness createHarness(UndoRecoveryOperatorFactory factory)
+ throws Exception {
+ return new UndoOperatorHarness(factory);
+ }
+
+ private UndoRecoveryOperatorFactory createFactory(
+ AggTable table, String producerId) {
+ return new UndoRecoveryOperatorFactory<>(
+ table.tablePath,
+ new Configuration(clientConf),
+ AGG_SCHEMA.getRowType(),
+ null,
+ table.numBuckets,
+ false,
+ producerId);
+ }
+
+ private static final class AggTable {
+ private final TablePath tablePath;
+ private final long tableId;
+ private final int numBuckets;
+
+ private AggTable(TablePath tablePath, long tableId, int numBuckets) {
+ this.tablePath = tablePath;
+ this.tableId = tableId;
+ this.numBuckets = numBuckets;
+ }
+ }
+
+ private static final class WrittenRow {
+ private final int key;
+ private final long value;
+ private final TableBucket bucket;
+
+ private WrittenRow(int key, long value, TableBucket bucket) {
+ this.key = key;
+ this.value = value;
+ this.bucket = bucket;
+ }
+ }
+
+ private static final class UndoOperatorHarness
+ extends OneInputStreamOperatorTestHarness {
+
+ private UndoOperatorHarness(UndoRecoveryOperatorFactory factory)
+ throws Exception {
+ super(factory, 1, 1, 0);
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map getBucketOffsets() {
+ return ((UndoRecoveryOperator) getOperator()).getBucketOffsets();
+ }
+ }
+
+ private static final class StateOperatorHarness
+ extends OneInputStreamOperatorTestHarness {
+
+ private StateOperatorHarness(OneInputStreamOperator operator)
+ throws Exception {
+ super(operator, 1, 1, 0);
+ }
+ }
+
+ private static final class StateSeedOperator extends AbstractStreamOperator
+ implements OneInputStreamOperator {
+
+ private final WriterState state;
+
+ private StateSeedOperator(WriterState state) {
+ this.state = state;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ context.getOperatorStateStore().getUnionListState(stateDescriptor()).add(state);
+ }
+
+ @Override
+ public void processElement(StreamRecord element) {}
+ }
+
+ private static final class StateInspectOperator extends AbstractStreamOperator
+ implements OneInputStreamOperator {
+
+ private final List restoredStates = new ArrayList<>();
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ ListState state =
+ context.getOperatorStateStore().getUnionListState(stateDescriptor());
+ for (WriterState writerState : state.get()) {
+ restoredStates.add(writerState);
+ }
+ }
+
+ private List getRestoredStates() {
+ return restoredStates;
+ }
+
+ @Override
+ public void processElement(StreamRecord element) {}
+ }
+
+ private static ListStateDescriptor stateDescriptor() {
+ return new ListStateDescriptor<>(UNDO_RECOVERY_STATE_NAME, new WriterStateSerializer());
+ }
+}