You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "platinumhamburg (via GitHub)" <gi...@apache.org> on 2026/02/02 03:31:41 UTC
[PR] Add undo recovery support for aggregation tables [fluss]
platinumhamburg opened a new pull request, #2545:
URL: https://github.com/apache/fluss/pull/2545
This PR introduces the undo recovery mechanism for Flink sink writer to handle failure recovery scenarios with aggregation merge engine tables.
Key components:
- ByteArrayWrapper: Utility class for using byte arrays as map keys
- UndoComputer: Computes undo operations by comparing checkpoint state with current log records, supporting both full row and partial update modes
- UndoRecoveryExecutor: Executes undo operations using UpsertWriter
- UndoRecoveryCoordinator: Coordinates the recovery process across buckets, managing log scanning, undo computation, and execution
- BucketRecoveryContext: Holds per-bucket recovery state
The undo recovery works by:
1. Scanning log records from checkpoint offset to current end offset
2. Computing inverse operations for uncommitted records
3. Executing undo operations to restore table state
For partial update mode, INSERT records require full row deletion (not partial column deletion), which is handled by using a separate delete writer with null target columns.
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #2544 2544
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add undo recovery support for aggregation tables [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on PR #2545:
URL: https://github.com/apache/fluss/pull/2545#issuecomment-3869021136
> Btw, it seems the UNDO recovery for aggregate merge engine table relies on the full changelog of the table. However, `table.changelog.image=WAL` will break this and lead to in-correct results. I think we need to forbid such case, a validation check can be done during table creation and throw exception for tables with agg merge engine and WAL mode. We can do that in another issue. What do you think @platinumhamburg ?
Yes, this restriction will be addressed in a separate issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add undo recovery support for aggregation tables [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #2545:
URL: https://github.com/apache/fluss/pull/2545#discussion_r2779027436
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/BucketRecoveryContext.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Encapsulates the recovery state for a single bucket.
+ *
+ * <p>This class tracks:
+ *
+ * <ul>
+ * <li>The bucket being recovered
+ * <li>The checkpoint offset (start point for reading changelog)
+ * <li>The log end offset (end point for reading changelog)
+ * <li>Processed primary keys for deduplication (streaming execution)
+ * <li>Progress tracking during changelog scanning
+ * </ul>
+ */
+public class BucketRecoveryContext {
+
+ private final TableBucket bucket;
+ private final long checkpointOffset;
+ private long logEndOffset;
+
+ private final Set<ByteArrayWrapper> processedKeys;
+ private long lastProcessedOffset;
+ private int totalRecordsProcessed;
+
+ public BucketRecoveryContext(TableBucket bucket, long checkpointOffset) {
+ this.bucket = bucket;
+ this.checkpointOffset = checkpointOffset;
+ this.logEndOffset = -1;
+ this.processedKeys = new HashSet<>();
+ this.lastProcessedOffset = checkpointOffset;
+ this.totalRecordsProcessed = 0;
+ }
+
+ public TableBucket getBucket() {
+ return bucket;
+ }
+
+ public long getCheckpointOffset() {
+ return checkpointOffset;
+ }
+
+ public long getLogEndOffset() {
+ return logEndOffset;
+ }
+
+ public void setLogEndOffset(long logEndOffset) {
+ this.logEndOffset = logEndOffset;
+ }
+
+ public Set<ByteArrayWrapper> getProcessedKeys() {
+ return processedKeys;
+ }
+
+ /**
+ * Checks if this bucket needs recovery.
+ *
+ * @return true if checkpoint offset is less than log end offset
+ */
+ public boolean needsRecovery() {
+ return checkpointOffset < logEndOffset;
+ }
+
+ /**
+ * Checks if changelog scanning is complete for this bucket.
+ *
+ * <p>Complete means either:
+ *
+ * <ul>
+ * <li>No recovery is needed (checkpointOffset >= logEndOffset), or
+ * <li>We have processed all expected records (totalRecordsProcessed >= logEndOffset -
+ * checkpointOffset)
+ * </ul>
+ *
+ * <p>This implementation assumes that changelog offsets are contiguous (no gaps). The number of
+ * records to process equals logEndOffset - checkpointOffset.
Review Comment:
That’s not accurate. There are certain special log record batches that **advance the log offset even though they contain no records**. Examples include:
- The first merge engine batch for a key that is **not the first row** ,
- Delete operations targeting **non-existent rows**.
In both cases, the batch is logically empty from a data perspective but still increments the log offset to maintain consistency and ordering guarantees.
See implementation details in:
https://github.com/apache/fluss/blob/main/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java#L369-L378
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoComputer.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Computes and executes undo operations from changelog records using streaming execution.
+ *
+ * <p>This class uses a {@link KeyEncoder} to encode primary keys as byte arrays for efficient
+ * deduplication, and executes undo operations immediately via {@link UpsertWriter}. The undo logic
+ * is:
+ *
+ * <ul>
+ * <li>{@code INSERT} → Delete the row (it didn't exist at checkpoint)
+ * <li>{@code UPDATE_BEFORE} → Restore the old value (this was the state at checkpoint)
+ * <li>{@code UPDATE_AFTER} → Ignored (UPDATE_BEFORE already handled the undo)
+ * <li>{@code DELETE} → Re-insert the deleted row (it existed at checkpoint)
+ * </ul>
+ *
+ * <p>For partial update mode, two writers are used:
+ *
+ * <ul>
+ * <li>{@code partialWriter} - for UPDATE_BEFORE and DELETE undo (restore/re-insert partial
+ * columns)
+ * <li>{@code deleteWriter} - for INSERT undo (delete entire row, not just partial columns)
+ * </ul>
+ *
+ * <p>For each primary key, only the first change after checkpoint determines the undo action. The
+ * original row from {@link ScanRecord} is directly used without copying, as each ScanRecord
+ * contains an independent row instance.
+ */
+public class UndoComputer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoComputer.class);
+
+ private final KeyEncoder keyEncoder;
+ private final UpsertWriter writer;
+ @Nullable private final UpsertWriter deleteWriter;
+
+ // Track which writer was last used to know when to flush.
+ // This state should be reset before processing each new bucket via resetWriterState().
+ private boolean lastUsedDeleteWriter = false;
+
+ /**
+ * Creates an UndoComputer for full row mode.
+ *
+ * @param keyEncoder the key encoder for primary key deduplication
+ * @param writer the writer for all undo operations
+ */
+ public UndoComputer(KeyEncoder keyEncoder, UpsertWriter writer) {
+ this(keyEncoder, writer, null);
+ }
+
+ /**
+ * Creates an UndoComputer with optional separate delete writer for partial update mode.
+ *
+ * <p>In partial update mode, INSERT undo requires deleting the entire row (not just partial
+ * columns), so a separate full-row delete writer is needed.
+ *
+ * @param keyEncoder the key encoder for primary key deduplication
+ * @param writer the writer for UPDATE_BEFORE and DELETE undo operations
+ * @param deleteWriter optional separate writer for INSERT undo (delete entire row), null for
+ * full row mode
+ */
+ public UndoComputer(
+ KeyEncoder keyEncoder, UpsertWriter writer, @Nullable UpsertWriter deleteWriter) {
+ this.keyEncoder = keyEncoder;
+ this.writer = writer;
+ this.deleteWriter = deleteWriter;
+ }
+
+ /**
+ * Resets the writer state before processing a new bucket.
+ *
+ * <p>This ensures that the lastUsedDeleteWriter flag is reset, preventing incorrect flush
+ * behavior when switching between buckets.
+ */
+ public void resetWriterState() {
+ this.lastUsedDeleteWriter = false;
+ }
+
+ /**
+ * Processes a changelog record and executes the undo operation immediately.
+ *
+ * <p>Only the first change for each primary key triggers an undo operation.
+ *
+ * @param record the changelog record
+ * @param processedKeys set of already processed primary keys for deduplication
+ * @return CompletableFuture for the async write, or null if skipped
+ */
+ @Nullable
+ public CompletableFuture<?> processRecord(
+ ScanRecord record, Set<ByteArrayWrapper> processedKeys) {
+ byte[] encodedKey = keyEncoder.encodeKey(record.getRow());
Review Comment:
We should skip key encoding if it is `UPDATE_AFTER` change type to optimize the performance.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoComputer.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Computes and executes undo operations from changelog records using streaming execution.
+ *
+ * <p>This class uses a {@link KeyEncoder} to encode primary keys as byte arrays for efficient
+ * deduplication, and executes undo operations immediately via {@link UpsertWriter}. The undo logic
+ * is:
+ *
+ * <ul>
+ * <li>{@code INSERT} → Delete the row (it didn't exist at checkpoint)
+ * <li>{@code UPDATE_BEFORE} → Restore the old value (this was the state at checkpoint)
+ * <li>{@code UPDATE_AFTER} → Ignored (UPDATE_BEFORE already handled the undo)
+ * <li>{@code DELETE} → Re-insert the deleted row (it existed at checkpoint)
+ * </ul>
+ *
+ * <p>For partial update mode, two writers are used:
+ *
+ * <ul>
+ * <li>{@code partialWriter} - for UPDATE_BEFORE and DELETE undo (restore/re-insert partial
+ * columns)
+ * <li>{@code deleteWriter} - for INSERT undo (delete entire row, not just partial columns)
+ * </ul>
+ *
+ * <p>For each primary key, only the first change after checkpoint determines the undo action. The
+ * original row from {@link ScanRecord} is directly used without copying, as each ScanRecord
+ * contains an independent row instance.
+ */
+public class UndoComputer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoComputer.class);
+
+ private final KeyEncoder keyEncoder;
+ private final UpsertWriter writer;
+ @Nullable private final UpsertWriter deleteWriter;
+
+ // Track which writer was last used to know when to flush.
+ // This state should be reset before processing each new bucket via resetWriterState().
+ private boolean lastUsedDeleteWriter = false;
+
+ /**
+ * Creates an UndoComputer for full row mode.
+ *
+ * @param keyEncoder the key encoder for primary key deduplication
+ * @param writer the writer for all undo operations
+ */
+ public UndoComputer(KeyEncoder keyEncoder, UpsertWriter writer) {
+ this(keyEncoder, writer, null);
+ }
+
+ /**
+ * Creates an UndoComputer with optional separate delete writer for partial update mode.
+ *
+ * <p>In partial update mode, INSERT undo requires deleting the entire row (not just partial
+ * columns), so a separate full-row delete writer is needed.
+ *
+ * @param keyEncoder the key encoder for primary key deduplication
+ * @param writer the writer for UPDATE_BEFORE and DELETE undo operations
+ * @param deleteWriter optional separate writer for INSERT undo (delete entire row), null for
+ * full row mode
+ */
+ public UndoComputer(
+ KeyEncoder keyEncoder, UpsertWriter writer, @Nullable UpsertWriter deleteWriter) {
+ this.keyEncoder = keyEncoder;
+ this.writer = writer;
+ this.deleteWriter = deleteWriter;
+ }
+
+ /**
+ * Resets the writer state before processing a new bucket.
+ *
+ * <p>This ensures that the lastUsedDeleteWriter flag is reset, preventing incorrect flush
+ * behavior when switching between buckets.
+ */
+ public void resetWriterState() {
+ this.lastUsedDeleteWriter = false;
+ }
+
+ /**
+ * Processes a changelog record and executes the undo operation immediately.
+ *
+ * <p>Only the first change for each primary key triggers an undo operation.
+ *
+ * @param record the changelog record
+ * @param processedKeys set of already processed primary keys for deduplication
+ * @return CompletableFuture for the async write, or null if skipped
+ */
+ @Nullable
+ public CompletableFuture<?> processRecord(
+ ScanRecord record, Set<ByteArrayWrapper> processedKeys) {
+ byte[] encodedKey = keyEncoder.encodeKey(record.getRow());
+ ByteArrayWrapper keyWrapper = new ByteArrayWrapper(encodedKey);
+
+ // Skip if we already processed this key
+ if (processedKeys.contains(keyWrapper)) {
+ return null;
+ }
+
+ CompletableFuture<?> future = computeAndExecuteUndo(record);
+ if (future != null) {
+ processedKeys.add(keyWrapper);
+ }
+ return future;
+ }
+
+ /**
+ * Computes and executes the undo operation for a changelog record.
+ *
+ * <p>UPDATE_AFTER is ignored because UPDATE_BEFORE and UPDATE_AFTER always come in pairs, with
+ * UPDATE_BEFORE appearing first. The undo logic is determined by UPDATE_BEFORE which contains
+ * the old value needed for restoration.
+ *
+ * <p>The original row is directly used without copying because:
+ *
+ * <ul>
+ * <li>Each ScanRecord contains an independent GenericRow instance (created in
+ * CompletedFetch.toScanRecord)
+ * <li>For DELETE operations, UpsertWriter.delete() only needs the primary key fields
+ * <li>For UPSERT operations, the original row contains the exact data to restore
+ * </ul>
+ *
+ * <p>For partial update mode, INSERT undo uses the separate deleteWriter (if provided) to
+ * delete the entire row, not just the partial columns. Before switching between writers, the
+ * previous writer is flushed to avoid mixing records with different target columns in the same
+ * batch.
+ *
+ * @param record the changelog record
+ * @return CompletableFuture for the async write, or null if no action needed (e.g.,
+ * UPDATE_AFTER)
+ */
+ @Nullable
+ private CompletableFuture<?> computeAndExecuteUndo(ScanRecord record) {
+ ChangeType changeType = record.getChangeType();
+ InternalRow row = record.getRow();
+
+ switch (changeType) {
+ case INSERT:
+ // Row was inserted after checkpoint → delete it
+ // Use deleteWriter if available (for partial update mode to delete entire row)
+ if (deleteWriter != null) {
+ // Flush partial writer before using delete writer to avoid mixing
+ // records with different target columns in the same batch
+ if (!lastUsedDeleteWriter) {
+ writer.flush();
+ lastUsedDeleteWriter = true;
+ }
+ return deleteWriter.delete(row);
+ }
Review Comment:
If we don't need the full row `deleteWriter`, we don't need this complex logic.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryCoordinator.java:
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.Table;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.writer.Upsert;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.rpc.protocol.MergeMode;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinates undo recovery operations during Flink sink writer initialization.
+ *
+ * <p>This coordinator ensures exactly-once semantics by reversing any writes that occurred after
+ * the last successful checkpoint but before a failure. The recovery process involves:
+ *
+ * <ol>
+ * <li>Reading changelog records from checkpoint offset to current latest offset
+ * <li>Computing inverse operations for each affected primary key
+ * <li>Applying undo operations using OVERWRITE mode to restore bucket state
+ * </ol>
+ *
+ * <p><b>Undo Logic:</b> For each primary key, only the first change after checkpoint determines the
+ * undo action:
+ *
+ * <ul>
+ * <li>{@code INSERT} → Delete the row (it didn't exist at checkpoint)
+ * <li>{@code UPDATE_BEFORE} → Restore the old value (this was the state at checkpoint)
+ * <li>{@code UPDATE_AFTER} → Ignored (UPDATE_BEFORE already handled the undo)
+ * <li>{@code DELETE} → Re-insert the deleted row (it existed at checkpoint)
+ * </ul>
+ *
+ * <p>This class delegates to specialized components:
+ *
+ * <ul>
+ * <li>{@link UndoComputer} - Undo operation computation using {@link KeyEncoder} for primary key
+ * encoding
+ * <li>{@link UndoRecoveryExecutor} - Changelog reading and write execution
+ * </ul>
+ *
+ * @see MergeMode#OVERWRITE
+ */
+public class UndoRecoveryCoordinator {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryCoordinator.class);
+
+ private final Table table;
+ @Nullable private final int[] targetColumnIndexes;
+
+ /**
+ * Creates a new UndoRecoveryCoordinator.
+ *
+ * @param table the Fluss table to perform recovery on
+ * @param targetColumnIndexes optional target columns for partial update (null for full row)
+ */
+ public UndoRecoveryCoordinator(Table table, @Nullable int[] targetColumnIndexes) {
+ this.table = table;
+ this.targetColumnIndexes = targetColumnIndexes;
+
+ logPartialUpdateConfig(targetColumnIndexes);
+ }
+
+ private void logPartialUpdateConfig(@Nullable int[] targetColumnIndexes) {
+ if (targetColumnIndexes != null) {
+ LOG.debug(
+ "Undo recovery configured with partial update columns: {}",
+ Arrays.toString(targetColumnIndexes));
+ }
+ }
+
+ // ==================== Public API ====================
+
+ /**
+ * Performs undo recovery for buckets with known undo offsets.
+ *
+ * <p>The caller is responsible for providing both the checkpoint offset (start) and the log end
+ * offset (end) for each bucket. This avoids redundant listOffset calls since the caller
+ * typically already has this information from determining which buckets need recovery.
+ *
+ * @param bucketUndoOffsets map of bucket to its undo offsets (checkpointOffset, logEndOffset)
+ * @param subtaskIndex the Flink subtask index (for logging)
+ * @param parallelism the total parallelism (for logging)
+ * @throws Exception if recovery fails
+ */
+ public void performUndoRecovery(
+ Map<TableBucket, UndoOffsets> bucketUndoOffsets, int subtaskIndex, int parallelism)
+ throws Exception {
+
+ if (bucketUndoOffsets.isEmpty()) {
+ LOG.debug("No buckets to recover on subtask {}/{}", subtaskIndex, parallelism);
+ return;
+ }
+
+ LOG.info(
+ "Starting undo recovery for {} bucket(s) on subtask {}/{}",
+ bucketUndoOffsets.size(),
+ subtaskIndex,
+ parallelism);
+
+ List<BucketRecoveryContext> contexts = buildRecoveryContexts(bucketUndoOffsets);
+
+ try (LogScanner scanner = createLogScanner()) {
+ Upsert recoveryUpsert = table.newUpsert().mergeMode(MergeMode.OVERWRITE);
+ if (targetColumnIndexes != null) {
+ recoveryUpsert = recoveryUpsert.partialUpdate(targetColumnIndexes);
+ }
+ UpsertWriter writer = recoveryUpsert.createWriter();
+
+ // For partial update mode, create a separate delete writer for INSERT undo
+ // INSERT undo needs to delete the entire row, not just partial columns
+ UpsertWriter deleteWriter = null;
+ if (targetColumnIndexes != null) {
+ deleteWriter = table.newUpsert().mergeMode(MergeMode.OVERWRITE).createWriter();
+ }
+
+ // Create UndoComputer with writer for streaming execution
+ Schema schema = table.getTableInfo().getSchema();
+ KeyEncoder keyEncoder =
+ KeyEncoder.of(
+ schema.getRowType(),
+ schema.getPrimaryKey().get().getColumnNames(),
+ null);
+ UndoComputer undoComputer = new UndoComputer(keyEncoder, writer, deleteWriter);
+
+ UndoRecoveryExecutor executor = new UndoRecoveryExecutor(scanner, writer, undoComputer);
+ executor.execute(contexts);
Review Comment:
nit: add comment to highligh this line it is a blocking operation which is very heavy and time consuming.
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryCoordinatorITCase.java:
##########
@@ -0,0 +1,977 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.client.admin.ListOffsetsResult;
+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.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+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.flink.sink.writer.undo.UndoRecoveryCoordinator.UndoOffsets;
+import org.apache.fluss.flink.utils.FlinkTestBase;
+import org.apache.fluss.metadata.MergeEngineType;
+import org.apache.fluss.metadata.PartitionInfo;
+import org.apache.fluss.metadata.PartitionSpec;
+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.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Integration tests for {@link UndoRecoveryCoordinator}.
+ *
+ * <p>Tests use complex multi-bucket, multi-key scenarios to ensure only correct implementations
+ * pass. Each test covers multiple aspects to maximize value while minimizing redundancy.
+ */
+public class UndoRecoveryCoordinatorITCase extends FlinkTestBase {
+
+ private static final int NUM_BUCKETS = 4;
+
+ private static final Schema TEST_SCHEMA =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("price", DataTypes.INT())
+ .column("stock", DataTypes.INT())
+ .primaryKey("id")
+ .build();
+
+ private static final Schema PARTITIONED_SCHEMA =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("price", DataTypes.INT())
+ .column("pt", DataTypes.STRING())
+ .primaryKey("id", "pt")
+ .build();
+
+ // ==================== Full Update Tests ====================
+
+ /**
+ * Comprehensive test for Full Update mode with 4 buckets and 100+ keys.
+ *
+ * <p>Covers: zero-offset recovery, DELETE undo, UPDATE undo with deduplication, mixed ops.
+ */
+ @Test
+ void testFullUpdateMultiBucketComprehensiveRecovery() throws Exception {
+ TablePath tablePath = createTestTablePath("full_update_test");
+ long tableId =
+ createTable(
+ tablePath,
+ TableDescriptor.builder()
+ .schema(TEST_SCHEMA)
+ .distributedBy(NUM_BUCKETS, "id")
+ .property(
+ ConfigOptions.TABLE_MERGE_ENGINE,
+ MergeEngineType.AGGREGATION)
+ .build());
+
+ try (Table table = conn.getTable(tablePath)) {
+ UpsertWriter writer = table.newUpsert().createWriter();
+ Lookuper lookuper = table.newLookup().createLookuper();
+ BucketTracker tracker = new BucketTracker(NUM_BUCKETS, tableId);
+
+ // Phase 1: Write initial 100 keys (batch async)
+ List<CompletableFuture<UpsertResult>> writeFutures = new ArrayList<>();
+ for (int id = 0; id < 100; id++) {
+ writeFutures.add(writer.upsert(row(id, "pre_" + id, id * 10, id)));
+ }
+ for (int i = 0; i < writeFutures.size(); i++) {
+ tracker.recordWrite(i, writeFutures.get(i).get());
+ }
+ writer.flush();
+
+ int zeroOffsetBucket = tracker.selectBucketForZeroOffsetRecovery();
+ Map<TableBucket, Long> checkpointOffsets =
+ tracker.buildCheckpointOffsetsWithZeroOffset(zeroOffsetBucket);
+
+ Map<Integer, Set<Integer>> keysAfterCheckpoint = new HashMap<>();
+ for (int i = 0; i < NUM_BUCKETS; i++) {
+ keysAfterCheckpoint.put(i, new HashSet<>());
+ }
+
+ // Phase 2: Operations after checkpoint (batch async)
+ List<CompletableFuture<UpsertResult>> newKeyFutures = new ArrayList<>();
+ for (int id = 200; id < 300; id++) {
+ newKeyFutures.add(writer.upsert(row(id, "new_" + id, id * 10, id)));
+ }
+ for (int i = 0; i < newKeyFutures.size(); i++) {
Review Comment:
nit: add `writer.flush();` before this line to actively flush data, rather than waiting for async flush.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryCoordinator.java:
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.Table;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.writer.Upsert;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.rpc.protocol.MergeMode;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinates undo recovery operations during Flink sink writer initialization.
+ *
+ * <p>This coordinator ensures exactly-once semantics by reversing any writes that occurred after
+ * the last successful checkpoint but before a failure. The recovery process involves:
+ *
+ * <ol>
+ * <li>Reading changelog records from checkpoint offset to current latest offset
+ * <li>Computing inverse operations for each affected primary key
+ * <li>Applying undo operations using OVERWRITE mode to restore bucket state
+ * </ol>
+ *
+ * <p><b>Undo Logic:</b> For each primary key, only the first change after checkpoint determines the
+ * undo action:
+ *
+ * <ul>
+ * <li>{@code INSERT} → Delete the row (it didn't exist at checkpoint)
+ * <li>{@code UPDATE_BEFORE} → Restore the old value (this was the state at checkpoint)
+ * <li>{@code UPDATE_AFTER} → Ignored (UPDATE_BEFORE already handled the undo)
+ * <li>{@code DELETE} → Re-insert the deleted row (it existed at checkpoint)
+ * </ul>
+ *
+ * <p>This class delegates to specialized components:
+ *
+ * <ul>
+ * <li>{@link UndoComputer} - Undo operation computation using {@link KeyEncoder} for primary key
+ * encoding
+ * <li>{@link UndoRecoveryExecutor} - Changelog reading and write execution
+ * </ul>
+ *
+ * @see MergeMode#OVERWRITE
+ */
+public class UndoRecoveryCoordinator {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryCoordinator.class);
+
+ private final Table table;
+ @Nullable private final int[] targetColumnIndexes;
+
+ /**
+ * Creates a new UndoRecoveryCoordinator.
+ *
+ * @param table the Fluss table to perform recovery on
+ * @param targetColumnIndexes optional target columns for partial update (null for full row)
+ */
+ public UndoRecoveryCoordinator(Table table, @Nullable int[] targetColumnIndexes) {
+ this.table = table;
+ this.targetColumnIndexes = targetColumnIndexes;
+
+ logPartialUpdateConfig(targetColumnIndexes);
+ }
+
+ private void logPartialUpdateConfig(@Nullable int[] targetColumnIndexes) {
+ if (targetColumnIndexes != null) {
+ LOG.debug(
+ "Undo recovery configured with partial update columns: {}",
+ Arrays.toString(targetColumnIndexes));
+ }
+ }
+
+ // ==================== Public API ====================
+
+ /**
+ * Performs undo recovery for buckets with known undo offsets.
+ *
+ * <p>The caller is responsible for providing both the checkpoint offset (start) and the log end
+ * offset (end) for each bucket. This avoids redundant listOffset calls since the caller
+ * typically already has this information from determining which buckets need recovery.
+ *
+ * @param bucketUndoOffsets map of bucket to its undo offsets (checkpointOffset, logEndOffset)
+ * @param subtaskIndex the Flink subtask index (for logging)
+ * @param parallelism the total parallelism (for logging)
+ * @throws Exception if recovery fails
+ */
+ public void performUndoRecovery(
+ Map<TableBucket, UndoOffsets> bucketUndoOffsets, int subtaskIndex, int parallelism)
+ throws Exception {
+
+ if (bucketUndoOffsets.isEmpty()) {
+ LOG.debug("No buckets to recover on subtask {}/{}", subtaskIndex, parallelism);
+ return;
+ }
+
+ LOG.info(
+ "Starting undo recovery for {} bucket(s) on subtask {}/{}",
+ bucketUndoOffsets.size(),
+ subtaskIndex,
+ parallelism);
+
+ List<BucketRecoveryContext> contexts = buildRecoveryContexts(bucketUndoOffsets);
+
+ try (LogScanner scanner = createLogScanner()) {
+ Upsert recoveryUpsert = table.newUpsert().mergeMode(MergeMode.OVERWRITE);
+ if (targetColumnIndexes != null) {
+ recoveryUpsert = recoveryUpsert.partialUpdate(targetColumnIndexes);
+ }
+ UpsertWriter writer = recoveryUpsert.createWriter();
+
+ // For partial update mode, create a separate delete writer for INSERT undo
+ // INSERT undo needs to delete the entire row, not just partial columns
+ UpsertWriter deleteWriter = null;
+ if (targetColumnIndexes != null) {
+ deleteWriter = table.newUpsert().mergeMode(MergeMode.OVERWRITE).createWriter();
+ }
+
+ // Create UndoComputer with writer for streaming execution
+ Schema schema = table.getTableInfo().getSchema();
+ KeyEncoder keyEncoder =
+ KeyEncoder.of(
+ schema.getRowType(),
+ schema.getPrimaryKey().get().getColumnNames(),
+ null);
+ UndoComputer undoComputer = new UndoComputer(keyEncoder, writer, deleteWriter);
+
+ UndoRecoveryExecutor executor = new UndoRecoveryExecutor(scanner, writer, undoComputer);
+ executor.execute(contexts);
+
+ // Flush both writers to ensure all writes complete
+ // Note: UpsertWriter does not have a close() method, flush() is sufficient
+ // to ensure all pending writes are sent to the server
+ writer.flush();
+ if (deleteWriter != null) {
+ deleteWriter.flush();
+ }
+ }
+
+ // Calculate total undo operations for summary
+ int totalUndoOps = 0;
+ for (BucketRecoveryContext ctx : contexts) {
+ totalUndoOps += ctx.getProcessedKeys().size();
+ }
+
+ LOG.info(
+ "Completed undo recovery for {} bucket(s) with {} undo operation(s) on subtask {}/{}",
+ bucketUndoOffsets.size(),
+ totalUndoOps,
+ subtaskIndex,
+ parallelism);
+ }
+
+ /**
+ * Encapsulates the offset information needed for undo recovery of a bucket.
+ *
+ * <p>This class holds the checkpoint offset (recovery start point) and log end offset (recovery
+ * end point) for a bucket's undo recovery operation.
+ */
+ public static class UndoOffsets {
+ private final long checkpointOffset;
+ private final long logEndOffset;
+
+ /**
+ * Creates a new UndoOffsets.
+ *
+ * @param checkpointOffset the checkpoint offset (start point, inclusive)
+ * @param logEndOffset the log end offset (end point, exclusive)
+ * @throws IllegalArgumentException if offsets are negative or checkpointOffset >
+ * logEndOffset
+ */
+ public UndoOffsets(long checkpointOffset, long logEndOffset) {
+ if (checkpointOffset < 0) {
+ throw new IllegalArgumentException(
+ "checkpointOffset must be non-negative: " + checkpointOffset);
+ }
+ if (logEndOffset < 0) {
+ throw new IllegalArgumentException(
+ "logEndOffset must be non-negative: " + logEndOffset);
+ }
+ if (checkpointOffset > logEndOffset) {
+ throw new IllegalArgumentException(
+ String.format(
+ "checkpointOffset (%d) must not be greater than logEndOffset (%d)",
+ checkpointOffset, logEndOffset));
+ }
+ // Note: checkpointOffset == logEndOffset is allowed, it simply means no recovery needed
+ this.checkpointOffset = checkpointOffset;
+ this.logEndOffset = logEndOffset;
+ }
+
+ public long getCheckpointOffset() {
+ return checkpointOffset;
+ }
+
+ public long getLogEndOffset() {
+ return logEndOffset;
+ }
+ }
+
+ // ==================== Scanner Factory ====================
+
+ /**
+ * Creates a LogScanner for reading changelog records.
+ *
+ * <p>This method is protected to allow test subclasses to inject custom scanners for fault
+ * injection testing.
+ *
+ * @return a new LogScanner instance
+ */
+ protected LogScanner createLogScanner() {
+ return table.newScan().createLogScanner();
+ }
+
+ // ==================== Recovery Context Building ====================
+
+ private List<BucketRecoveryContext> buildRecoveryContexts(
+ Map<TableBucket, UndoOffsets> bucketUndoOffsets) {
+
+ List<BucketRecoveryContext> contexts = new ArrayList<>();
+
+ for (Map.Entry<TableBucket, UndoOffsets> entry : bucketUndoOffsets.entrySet()) {
+ TableBucket bucket = entry.getKey();
+ UndoOffsets offsets = entry.getValue();
+
+ BucketRecoveryContext ctx =
+ new BucketRecoveryContext(bucket, offsets.getCheckpointOffset());
+ ctx.setLogEndOffset(offsets.getLogEndOffset());
Review Comment:
It seems there is no need to **update** LogEndOffset for the context. So we can change it to a `final` field and initialize it in constructor.
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryCoordinatorITCase.java:
##########
@@ -0,0 +1,977 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.client.admin.ListOffsetsResult;
+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.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+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.flink.sink.writer.undo.UndoRecoveryCoordinator.UndoOffsets;
+import org.apache.fluss.flink.utils.FlinkTestBase;
+import org.apache.fluss.metadata.MergeEngineType;
+import org.apache.fluss.metadata.PartitionInfo;
+import org.apache.fluss.metadata.PartitionSpec;
+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.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Integration tests for {@link UndoRecoveryCoordinator}.
+ *
+ * <p>Tests use complex multi-bucket, multi-key scenarios to ensure only correct implementations
+ * pass. Each test covers multiple aspects to maximize value while minimizing redundancy.
+ */
+public class UndoRecoveryCoordinatorITCase extends FlinkTestBase {
Review Comment:
We don't need to extend `FlinkTestBase`, because there is no Flink cluster needed in this IT case.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoComputer.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Computes and executes undo operations from changelog records using streaming execution.
+ *
+ * <p>This class uses a {@link KeyEncoder} to encode primary keys as byte arrays for efficient
+ * deduplication, and executes undo operations immediately via {@link UpsertWriter}. The undo logic
+ * is:
+ *
+ * <ul>
+ * <li>{@code INSERT} → Delete the row (it didn't exist at checkpoint)
+ * <li>{@code UPDATE_BEFORE} → Restore the old value (this was the state at checkpoint)
+ * <li>{@code UPDATE_AFTER} → Ignored (UPDATE_BEFORE already handled the undo)
+ * <li>{@code DELETE} → Re-insert the deleted row (it existed at checkpoint)
+ * </ul>
+ *
+ * <p>For partial update mode, two writers are used:
+ *
+ * <ul>
+ * <li>{@code partialWriter} - for UPDATE_BEFORE and DELETE undo (restore/re-insert partial
+ * columns)
+ * <li>{@code deleteWriter} - for INSERT undo (delete entire row, not just partial columns)
+ * </ul>
Review Comment:
I don’t understand why a full-row delete writer is needed for partial update recovery. Let me illustrate with an example:
1. **Initial state**:
- Job-1 upserts into the table on columns `(id, name)` with records `(1, "user1")` and `(2, "user2")`.
- Job-2 upserts on columns `(id, price)` with records `(1, 100)` and `(2, 200)`.
→ The resulting table rows are:
`(1, "user1", 100)` and `(2, "user2", 200)`.
2. **After Job-1 fails over and recovers**:
To roll back Job-1’s effects, we apply its undo log (i.e., set `name = null` for IDs 1 and 2).
The expected result should be:
`(1, null, 100)` and `(2, null, 200)` — **not an empty table**.
However, according to the current Javadoc, the outcome would incorrectly be an **empty table**, which is clearly wrong.
Note that **partial deletes already handle row removal correctly**: if all non-primary-key columns become `null`, the row is automatically deleted. Therefore:
- If **only Job-1 existed** (no Job-2), rolling back to a prior checkpoint via partial delete would correctly remove the entire row—**no full-row delete writer is needed**.
- But when **multiple jobs update different column subsets**, we must preserve columns written by other jobs during recovery. A full-row delete would erroneously wipe out those contributions.
Thus, using a full-row delete writer in partial update recovery breaks correctness in multi-writer scenarios. The existing partial delete mechanism, which respects column-level null semantics, is sufficient and more accurate.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryExecutor.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Executes undo recovery for multiple buckets using streaming execution.
+ *
+ * <p>This executor manages:
+ *
+ * <ul>
+ * <li>Single LogScanner with multiple bucket subscriptions
+ * <li>Changelog reading with immediate undo operation execution
+ * <li>Async write operations via UpsertWriter
+ * </ul>
+ *
+ * <p>The execution flow (streaming):
+ *
+ * <ol>
+ * <li>Subscribe all buckets to the LogScanner
+ * <li>Poll changelog records and execute undo operations immediately
+ * <li>Collect CompletableFutures for completion tracking
+ * <li>Wait for all futures and flush to ensure all writes complete
+ * </ol>
+ */
+public class UndoRecoveryExecutor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryExecutor.class);
+
+ /** Default initial poll timeout in milliseconds. */
+ private static final long DEFAULT_INITIAL_POLL_TIMEOUT_MS = 100;
+
+ /** Default maximum poll timeout in milliseconds after exponential backoff. */
+ private static final long DEFAULT_MAX_POLL_TIMEOUT_MS = 30_000; // 30 seconds
+
+ /** Default maximum total wait time in milliseconds before failing (1 hour). */
+ private static final long DEFAULT_MAX_TOTAL_WAIT_TIME_MS = 60 * 60 * 1000; // 1 hour
+
+ /** Interval for logging progress during long waits (5 minutes). */
+ private static final long PROGRESS_LOG_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
+
+ private final LogScanner scanner;
+ private final UpsertWriter writer;
+ private final UndoComputer undoComputer;
+
+ // Configurable timeout parameters
+ private final long initialPollTimeoutMs;
+ private final long maxPollTimeoutMs;
+ private final long maxTotalWaitTimeMs;
+
+ public UndoRecoveryExecutor(
+ LogScanner scanner, UpsertWriter writer, UndoComputer undoComputer) {
+ this(
+ scanner,
+ writer,
+ undoComputer,
+ DEFAULT_INITIAL_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_TOTAL_WAIT_TIME_MS);
+ }
+
+ /**
+ * Creates an executor with custom timeout configuration (for testing).
+ *
+ * @param scanner the log scanner
+ * @param writer the upsert writer
+ * @param undoComputer the undo computer
+ * @param initialPollTimeoutMs initial poll timeout in milliseconds
+ * @param maxPollTimeoutMs maximum poll timeout after exponential backoff
+ * @param maxTotalWaitTimeMs maximum total wait time before failing
+ */
+ public UndoRecoveryExecutor(
+ LogScanner scanner,
+ UpsertWriter writer,
+ UndoComputer undoComputer,
+ long initialPollTimeoutMs,
+ long maxPollTimeoutMs,
+ long maxTotalWaitTimeMs) {
+ this.scanner = scanner;
+ this.writer = writer;
+ this.undoComputer = undoComputer;
+ this.initialPollTimeoutMs = initialPollTimeoutMs;
+ this.maxPollTimeoutMs = maxPollTimeoutMs;
+ this.maxTotalWaitTimeMs = maxTotalWaitTimeMs;
+ }
+
+ /**
+ * Executes undo recovery for the given bucket contexts.
+ *
+ * @param contexts the bucket recovery contexts (must have target offsets set)
+ * @throws Exception if recovery fails
+ */
+ public void execute(List<BucketRecoveryContext> contexts) throws Exception {
+ // Filter contexts that need recovery
+ List<BucketRecoveryContext> toRecover = filterContextsNeedingRecovery(contexts);
+ if (toRecover.isEmpty()) {
+ LOG.debug("No buckets need recovery after filtering");
+ return;
+ }
+
+ LOG.debug("Executing undo recovery for {} bucket(s)", toRecover.size());
+
+ // Subscribe and read changelog with streaming execution
+ subscribeAll(toRecover);
+ List<CompletableFuture<?>> allFutures = readChangelogAndExecute(toRecover);
+
+ // Wait for all async writes to complete
+ if (!allFutures.isEmpty()) {
+ CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[0])).get();
+ }
+ writer.flush();
+
+ // Log summary at INFO level for visibility
+ int totalUndoOps = 0;
+ for (BucketRecoveryContext ctx : toRecover) {
+ totalUndoOps += ctx.getProcessedKeys().size();
+ }
+
+ LOG.debug(
+ "Undo recovery execution completed: {} bucket(s), {} total undo operation(s)",
+ toRecover.size(),
+ totalUndoOps);
+ }
+
+ private List<BucketRecoveryContext> filterContextsNeedingRecovery(
+ List<BucketRecoveryContext> contexts) {
+ List<BucketRecoveryContext> result = new ArrayList<>();
+ for (BucketRecoveryContext ctx : contexts) {
+ if (ctx.needsRecovery()) {
+ LOG.debug(
+ "Bucket {} needs recovery: checkpoint={}, logEndOffset={}",
+ ctx.getBucket(),
+ ctx.getCheckpointOffset(),
+ ctx.getLogEndOffset());
+ result.add(ctx);
+ } else {
+ LOG.debug("Bucket {} already up-to-date, no recovery needed", ctx.getBucket());
+ }
+ }
+ return result;
+ }
+
+ private void subscribeAll(List<BucketRecoveryContext> contexts) {
+ for (BucketRecoveryContext ctx : contexts) {
+ TableBucket bucket = ctx.getBucket();
+ if (bucket.getPartitionId() != null) {
+ scanner.subscribe(
+ bucket.getPartitionId(), bucket.getBucket(), ctx.getCheckpointOffset());
+ } else {
+ scanner.subscribe(bucket.getBucket(), ctx.getCheckpointOffset());
+ }
+ }
+ }
+
+ private List<CompletableFuture<?>> readChangelogAndExecute(List<BucketRecoveryContext> contexts)
+ throws Exception {
+ List<CompletableFuture<?>> allFutures = new ArrayList<>();
+
+ // Early exit if all contexts are already complete (no records to read)
+ if (allComplete(contexts)) {
+ LOG.debug("All buckets already complete, no changelog reading needed");
+ return allFutures;
+ }
+
+ long currentPollTimeoutMs = initialPollTimeoutMs;
+ long totalWaitTimeMs = 0;
+ long lastProgressLogTime = System.currentTimeMillis();
+
+ while (!allComplete(contexts)) {
+ ScanRecords records = scanner.poll(Duration.ofMillis(currentPollTimeoutMs));
+
+ if (records.isEmpty()) {
+ totalWaitTimeMs += currentPollTimeoutMs;
+
+ // Check if we've exceeded the maximum total wait time
+ if (totalWaitTimeMs >= maxTotalWaitTimeMs) {
+ // Undo recovery failure is fatal - incomplete undo leads to data inconsistency
+ String incompleteBucketsDetail = getIncompleteBucketsDetail(contexts);
+ throw new RuntimeException(
+ String.format(
+ "Undo recovery failed: unable to read all changelog records after "
+ + "%.1f minutes of waiting. Incomplete buckets: %s. "
+ + "This is a fatal error as incomplete undo recovery "
+ + "would lead to data inconsistency.",
+ totalWaitTimeMs / 60000.0, incompleteBucketsDetail));
+ }
+
+ // Log progress periodically during long waits
+ long now = System.currentTimeMillis();
+ if (now - lastProgressLogTime >= PROGRESS_LOG_INTERVAL_MS) {
+ LOG.info(
Review Comment:
This logging is useful; however, it seems to print only when `records` is empty, which makes the logged information highly unpredictable and hard to control.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryExecutor.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Executes undo recovery for multiple buckets using streaming execution.
+ *
+ * <p>This executor manages:
+ *
+ * <ul>
+ * <li>Single LogScanner with multiple bucket subscriptions
+ * <li>Changelog reading with immediate undo operation execution
+ * <li>Async write operations via UpsertWriter
+ * </ul>
+ *
+ * <p>The execution flow (streaming):
+ *
+ * <ol>
+ * <li>Subscribe all buckets to the LogScanner
+ * <li>Poll changelog records and execute undo operations immediately
+ * <li>Collect CompletableFutures for completion tracking
+ * <li>Wait for all futures and flush to ensure all writes complete
+ * </ol>
+ */
+public class UndoRecoveryExecutor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryExecutor.class);
+
+ /** Default initial poll timeout in milliseconds. */
+ private static final long DEFAULT_INITIAL_POLL_TIMEOUT_MS = 100;
+
+ /** Default maximum poll timeout in milliseconds after exponential backoff. */
+ private static final long DEFAULT_MAX_POLL_TIMEOUT_MS = 30_000; // 30 seconds
+
+ /** Default maximum total wait time in milliseconds before failing (1 hour). */
+ private static final long DEFAULT_MAX_TOTAL_WAIT_TIME_MS = 60 * 60 * 1000; // 1 hour
+
+ /** Interval for logging progress during long waits (5 minutes). */
+ private static final long PROGRESS_LOG_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
+
+ private final LogScanner scanner;
+ private final UpsertWriter writer;
+ private final UndoComputer undoComputer;
+
+ // Configurable timeout parameters
+ private final long initialPollTimeoutMs;
+ private final long maxPollTimeoutMs;
+ private final long maxTotalWaitTimeMs;
+
+ public UndoRecoveryExecutor(
+ LogScanner scanner, UpsertWriter writer, UndoComputer undoComputer) {
+ this(
+ scanner,
+ writer,
+ undoComputer,
+ DEFAULT_INITIAL_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_TOTAL_WAIT_TIME_MS);
+ }
+
+ /**
+ * Creates an executor with custom timeout configuration (for testing).
+ *
+ * @param scanner the log scanner
+ * @param writer the upsert writer
+ * @param undoComputer the undo computer
+ * @param initialPollTimeoutMs initial poll timeout in milliseconds
+ * @param maxPollTimeoutMs maximum poll timeout after exponential backoff
+ * @param maxTotalWaitTimeMs maximum total wait time before failing
+ */
+ public UndoRecoveryExecutor(
+ LogScanner scanner,
+ UpsertWriter writer,
+ UndoComputer undoComputer,
+ long initialPollTimeoutMs,
+ long maxPollTimeoutMs,
+ long maxTotalWaitTimeMs) {
+ this.scanner = scanner;
+ this.writer = writer;
+ this.undoComputer = undoComputer;
+ this.initialPollTimeoutMs = initialPollTimeoutMs;
+ this.maxPollTimeoutMs = maxPollTimeoutMs;
+ this.maxTotalWaitTimeMs = maxTotalWaitTimeMs;
+ }
+
+ /**
+ * Executes undo recovery for the given bucket contexts.
+ *
+ * @param contexts the bucket recovery contexts (must have target offsets set)
+ * @throws Exception if recovery fails
+ */
+ public void execute(List<BucketRecoveryContext> contexts) throws Exception {
+ // Filter contexts that need recovery
+ List<BucketRecoveryContext> toRecover = filterContextsNeedingRecovery(contexts);
+ if (toRecover.isEmpty()) {
+ LOG.debug("No buckets need recovery after filtering");
+ return;
+ }
+
+ LOG.debug("Executing undo recovery for {} bucket(s)", toRecover.size());
+
+ // Subscribe and read changelog with streaming execution
+ subscribeAll(toRecover);
+ List<CompletableFuture<?>> allFutures = readChangelogAndExecute(toRecover);
+
+ // Wait for all async writes to complete
+ if (!allFutures.isEmpty()) {
+ CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[0])).get();
+ }
+ writer.flush();
+
+ // Log summary at INFO level for visibility
+ int totalUndoOps = 0;
+ for (BucketRecoveryContext ctx : toRecover) {
+ totalUndoOps += ctx.getProcessedKeys().size();
+ }
+
+ LOG.debug(
+ "Undo recovery execution completed: {} bucket(s), {} total undo operation(s)",
+ toRecover.size(),
+ totalUndoOps);
+ }
+
+ private List<BucketRecoveryContext> filterContextsNeedingRecovery(
+ List<BucketRecoveryContext> contexts) {
+ List<BucketRecoveryContext> result = new ArrayList<>();
+ for (BucketRecoveryContext ctx : contexts) {
+ if (ctx.needsRecovery()) {
+ LOG.debug(
+ "Bucket {} needs recovery: checkpoint={}, logEndOffset={}",
+ ctx.getBucket(),
+ ctx.getCheckpointOffset(),
+ ctx.getLogEndOffset());
+ result.add(ctx);
+ } else {
+ LOG.debug("Bucket {} already up-to-date, no recovery needed", ctx.getBucket());
+ }
+ }
+ return result;
+ }
+
+ private void subscribeAll(List<BucketRecoveryContext> contexts) {
+ for (BucketRecoveryContext ctx : contexts) {
+ TableBucket bucket = ctx.getBucket();
+ if (bucket.getPartitionId() != null) {
+ scanner.subscribe(
+ bucket.getPartitionId(), bucket.getBucket(), ctx.getCheckpointOffset());
+ } else {
+ scanner.subscribe(bucket.getBucket(), ctx.getCheckpointOffset());
+ }
+ }
+ }
+
+ private List<CompletableFuture<?>> readChangelogAndExecute(List<BucketRecoveryContext> contexts)
+ throws Exception {
+ List<CompletableFuture<?>> allFutures = new ArrayList<>();
+
+ // Early exit if all contexts are already complete (no records to read)
+ if (allComplete(contexts)) {
+ LOG.debug("All buckets already complete, no changelog reading needed");
+ return allFutures;
+ }
+
+ long currentPollTimeoutMs = initialPollTimeoutMs;
+ long totalWaitTimeMs = 0;
+ long lastProgressLogTime = System.currentTimeMillis();
+
+ while (!allComplete(contexts)) {
+ ScanRecords records = scanner.poll(Duration.ofMillis(currentPollTimeoutMs));
+
+ if (records.isEmpty()) {
+ totalWaitTimeMs += currentPollTimeoutMs;
Review Comment:
I think the calculation of `totalWaitTimeMs` is not correct.
1. The `currentPollTimeoutMs` is not the actual waiting time of the `scanner.poll` method. Because, if there is already data fetched in scanner, it will return directly without waiting.
2. The `totalWaitTimeMs` is only updated when `records.isEmpty()`. It is possible to wait some time to fetch a record, so when records is not empty, it may cost serveral seconds to wait.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryExecutor.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Executes undo recovery for multiple buckets using streaming execution.
+ *
+ * <p>This executor manages:
+ *
+ * <ul>
+ * <li>Single LogScanner with multiple bucket subscriptions
+ * <li>Changelog reading with immediate undo operation execution
+ * <li>Async write operations via UpsertWriter
+ * </ul>
+ *
+ * <p>The execution flow (streaming):
+ *
+ * <ol>
+ * <li>Subscribe all buckets to the LogScanner
+ * <li>Poll changelog records and execute undo operations immediately
+ * <li>Collect CompletableFutures for completion tracking
+ * <li>Wait for all futures and flush to ensure all writes complete
+ * </ol>
+ */
+public class UndoRecoveryExecutor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryExecutor.class);
+
+ /** Default initial poll timeout in milliseconds. */
+ private static final long DEFAULT_INITIAL_POLL_TIMEOUT_MS = 100;
+
+ /** Default maximum poll timeout in milliseconds after exponential backoff. */
+ private static final long DEFAULT_MAX_POLL_TIMEOUT_MS = 30_000; // 30 seconds
+
+ /** Default maximum total wait time in milliseconds before failing (1 hour). */
+ private static final long DEFAULT_MAX_TOTAL_WAIT_TIME_MS = 60 * 60 * 1000; // 1 hour
+
+ /** Interval for logging progress during long waits (5 minutes). */
+ private static final long PROGRESS_LOG_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
+
+ private final LogScanner scanner;
+ private final UpsertWriter writer;
+ private final UndoComputer undoComputer;
+
+ // Configurable timeout parameters
+ private final long initialPollTimeoutMs;
+ private final long maxPollTimeoutMs;
+ private final long maxTotalWaitTimeMs;
+
+ public UndoRecoveryExecutor(
+ LogScanner scanner, UpsertWriter writer, UndoComputer undoComputer) {
+ this(
+ scanner,
+ writer,
+ undoComputer,
+ DEFAULT_INITIAL_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_TOTAL_WAIT_TIME_MS);
+ }
+
+ /**
+ * Creates an executor with custom timeout configuration (for testing).
+ *
+ * @param scanner the log scanner
+ * @param writer the upsert writer
+ * @param undoComputer the undo computer
+ * @param initialPollTimeoutMs initial poll timeout in milliseconds
+ * @param maxPollTimeoutMs maximum poll timeout after exponential backoff
+ * @param maxTotalWaitTimeMs maximum total wait time before failing
+ */
+ public UndoRecoveryExecutor(
+ LogScanner scanner,
+ UpsertWriter writer,
+ UndoComputer undoComputer,
+ long initialPollTimeoutMs,
+ long maxPollTimeoutMs,
+ long maxTotalWaitTimeMs) {
+ this.scanner = scanner;
+ this.writer = writer;
+ this.undoComputer = undoComputer;
+ this.initialPollTimeoutMs = initialPollTimeoutMs;
+ this.maxPollTimeoutMs = maxPollTimeoutMs;
+ this.maxTotalWaitTimeMs = maxTotalWaitTimeMs;
+ }
+
+ /**
+ * Executes undo recovery for the given bucket contexts.
+ *
+ * @param contexts the bucket recovery contexts (must have target offsets set)
+ * @throws Exception if recovery fails
+ */
+ public void execute(List<BucketRecoveryContext> contexts) throws Exception {
+ // Filter contexts that need recovery
+ List<BucketRecoveryContext> toRecover = filterContextsNeedingRecovery(contexts);
+ if (toRecover.isEmpty()) {
+ LOG.debug("No buckets need recovery after filtering");
+ return;
+ }
+
+ LOG.debug("Executing undo recovery for {} bucket(s)", toRecover.size());
+
+ // Subscribe and read changelog with streaming execution
+ subscribeAll(toRecover);
+ List<CompletableFuture<?>> allFutures = readChangelogAndExecute(toRecover);
+
+ // Wait for all async writes to complete
+ if (!allFutures.isEmpty()) {
+ CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[0])).get();
+ }
+ writer.flush();
+
+ // Log summary at INFO level for visibility
+ int totalUndoOps = 0;
+ for (BucketRecoveryContext ctx : toRecover) {
+ totalUndoOps += ctx.getProcessedKeys().size();
+ }
+
+ LOG.debug(
Review Comment:
I think this is an important information, change it to `LOG.info`?
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryExecutor.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Executes undo recovery for multiple buckets using streaming execution.
+ *
+ * <p>This executor manages:
+ *
+ * <ul>
+ * <li>Single LogScanner with multiple bucket subscriptions
+ * <li>Changelog reading with immediate undo operation execution
+ * <li>Async write operations via UpsertWriter
+ * </ul>
+ *
+ * <p>The execution flow (streaming):
+ *
+ * <ol>
+ * <li>Subscribe all buckets to the LogScanner
+ * <li>Poll changelog records and execute undo operations immediately
+ * <li>Collect CompletableFutures for completion tracking
+ * <li>Wait for all futures and flush to ensure all writes complete
+ * </ol>
+ */
+public class UndoRecoveryExecutor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryExecutor.class);
+
+ /** Default initial poll timeout in milliseconds. */
+ private static final long DEFAULT_INITIAL_POLL_TIMEOUT_MS = 100;
+
+ /** Default maximum poll timeout in milliseconds after exponential backoff. */
+ private static final long DEFAULT_MAX_POLL_TIMEOUT_MS = 30_000; // 30 seconds
+
+ /** Default maximum total wait time in milliseconds before failing (1 hour). */
+ private static final long DEFAULT_MAX_TOTAL_WAIT_TIME_MS = 60 * 60 * 1000; // 1 hour
+
+ /** Interval for logging progress during long waits (5 minutes). */
+ private static final long PROGRESS_LOG_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
+
+ private final LogScanner scanner;
+ private final UpsertWriter writer;
+ private final UndoComputer undoComputer;
+
+ // Configurable timeout parameters
+ private final long initialPollTimeoutMs;
+ private final long maxPollTimeoutMs;
+ private final long maxTotalWaitTimeMs;
+
+ public UndoRecoveryExecutor(
+ LogScanner scanner, UpsertWriter writer, UndoComputer undoComputer) {
+ this(
+ scanner,
+ writer,
+ undoComputer,
+ DEFAULT_INITIAL_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_TOTAL_WAIT_TIME_MS);
+ }
+
+ /**
+ * Creates an executor with custom timeout configuration (for testing).
+ *
+ * @param scanner the log scanner
+ * @param writer the upsert writer
+ * @param undoComputer the undo computer
+ * @param initialPollTimeoutMs initial poll timeout in milliseconds
+ * @param maxPollTimeoutMs maximum poll timeout after exponential backoff
+ * @param maxTotalWaitTimeMs maximum total wait time before failing
+ */
+ public UndoRecoveryExecutor(
+ LogScanner scanner,
+ UpsertWriter writer,
+ UndoComputer undoComputer,
+ long initialPollTimeoutMs,
+ long maxPollTimeoutMs,
+ long maxTotalWaitTimeMs) {
+ this.scanner = scanner;
+ this.writer = writer;
+ this.undoComputer = undoComputer;
+ this.initialPollTimeoutMs = initialPollTimeoutMs;
+ this.maxPollTimeoutMs = maxPollTimeoutMs;
+ this.maxTotalWaitTimeMs = maxTotalWaitTimeMs;
+ }
+
+ /**
+ * Executes undo recovery for the given bucket contexts.
+ *
+ * @param contexts the bucket recovery contexts (must have target offsets set)
+ * @throws Exception if recovery fails
+ */
+ public void execute(List<BucketRecoveryContext> contexts) throws Exception {
+ // Filter contexts that need recovery
+ List<BucketRecoveryContext> toRecover = filterContextsNeedingRecovery(contexts);
+ if (toRecover.isEmpty()) {
+ LOG.debug("No buckets need recovery after filtering");
+ return;
+ }
+
+ LOG.debug("Executing undo recovery for {} bucket(s)", toRecover.size());
+
+ // Subscribe and read changelog with streaming execution
+ subscribeAll(toRecover);
+ List<CompletableFuture<?>> allFutures = readChangelogAndExecute(toRecover);
+
+ // Wait for all async writes to complete
+ if (!allFutures.isEmpty()) {
+ CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[0])).get();
+ }
+ writer.flush();
+
+ // Log summary at INFO level for visibility
+ int totalUndoOps = 0;
+ for (BucketRecoveryContext ctx : toRecover) {
+ totalUndoOps += ctx.getProcessedKeys().size();
+ }
+
+ LOG.debug(
+ "Undo recovery execution completed: {} bucket(s), {} total undo operation(s)",
+ toRecover.size(),
+ totalUndoOps);
+ }
+
+ private List<BucketRecoveryContext> filterContextsNeedingRecovery(
+ List<BucketRecoveryContext> contexts) {
+ List<BucketRecoveryContext> result = new ArrayList<>();
+ for (BucketRecoveryContext ctx : contexts) {
+ if (ctx.needsRecovery()) {
+ LOG.debug(
+ "Bucket {} needs recovery: checkpoint={}, logEndOffset={}",
+ ctx.getBucket(),
+ ctx.getCheckpointOffset(),
+ ctx.getLogEndOffset());
+ result.add(ctx);
+ } else {
+ LOG.debug("Bucket {} already up-to-date, no recovery needed", ctx.getBucket());
+ }
+ }
+ return result;
+ }
+
+ private void subscribeAll(List<BucketRecoveryContext> contexts) {
+ for (BucketRecoveryContext ctx : contexts) {
+ TableBucket bucket = ctx.getBucket();
+ if (bucket.getPartitionId() != null) {
+ scanner.subscribe(
+ bucket.getPartitionId(), bucket.getBucket(), ctx.getCheckpointOffset());
+ } else {
+ scanner.subscribe(bucket.getBucket(), ctx.getCheckpointOffset());
+ }
+ }
+ }
+
+ private List<CompletableFuture<?>> readChangelogAndExecute(List<BucketRecoveryContext> contexts)
+ throws Exception {
+ List<CompletableFuture<?>> allFutures = new ArrayList<>();
+
+ // Early exit if all contexts are already complete (no records to read)
+ if (allComplete(contexts)) {
+ LOG.debug("All buckets already complete, no changelog reading needed");
+ return allFutures;
+ }
+
+ long currentPollTimeoutMs = initialPollTimeoutMs;
+ long totalWaitTimeMs = 0;
+ long lastProgressLogTime = System.currentTimeMillis();
+
+ while (!allComplete(contexts)) {
+ ScanRecords records = scanner.poll(Duration.ofMillis(currentPollTimeoutMs));
+
+ if (records.isEmpty()) {
+ totalWaitTimeMs += currentPollTimeoutMs;
+
+ // Check if we've exceeded the maximum total wait time
+ if (totalWaitTimeMs >= maxTotalWaitTimeMs) {
+ // Undo recovery failure is fatal - incomplete undo leads to data inconsistency
+ String incompleteBucketsDetail = getIncompleteBucketsDetail(contexts);
+ throw new RuntimeException(
+ String.format(
+ "Undo recovery failed: unable to read all changelog records after "
+ + "%.1f minutes of waiting. Incomplete buckets: %s. "
+ + "This is a fatal error as incomplete undo recovery "
Review Comment:
Is this really a fatal error and data inconsistency? I think if retry success, it will still recovery the job and data to the checkpointed point?
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryExecutor.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Executes undo recovery for multiple buckets using streaming execution.
+ *
+ * <p>This executor manages:
+ *
+ * <ul>
+ * <li>Single LogScanner with multiple bucket subscriptions
+ * <li>Changelog reading with immediate undo operation execution
+ * <li>Async write operations via UpsertWriter
+ * </ul>
+ *
+ * <p>The execution flow (streaming):
+ *
+ * <ol>
+ * <li>Subscribe all buckets to the LogScanner
+ * <li>Poll changelog records and execute undo operations immediately
+ * <li>Collect CompletableFutures for completion tracking
+ * <li>Wait for all futures and flush to ensure all writes complete
+ * </ol>
+ */
+public class UndoRecoveryExecutor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryExecutor.class);
+
+ /** Default initial poll timeout in milliseconds. */
+ private static final long DEFAULT_INITIAL_POLL_TIMEOUT_MS = 100;
+
+ /** Default maximum poll timeout in milliseconds after exponential backoff. */
+ private static final long DEFAULT_MAX_POLL_TIMEOUT_MS = 30_000; // 30 seconds
+
+ /** Default maximum total wait time in milliseconds before failing (1 hour). */
+ private static final long DEFAULT_MAX_TOTAL_WAIT_TIME_MS = 60 * 60 * 1000; // 1 hour
+
+ /** Interval for logging progress during long waits (5 minutes). */
+ private static final long PROGRESS_LOG_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
+
+ private final LogScanner scanner;
+ private final UpsertWriter writer;
+ private final UndoComputer undoComputer;
+
+ // Configurable timeout parameters
+ private final long initialPollTimeoutMs;
+ private final long maxPollTimeoutMs;
+ private final long maxTotalWaitTimeMs;
+
+ public UndoRecoveryExecutor(
+ LogScanner scanner, UpsertWriter writer, UndoComputer undoComputer) {
+ this(
+ scanner,
+ writer,
+ undoComputer,
+ DEFAULT_INITIAL_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_TOTAL_WAIT_TIME_MS);
+ }
+
+ /**
+ * Creates an executor with custom timeout configuration (for testing).
+ *
+ * @param scanner the log scanner
+ * @param writer the upsert writer
+ * @param undoComputer the undo computer
+ * @param initialPollTimeoutMs initial poll timeout in milliseconds
+ * @param maxPollTimeoutMs maximum poll timeout after exponential backoff
+ * @param maxTotalWaitTimeMs maximum total wait time before failing
+ */
+ public UndoRecoveryExecutor(
+ LogScanner scanner,
+ UpsertWriter writer,
+ UndoComputer undoComputer,
+ long initialPollTimeoutMs,
+ long maxPollTimeoutMs,
+ long maxTotalWaitTimeMs) {
+ this.scanner = scanner;
+ this.writer = writer;
+ this.undoComputer = undoComputer;
+ this.initialPollTimeoutMs = initialPollTimeoutMs;
+ this.maxPollTimeoutMs = maxPollTimeoutMs;
+ this.maxTotalWaitTimeMs = maxTotalWaitTimeMs;
+ }
+
+ /**
+ * Executes undo recovery for the given bucket contexts.
+ *
+ * @param contexts the bucket recovery contexts (must have target offsets set)
+ * @throws Exception if recovery fails
+ */
+ public void execute(List<BucketRecoveryContext> contexts) throws Exception {
+ // Filter contexts that need recovery
+ List<BucketRecoveryContext> toRecover = filterContextsNeedingRecovery(contexts);
+ if (toRecover.isEmpty()) {
+ LOG.debug("No buckets need recovery after filtering");
+ return;
+ }
+
+ LOG.debug("Executing undo recovery for {} bucket(s)", toRecover.size());
+
+ // Subscribe and read changelog with streaming execution
+ subscribeAll(toRecover);
+ List<CompletableFuture<?>> allFutures = readChangelogAndExecute(toRecover);
+
+ // Wait for all async writes to complete
+ if (!allFutures.isEmpty()) {
+ CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[0])).get();
+ }
+ writer.flush();
+
+ // Log summary at INFO level for visibility
+ int totalUndoOps = 0;
+ for (BucketRecoveryContext ctx : toRecover) {
+ totalUndoOps += ctx.getProcessedKeys().size();
+ }
+
+ LOG.debug(
+ "Undo recovery execution completed: {} bucket(s), {} total undo operation(s)",
+ toRecover.size(),
+ totalUndoOps);
+ }
+
+ private List<BucketRecoveryContext> filterContextsNeedingRecovery(
+ List<BucketRecoveryContext> contexts) {
+ List<BucketRecoveryContext> result = new ArrayList<>();
+ for (BucketRecoveryContext ctx : contexts) {
+ if (ctx.needsRecovery()) {
+ LOG.debug(
+ "Bucket {} needs recovery: checkpoint={}, logEndOffset={}",
+ ctx.getBucket(),
+ ctx.getCheckpointOffset(),
+ ctx.getLogEndOffset());
+ result.add(ctx);
+ } else {
+ LOG.debug("Bucket {} already up-to-date, no recovery needed", ctx.getBucket());
+ }
+ }
+ return result;
+ }
+
+ private void subscribeAll(List<BucketRecoveryContext> contexts) {
+ for (BucketRecoveryContext ctx : contexts) {
+ TableBucket bucket = ctx.getBucket();
+ if (bucket.getPartitionId() != null) {
+ scanner.subscribe(
+ bucket.getPartitionId(), bucket.getBucket(), ctx.getCheckpointOffset());
+ } else {
+ scanner.subscribe(bucket.getBucket(), ctx.getCheckpointOffset());
+ }
+ }
+ }
+
+ private List<CompletableFuture<?>> readChangelogAndExecute(List<BucketRecoveryContext> contexts)
+ throws Exception {
+ List<CompletableFuture<?>> allFutures = new ArrayList<>();
+
+ // Early exit if all contexts are already complete (no records to read)
+ if (allComplete(contexts)) {
+ LOG.debug("All buckets already complete, no changelog reading needed");
+ return allFutures;
+ }
+
+ long currentPollTimeoutMs = initialPollTimeoutMs;
+ long totalWaitTimeMs = 0;
+ long lastProgressLogTime = System.currentTimeMillis();
+
+ while (!allComplete(contexts)) {
+ ScanRecords records = scanner.poll(Duration.ofMillis(currentPollTimeoutMs));
+
+ if (records.isEmpty()) {
+ totalWaitTimeMs += currentPollTimeoutMs;
+
+ // Check if we've exceeded the maximum total wait time
+ if (totalWaitTimeMs >= maxTotalWaitTimeMs) {
+ // Undo recovery failure is fatal - incomplete undo leads to data inconsistency
+ String incompleteBucketsDetail = getIncompleteBucketsDetail(contexts);
+ throw new RuntimeException(
+ String.format(
+ "Undo recovery failed: unable to read all changelog records after "
+ + "%.1f minutes of waiting. Incomplete buckets: %s. "
+ + "This is a fatal error as incomplete undo recovery "
+ + "would lead to data inconsistency.",
+ totalWaitTimeMs / 60000.0, incompleteBucketsDetail));
+ }
+
+ // Log progress periodically during long waits
+ long now = System.currentTimeMillis();
+ if (now - lastProgressLogTime >= PROGRESS_LOG_INTERVAL_MS) {
+ LOG.info(
+ "Undo recovery waiting for changelog records: {} bucket(s) incomplete, "
+ + "waited {} minutes so far (max: {} minutes)",
+ countIncomplete(contexts),
+ String.format("%.1f", totalWaitTimeMs / 60000.0),
+ String.format("%.1f", maxTotalWaitTimeMs / 60000.0));
+ lastProgressLogTime = now;
+ }
+
+ // Exponential backoff: double the timeout up to maxPollTimeoutMs
+ currentPollTimeoutMs = Math.min(currentPollTimeoutMs * 2, maxPollTimeoutMs);
+ LOG.debug(
+ "Empty poll, backing off to {}ms timeout (total wait: {}ms)",
+ currentPollTimeoutMs,
+ totalWaitTimeMs);
+ continue;
+ }
+
+ // Reset timeout on successful poll (but keep tracking total wait time for logging)
+ currentPollTimeoutMs = initialPollTimeoutMs;
+
+ // Process records for each bucket, resetting writer state before each bucket
+ for (BucketRecoveryContext ctx : contexts) {
+ if (ctx.isComplete()) {
+ continue;
+ }
+ List<ScanRecord> bucketRecords = records.records(ctx.getBucket());
+ if (!bucketRecords.isEmpty()) {
+ undoComputer.resetWriterState();
+ processRecords(ctx, bucketRecords, allFutures);
+ }
Review Comment:
We should add scanner unsubscribe logic when the context is completed, otherwise, we will continuesly polling data from the completed buckets, and waste a lot of time and resource.
```
if (ctx.isComplete()) {
// TODO scanner unsubscribe
}
```
We may also need to introduce `unsubscribe(int bucket)` method to the `Scanner` interface, but it is not complex.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/BucketRecoveryContext.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Encapsulates the recovery state for a single bucket.
+ *
+ * <p>This class tracks:
+ *
+ * <ul>
+ * <li>The bucket being recovered
+ * <li>The checkpoint offset (start point for reading changelog)
+ * <li>The log end offset (end point for reading changelog)
+ * <li>Processed primary keys for deduplication (streaming execution)
+ * <li>Progress tracking during changelog scanning
+ * </ul>
+ */
+public class BucketRecoveryContext {
+
+ private final TableBucket bucket;
+ private final long checkpointOffset;
+ private long logEndOffset;
+
+ private final Set<ByteArrayWrapper> processedKeys;
+ private long lastProcessedOffset;
+ private int totalRecordsProcessed;
+
+ public BucketRecoveryContext(TableBucket bucket, long checkpointOffset) {
+ this.bucket = bucket;
+ this.checkpointOffset = checkpointOffset;
+ this.logEndOffset = -1;
+ this.processedKeys = new HashSet<>();
+ this.lastProcessedOffset = checkpointOffset;
+ this.totalRecordsProcessed = 0;
+ }
+
+ public TableBucket getBucket() {
+ return bucket;
+ }
+
+ public long getCheckpointOffset() {
+ return checkpointOffset;
+ }
+
+ public long getLogEndOffset() {
+ return logEndOffset;
+ }
+
+ public void setLogEndOffset(long logEndOffset) {
+ this.logEndOffset = logEndOffset;
+ }
+
+ public Set<ByteArrayWrapper> getProcessedKeys() {
+ return processedKeys;
+ }
+
+ /**
+ * Checks if this bucket needs recovery.
+ *
+ * @return true if checkpoint offset is less than log end offset
+ */
+ public boolean needsRecovery() {
+ return checkpointOffset < logEndOffset;
+ }
+
+ /**
+ * Checks if changelog scanning is complete for this bucket.
+ *
+ * <p>Complete means either:
+ *
+ * <ul>
+ * <li>No recovery is needed (checkpointOffset >= logEndOffset), or
+ * <li>We have processed all expected records (totalRecordsProcessed >= logEndOffset -
+ * checkpointOffset)
+ * </ul>
+ *
+ * <p>This implementation assumes that changelog offsets are contiguous (no gaps). The number of
+ * records to process equals logEndOffset - checkpointOffset.
+ *
+ * <p><b>Edge case:</b> If checkpointOffset == logEndOffset - 1 (only one record to process) and
+ * that record is skipped (e.g., UPDATE_AFTER), totalRecordsProcessed will be incremented but
+ * processedKeys may be empty. This is correct behavior - the record was processed (read and
+ * evaluated), it just didn't require an undo operation.
+ *
+ * @return true if changelog scanning is complete
+ */
+ public boolean isComplete() {
+ // If no recovery is needed, we're already complete
+ if (!needsRecovery()) {
+ return true;
+ }
+ // Number of records to process = logEndOffset - checkpointOffset
+ // (offsets are contiguous, no gaps)
+ long expectedRecords = logEndOffset - checkpointOffset;
+ return totalRecordsProcessed >= expectedRecords;
Review Comment:
Ditto. We shouldn’t rely on the number of records to determine completeness. Instead, we should simply track the processed log offset using `lastProcessedOffset`.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryExecutor.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Executes undo recovery for multiple buckets using streaming execution.
+ *
+ * <p>This executor manages:
+ *
+ * <ul>
+ * <li>Single LogScanner with multiple bucket subscriptions
+ * <li>Changelog reading with immediate undo operation execution
+ * <li>Async write operations via UpsertWriter
+ * </ul>
+ *
+ * <p>The execution flow (streaming):
+ *
+ * <ol>
+ * <li>Subscribe all buckets to the LogScanner
+ * <li>Poll changelog records and execute undo operations immediately
+ * <li>Collect CompletableFutures for completion tracking
+ * <li>Wait for all futures and flush to ensure all writes complete
+ * </ol>
+ */
+public class UndoRecoveryExecutor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryExecutor.class);
+
+ /** Default initial poll timeout in milliseconds. */
+ private static final long DEFAULT_INITIAL_POLL_TIMEOUT_MS = 100;
+
+ /** Default maximum poll timeout in milliseconds after exponential backoff. */
+ private static final long DEFAULT_MAX_POLL_TIMEOUT_MS = 30_000; // 30 seconds
+
+ /** Default maximum total wait time in milliseconds before failing (1 hour). */
+ private static final long DEFAULT_MAX_TOTAL_WAIT_TIME_MS = 60 * 60 * 1000; // 1 hour
+
+ /** Interval for logging progress during long waits (5 minutes). */
+ private static final long PROGRESS_LOG_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
+
+ private final LogScanner scanner;
+ private final UpsertWriter writer;
+ private final UndoComputer undoComputer;
+
+ // Configurable timeout parameters
+ private final long initialPollTimeoutMs;
+ private final long maxPollTimeoutMs;
+ private final long maxTotalWaitTimeMs;
+
+ public UndoRecoveryExecutor(
+ LogScanner scanner, UpsertWriter writer, UndoComputer undoComputer) {
+ this(
+ scanner,
+ writer,
+ undoComputer,
+ DEFAULT_INITIAL_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_TOTAL_WAIT_TIME_MS);
+ }
+
+ /**
+ * Creates an executor with custom timeout configuration (for testing).
+ *
+ * @param scanner the log scanner
+ * @param writer the upsert writer
+ * @param undoComputer the undo computer
+ * @param initialPollTimeoutMs initial poll timeout in milliseconds
+ * @param maxPollTimeoutMs maximum poll timeout after exponential backoff
+ * @param maxTotalWaitTimeMs maximum total wait time before failing
+ */
+ public UndoRecoveryExecutor(
+ LogScanner scanner,
+ UpsertWriter writer,
+ UndoComputer undoComputer,
+ long initialPollTimeoutMs,
+ long maxPollTimeoutMs,
+ long maxTotalWaitTimeMs) {
+ this.scanner = scanner;
+ this.writer = writer;
+ this.undoComputer = undoComputer;
+ this.initialPollTimeoutMs = initialPollTimeoutMs;
+ this.maxPollTimeoutMs = maxPollTimeoutMs;
+ this.maxTotalWaitTimeMs = maxTotalWaitTimeMs;
+ }
+
+ /**
+ * Executes undo recovery for the given bucket contexts.
+ *
+ * @param contexts the bucket recovery contexts (must have target offsets set)
+ * @throws Exception if recovery fails
+ */
+ public void execute(List<BucketRecoveryContext> contexts) throws Exception {
+ // Filter contexts that need recovery
+ List<BucketRecoveryContext> toRecover = filterContextsNeedingRecovery(contexts);
+ if (toRecover.isEmpty()) {
+ LOG.debug("No buckets need recovery after filtering");
+ return;
+ }
+
+ LOG.debug("Executing undo recovery for {} bucket(s)", toRecover.size());
+
+ // Subscribe and read changelog with streaming execution
+ subscribeAll(toRecover);
+ List<CompletableFuture<?>> allFutures = readChangelogAndExecute(toRecover);
+
+ // Wait for all async writes to complete
+ if (!allFutures.isEmpty()) {
+ CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[0])).get();
+ }
+ writer.flush();
+
+ // Log summary at INFO level for visibility
+ int totalUndoOps = 0;
+ for (BucketRecoveryContext ctx : toRecover) {
+ totalUndoOps += ctx.getProcessedKeys().size();
+ }
+
+ LOG.debug(
+ "Undo recovery execution completed: {} bucket(s), {} total undo operation(s)",
+ toRecover.size(),
+ totalUndoOps);
+ }
+
+ private List<BucketRecoveryContext> filterContextsNeedingRecovery(
+ List<BucketRecoveryContext> contexts) {
+ List<BucketRecoveryContext> result = new ArrayList<>();
+ for (BucketRecoveryContext ctx : contexts) {
+ if (ctx.needsRecovery()) {
+ LOG.debug(
+ "Bucket {} needs recovery: checkpoint={}, logEndOffset={}",
+ ctx.getBucket(),
+ ctx.getCheckpointOffset(),
+ ctx.getLogEndOffset());
+ result.add(ctx);
+ } else {
+ LOG.debug("Bucket {} already up-to-date, no recovery needed", ctx.getBucket());
+ }
+ }
+ return result;
+ }
+
+ private void subscribeAll(List<BucketRecoveryContext> contexts) {
+ for (BucketRecoveryContext ctx : contexts) {
+ TableBucket bucket = ctx.getBucket();
+ if (bucket.getPartitionId() != null) {
+ scanner.subscribe(
+ bucket.getPartitionId(), bucket.getBucket(), ctx.getCheckpointOffset());
+ } else {
+ scanner.subscribe(bucket.getBucket(), ctx.getCheckpointOffset());
+ }
+ }
+ }
+
+ private List<CompletableFuture<?>> readChangelogAndExecute(List<BucketRecoveryContext> contexts)
+ throws Exception {
+ List<CompletableFuture<?>> allFutures = new ArrayList<>();
+
+ // Early exit if all contexts are already complete (no records to read)
+ if (allComplete(contexts)) {
+ LOG.debug("All buckets already complete, no changelog reading needed");
+ return allFutures;
+ }
+
+ long currentPollTimeoutMs = initialPollTimeoutMs;
+ long totalWaitTimeMs = 0;
+ long lastProgressLogTime = System.currentTimeMillis();
+
+ while (!allComplete(contexts)) {
+ ScanRecords records = scanner.poll(Duration.ofMillis(currentPollTimeoutMs));
+
+ if (records.isEmpty()) {
+ totalWaitTimeMs += currentPollTimeoutMs;
+
+ // Check if we've exceeded the maximum total wait time
+ if (totalWaitTimeMs >= maxTotalWaitTimeMs) {
+ // Undo recovery failure is fatal - incomplete undo leads to data inconsistency
+ String incompleteBucketsDetail = getIncompleteBucketsDetail(contexts);
+ throw new RuntimeException(
+ String.format(
+ "Undo recovery failed: unable to read all changelog records after "
+ + "%.1f minutes of waiting. Incomplete buckets: %s. "
+ + "This is a fatal error as incomplete undo recovery "
+ + "would lead to data inconsistency.",
+ totalWaitTimeMs / 60000.0, incompleteBucketsDetail));
+ }
+
+ // Log progress periodically during long waits
+ long now = System.currentTimeMillis();
+ if (now - lastProgressLogTime >= PROGRESS_LOG_INTERVAL_MS) {
+ LOG.info(
+ "Undo recovery waiting for changelog records: {} bucket(s) incomplete, "
+ + "waited {} minutes so far (max: {} minutes)",
+ countIncomplete(contexts),
+ String.format("%.1f", totalWaitTimeMs / 60000.0),
+ String.format("%.1f", maxTotalWaitTimeMs / 60000.0));
+ lastProgressLogTime = now;
+ }
+
+ // Exponential backoff: double the timeout up to maxPollTimeoutMs
+ currentPollTimeoutMs = Math.min(currentPollTimeoutMs * 2, maxPollTimeoutMs);
Review Comment:
I don’t understand why we need to double the poll timeout. It’s safe—and preferable—to use a consistent timeout value (e.g., 10 seconds, as used in `FlinkSourceSplitReader`) throughout. Introducing a doubled or variable timeout adds unnecessary complexity and can make behavior less predictable without clear benefit.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryCoordinator.java:
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.Table;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.writer.Upsert;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.rpc.protocol.MergeMode;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinates undo recovery operations during Flink sink writer initialization.
+ *
+ * <p>This coordinator ensures exactly-once semantics by reversing any writes that occurred after
+ * the last successful checkpoint but before a failure. The recovery process involves:
+ *
+ * <ol>
+ * <li>Reading changelog records from checkpoint offset to current latest offset
+ * <li>Computing inverse operations for each affected primary key
+ * <li>Applying undo operations using OVERWRITE mode to restore bucket state
+ * </ol>
+ *
+ * <p><b>Undo Logic:</b> For each primary key, only the first change after checkpoint determines the
+ * undo action:
+ *
+ * <ul>
+ * <li>{@code INSERT} → Delete the row (it didn't exist at checkpoint)
+ * <li>{@code UPDATE_BEFORE} → Restore the old value (this was the state at checkpoint)
+ * <li>{@code UPDATE_AFTER} → Ignored (UPDATE_BEFORE already handled the undo)
+ * <li>{@code DELETE} → Re-insert the deleted row (it existed at checkpoint)
+ * </ul>
+ *
+ * <p>This class delegates to specialized components:
+ *
+ * <ul>
+ * <li>{@link UndoComputer} - Undo operation computation using {@link KeyEncoder} for primary key
+ * encoding
+ * <li>{@link UndoRecoveryExecutor} - Changelog reading and write execution
+ * </ul>
+ *
+ * @see MergeMode#OVERWRITE
+ */
+public class UndoRecoveryCoordinator {
Review Comment:
Coordinator is confusing because there is Coordinator in both Flink and Fluss. I suggest to rename it to `UndoRecoveryManager` or something like that.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryExecutor.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Executes undo recovery for multiple buckets using streaming execution.
+ *
+ * <p>This executor manages:
+ *
+ * <ul>
+ * <li>Single LogScanner with multiple bucket subscriptions
+ * <li>Changelog reading with immediate undo operation execution
+ * <li>Async write operations via UpsertWriter
+ * </ul>
+ *
+ * <p>The execution flow (streaming):
+ *
+ * <ol>
+ * <li>Subscribe all buckets to the LogScanner
+ * <li>Poll changelog records and execute undo operations immediately
+ * <li>Collect CompletableFutures for completion tracking
+ * <li>Wait for all futures and flush to ensure all writes complete
+ * </ol>
+ */
+public class UndoRecoveryExecutor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryExecutor.class);
+
+ /** Default initial poll timeout in milliseconds. */
+ private static final long DEFAULT_INITIAL_POLL_TIMEOUT_MS = 100;
+
+ /** Default maximum poll timeout in milliseconds after exponential backoff. */
+ private static final long DEFAULT_MAX_POLL_TIMEOUT_MS = 30_000; // 30 seconds
+
+ /** Default maximum total wait time in milliseconds before failing (1 hour). */
+ private static final long DEFAULT_MAX_TOTAL_WAIT_TIME_MS = 60 * 60 * 1000; // 1 hour
+
+ /** Interval for logging progress during long waits (5 minutes). */
+ private static final long PROGRESS_LOG_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
+
+ private final LogScanner scanner;
+ private final UpsertWriter writer;
+ private final UndoComputer undoComputer;
+
+ // Configurable timeout parameters
+ private final long initialPollTimeoutMs;
+ private final long maxPollTimeoutMs;
+ private final long maxTotalWaitTimeMs;
+
+ public UndoRecoveryExecutor(
+ LogScanner scanner, UpsertWriter writer, UndoComputer undoComputer) {
+ this(
+ scanner,
+ writer,
+ undoComputer,
+ DEFAULT_INITIAL_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_TOTAL_WAIT_TIME_MS);
+ }
+
+ /**
+ * Creates an executor with custom timeout configuration (for testing).
+ *
+ * @param scanner the log scanner
+ * @param writer the upsert writer
+ * @param undoComputer the undo computer
+ * @param initialPollTimeoutMs initial poll timeout in milliseconds
+ * @param maxPollTimeoutMs maximum poll timeout after exponential backoff
+ * @param maxTotalWaitTimeMs maximum total wait time before failing
+ */
+ public UndoRecoveryExecutor(
+ LogScanner scanner,
+ UpsertWriter writer,
+ UndoComputer undoComputer,
+ long initialPollTimeoutMs,
+ long maxPollTimeoutMs,
+ long maxTotalWaitTimeMs) {
+ this.scanner = scanner;
+ this.writer = writer;
+ this.undoComputer = undoComputer;
+ this.initialPollTimeoutMs = initialPollTimeoutMs;
+ this.maxPollTimeoutMs = maxPollTimeoutMs;
+ this.maxTotalWaitTimeMs = maxTotalWaitTimeMs;
+ }
+
+ /**
+ * Executes undo recovery for the given bucket contexts.
+ *
+ * @param contexts the bucket recovery contexts (must have target offsets set)
+ * @throws Exception if recovery fails
+ */
+ public void execute(List<BucketRecoveryContext> contexts) throws Exception {
+ // Filter contexts that need recovery
+ List<BucketRecoveryContext> toRecover = filterContextsNeedingRecovery(contexts);
+ if (toRecover.isEmpty()) {
+ LOG.debug("No buckets need recovery after filtering");
+ return;
+ }
+
+ LOG.debug("Executing undo recovery for {} bucket(s)", toRecover.size());
+
+ // Subscribe and read changelog with streaming execution
+ subscribeAll(toRecover);
+ List<CompletableFuture<?>> allFutures = readChangelogAndExecute(toRecover);
+
+ // Wait for all async writes to complete
+ if (!allFutures.isEmpty()) {
+ CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[0])).get();
+ }
+ writer.flush();
Review Comment:
Move the `writer.flush()` before the future wait. This improves the performance by actively trigger flush, rather than waiting for async flush.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryCoordinator.java:
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.Table;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.writer.Upsert;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.rpc.protocol.MergeMode;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinates undo recovery operations during Flink sink writer initialization.
+ *
+ * <p>This coordinator ensures exactly-once semantics by reversing any writes that occurred after
+ * the last successful checkpoint but before a failure. The recovery process involves:
+ *
+ * <ol>
+ * <li>Reading changelog records from checkpoint offset to current latest offset
+ * <li>Computing inverse operations for each affected primary key
+ * <li>Applying undo operations using OVERWRITE mode to restore bucket state
+ * </ol>
+ *
+ * <p><b>Undo Logic:</b> For each primary key, only the first change after checkpoint determines the
+ * undo action:
+ *
+ * <ul>
+ * <li>{@code INSERT} → Delete the row (it didn't exist at checkpoint)
+ * <li>{@code UPDATE_BEFORE} → Restore the old value (this was the state at checkpoint)
+ * <li>{@code UPDATE_AFTER} → Ignored (UPDATE_BEFORE already handled the undo)
+ * <li>{@code DELETE} → Re-insert the deleted row (it existed at checkpoint)
+ * </ul>
+ *
+ * <p>This class delegates to specialized components:
+ *
+ * <ul>
+ * <li>{@link UndoComputer} - Undo operation computation using {@link KeyEncoder} for primary key
+ * encoding
+ * <li>{@link UndoRecoveryExecutor} - Changelog reading and write execution
+ * </ul>
+ *
+ * @see MergeMode#OVERWRITE
+ */
+public class UndoRecoveryCoordinator {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryCoordinator.class);
+
+ private final Table table;
+ @Nullable private final int[] targetColumnIndexes;
+
+ /**
+ * Creates a new UndoRecoveryCoordinator.
+ *
+ * @param table the Fluss table to perform recovery on
+ * @param targetColumnIndexes optional target columns for partial update (null for full row)
+ */
+ public UndoRecoveryCoordinator(Table table, @Nullable int[] targetColumnIndexes) {
+ this.table = table;
+ this.targetColumnIndexes = targetColumnIndexes;
+
+ logPartialUpdateConfig(targetColumnIndexes);
+ }
+
+ private void logPartialUpdateConfig(@Nullable int[] targetColumnIndexes) {
+ if (targetColumnIndexes != null) {
+ LOG.debug(
Review Comment:
I think this is very imporatnt information, and should use `LOG.info`
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryCoordinator.java:
##########
@@ -0,0 +1,259 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.Table;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.writer.Upsert;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.rpc.protocol.MergeMode;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Coordinates undo recovery operations during Flink sink writer initialization.
+ *
+ * <p>This coordinator ensures exactly-once semantics by reversing any writes that occurred after
+ * the last successful checkpoint but before a failure. The recovery process involves:
+ *
+ * <ol>
+ * <li>Reading changelog records from checkpoint offset to current latest offset
+ * <li>Computing inverse operations for each affected primary key
+ * <li>Applying undo operations using OVERWRITE mode to restore bucket state
+ * </ol>
+ *
+ * <p><b>Undo Logic:</b> For each primary key, only the first change after checkpoint determines the
+ * undo action:
+ *
+ * <ul>
+ * <li>{@code INSERT} → Delete the row (it didn't exist at checkpoint)
+ * <li>{@code UPDATE_BEFORE} → Restore the old value (this was the state at checkpoint)
+ * <li>{@code UPDATE_AFTER} → Ignored (UPDATE_BEFORE already handled the undo)
+ * <li>{@code DELETE} → Re-insert the deleted row (it existed at checkpoint)
+ * </ul>
+ *
+ * <p>This class delegates to specialized components:
+ *
+ * <ul>
+ * <li>{@link UndoComputer} - Undo operation computation using {@link KeyEncoder} for primary key
+ * encoding
+ * <li>{@link UndoRecoveryExecutor} - Changelog reading and write execution
+ * </ul>
+ *
+ * @see MergeMode#OVERWRITE
+ */
+public class UndoRecoveryCoordinator {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryCoordinator.class);
+
+ private final Table table;
+ @Nullable private final int[] targetColumnIndexes;
+
+ /**
+ * Creates a new UndoRecoveryCoordinator.
+ *
+ * @param table the Fluss table to perform recovery on
+ * @param targetColumnIndexes optional target columns for partial update (null for full row)
+ */
+ public UndoRecoveryCoordinator(Table table, @Nullable int[] targetColumnIndexes) {
+ this.table = table;
+ this.targetColumnIndexes = targetColumnIndexes;
+
+ logPartialUpdateConfig(targetColumnIndexes);
+ }
+
+ private void logPartialUpdateConfig(@Nullable int[] targetColumnIndexes) {
+ if (targetColumnIndexes != null) {
+ LOG.debug(
+ "Undo recovery configured with partial update columns: {}",
+ Arrays.toString(targetColumnIndexes));
+ }
+ }
+
+ // ==================== Public API ====================
+
+ /**
+ * Performs undo recovery for buckets with known undo offsets.
+ *
+ * <p>The caller is responsible for providing both the checkpoint offset (start) and the log end
+ * offset (end) for each bucket. This avoids redundant listOffset calls since the caller
+ * typically already has this information from determining which buckets need recovery.
+ *
+ * @param bucketUndoOffsets map of bucket to its undo offsets (checkpointOffset, logEndOffset)
+ * @param subtaskIndex the Flink subtask index (for logging)
+ * @param parallelism the total parallelism (for logging)
+ * @throws Exception if recovery fails
+ */
+ public void performUndoRecovery(
+ Map<TableBucket, UndoOffsets> bucketUndoOffsets, int subtaskIndex, int parallelism)
+ throws Exception {
+
+ if (bucketUndoOffsets.isEmpty()) {
+ LOG.debug("No buckets to recover on subtask {}/{}", subtaskIndex, parallelism);
+ return;
+ }
+
+ LOG.info(
+ "Starting undo recovery for {} bucket(s) on subtask {}/{}",
+ bucketUndoOffsets.size(),
+ subtaskIndex,
+ parallelism);
+
+ List<BucketRecoveryContext> contexts = buildRecoveryContexts(bucketUndoOffsets);
+
+ try (LogScanner scanner = createLogScanner()) {
+ Upsert recoveryUpsert = table.newUpsert().mergeMode(MergeMode.OVERWRITE);
+ if (targetColumnIndexes != null) {
+ recoveryUpsert = recoveryUpsert.partialUpdate(targetColumnIndexes);
+ }
+ UpsertWriter writer = recoveryUpsert.createWriter();
+
+ // For partial update mode, create a separate delete writer for INSERT undo
+ // INSERT undo needs to delete the entire row, not just partial columns
+ UpsertWriter deleteWriter = null;
+ if (targetColumnIndexes != null) {
+ deleteWriter = table.newUpsert().mergeMode(MergeMode.OVERWRITE).createWriter();
+ }
+
+ // Create UndoComputer with writer for streaming execution
+ Schema schema = table.getTableInfo().getSchema();
+ KeyEncoder keyEncoder =
+ KeyEncoder.of(
+ schema.getRowType(),
+ schema.getPrimaryKey().get().getColumnNames(),
+ null);
Review Comment:
Use `CompactedKeyEncoder.createKeyEncoder` directly, because `KeyEncoder.of` is
deprecated, and it's hack no passing the lake format parameter. Besides, we should add comment to explain why
we not use the general `KeyEncoder.of` here to avoid confusion.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryExecutor.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Executes undo recovery for multiple buckets using streaming execution.
+ *
+ * <p>This executor manages:
+ *
+ * <ul>
+ * <li>Single LogScanner with multiple bucket subscriptions
+ * <li>Changelog reading with immediate undo operation execution
+ * <li>Async write operations via UpsertWriter
+ * </ul>
+ *
+ * <p>The execution flow (streaming):
+ *
+ * <ol>
+ * <li>Subscribe all buckets to the LogScanner
+ * <li>Poll changelog records and execute undo operations immediately
+ * <li>Collect CompletableFutures for completion tracking
+ * <li>Wait for all futures and flush to ensure all writes complete
+ * </ol>
+ */
+public class UndoRecoveryExecutor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryExecutor.class);
+
+ /** Default initial poll timeout in milliseconds. */
+ private static final long DEFAULT_INITIAL_POLL_TIMEOUT_MS = 100;
+
+ /** Default maximum poll timeout in milliseconds after exponential backoff. */
+ private static final long DEFAULT_MAX_POLL_TIMEOUT_MS = 30_000; // 30 seconds
+
+ /** Default maximum total wait time in milliseconds before failing (1 hour). */
+ private static final long DEFAULT_MAX_TOTAL_WAIT_TIME_MS = 60 * 60 * 1000; // 1 hour
+
+ /** Interval for logging progress during long waits (5 minutes). */
+ private static final long PROGRESS_LOG_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
+
+ private final LogScanner scanner;
+ private final UpsertWriter writer;
+ private final UndoComputer undoComputer;
+
+ // Configurable timeout parameters
+ private final long initialPollTimeoutMs;
+ private final long maxPollTimeoutMs;
+ private final long maxTotalWaitTimeMs;
+
+ public UndoRecoveryExecutor(
+ LogScanner scanner, UpsertWriter writer, UndoComputer undoComputer) {
+ this(
+ scanner,
+ writer,
+ undoComputer,
+ DEFAULT_INITIAL_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_TOTAL_WAIT_TIME_MS);
+ }
+
+ /**
+ * Creates an executor with custom timeout configuration (for testing).
+ *
+ * @param scanner the log scanner
+ * @param writer the upsert writer
+ * @param undoComputer the undo computer
+ * @param initialPollTimeoutMs initial poll timeout in milliseconds
+ * @param maxPollTimeoutMs maximum poll timeout after exponential backoff
+ * @param maxTotalWaitTimeMs maximum total wait time before failing
+ */
+ public UndoRecoveryExecutor(
+ LogScanner scanner,
+ UpsertWriter writer,
+ UndoComputer undoComputer,
+ long initialPollTimeoutMs,
+ long maxPollTimeoutMs,
+ long maxTotalWaitTimeMs) {
+ this.scanner = scanner;
+ this.writer = writer;
+ this.undoComputer = undoComputer;
+ this.initialPollTimeoutMs = initialPollTimeoutMs;
+ this.maxPollTimeoutMs = maxPollTimeoutMs;
+ this.maxTotalWaitTimeMs = maxTotalWaitTimeMs;
+ }
+
+ /**
+ * Executes undo recovery for the given bucket contexts.
+ *
+ * @param contexts the bucket recovery contexts (must have target offsets set)
+ * @throws Exception if recovery fails
+ */
+ public void execute(List<BucketRecoveryContext> contexts) throws Exception {
+ // Filter contexts that need recovery
+ List<BucketRecoveryContext> toRecover = filterContextsNeedingRecovery(contexts);
+ if (toRecover.isEmpty()) {
+ LOG.debug("No buckets need recovery after filtering");
+ return;
+ }
+
+ LOG.debug("Executing undo recovery for {} bucket(s)", toRecover.size());
+
+ // Subscribe and read changelog with streaming execution
+ subscribeAll(toRecover);
+ List<CompletableFuture<?>> allFutures = readChangelogAndExecute(toRecover);
+
+ // Wait for all async writes to complete
+ if (!allFutures.isEmpty()) {
+ CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[0])).get();
+ }
+ writer.flush();
+
+ // Log summary at INFO level for visibility
+ int totalUndoOps = 0;
+ for (BucketRecoveryContext ctx : toRecover) {
+ totalUndoOps += ctx.getProcessedKeys().size();
+ }
+
+ LOG.debug(
+ "Undo recovery execution completed: {} bucket(s), {} total undo operation(s)",
+ toRecover.size(),
+ totalUndoOps);
+ }
+
+ private List<BucketRecoveryContext> filterContextsNeedingRecovery(
+ List<BucketRecoveryContext> contexts) {
+ List<BucketRecoveryContext> result = new ArrayList<>();
+ for (BucketRecoveryContext ctx : contexts) {
+ if (ctx.needsRecovery()) {
+ LOG.debug(
+ "Bucket {} needs recovery: checkpoint={}, logEndOffset={}",
+ ctx.getBucket(),
+ ctx.getCheckpointOffset(),
+ ctx.getLogEndOffset());
+ result.add(ctx);
+ } else {
+ LOG.debug("Bucket {} already up-to-date, no recovery needed", ctx.getBucket());
+ }
+ }
+ return result;
+ }
+
+ private void subscribeAll(List<BucketRecoveryContext> contexts) {
+ for (BucketRecoveryContext ctx : contexts) {
+ TableBucket bucket = ctx.getBucket();
+ if (bucket.getPartitionId() != null) {
+ scanner.subscribe(
+ bucket.getPartitionId(), bucket.getBucket(), ctx.getCheckpointOffset());
+ } else {
+ scanner.subscribe(bucket.getBucket(), ctx.getCheckpointOffset());
+ }
+ }
+ }
+
+ private List<CompletableFuture<?>> readChangelogAndExecute(List<BucketRecoveryContext> contexts)
+ throws Exception {
+ List<CompletableFuture<?>> allFutures = new ArrayList<>();
+
+ // Early exit if all contexts are already complete (no records to read)
+ if (allComplete(contexts)) {
+ LOG.debug("All buckets already complete, no changelog reading needed");
+ return allFutures;
+ }
+
+ long currentPollTimeoutMs = initialPollTimeoutMs;
+ long totalWaitTimeMs = 0;
+ long lastProgressLogTime = System.currentTimeMillis();
+
+ while (!allComplete(contexts)) {
+ ScanRecords records = scanner.poll(Duration.ofMillis(currentPollTimeoutMs));
+
+ if (records.isEmpty()) {
+ totalWaitTimeMs += currentPollTimeoutMs;
+
+ // Check if we've exceeded the maximum total wait time
+ if (totalWaitTimeMs >= maxTotalWaitTimeMs) {
+ // Undo recovery failure is fatal - incomplete undo leads to data inconsistency
+ String incompleteBucketsDetail = getIncompleteBucketsDetail(contexts);
Review Comment:
Just print the the number of buckets in the error message? Otherwise, it maybe a very huge exception message if there are thousands of buckets but very little Flink parallelism.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryExecutor.java:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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.writer.undo;
+
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Executes undo recovery for multiple buckets using streaming execution.
+ *
+ * <p>This executor manages:
+ *
+ * <ul>
+ * <li>Single LogScanner with multiple bucket subscriptions
+ * <li>Changelog reading with immediate undo operation execution
+ * <li>Async write operations via UpsertWriter
+ * </ul>
+ *
+ * <p>The execution flow (streaming):
+ *
+ * <ol>
+ * <li>Subscribe all buckets to the LogScanner
+ * <li>Poll changelog records and execute undo operations immediately
+ * <li>Collect CompletableFutures for completion tracking
+ * <li>Wait for all futures and flush to ensure all writes complete
+ * </ol>
+ */
+public class UndoRecoveryExecutor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UndoRecoveryExecutor.class);
+
+ /** Default initial poll timeout in milliseconds. */
+ private static final long DEFAULT_INITIAL_POLL_TIMEOUT_MS = 100;
+
+ /** Default maximum poll timeout in milliseconds after exponential backoff. */
+ private static final long DEFAULT_MAX_POLL_TIMEOUT_MS = 30_000; // 30 seconds
+
+ /** Default maximum total wait time in milliseconds before failing (1 hour). */
+ private static final long DEFAULT_MAX_TOTAL_WAIT_TIME_MS = 60 * 60 * 1000; // 1 hour
+
+ /** Interval for logging progress during long waits (5 minutes). */
+ private static final long PROGRESS_LOG_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
+
+ private final LogScanner scanner;
+ private final UpsertWriter writer;
+ private final UndoComputer undoComputer;
+
+ // Configurable timeout parameters
+ private final long initialPollTimeoutMs;
+ private final long maxPollTimeoutMs;
+ private final long maxTotalWaitTimeMs;
+
+ public UndoRecoveryExecutor(
+ LogScanner scanner, UpsertWriter writer, UndoComputer undoComputer) {
+ this(
+ scanner,
+ writer,
+ undoComputer,
+ DEFAULT_INITIAL_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_POLL_TIMEOUT_MS,
+ DEFAULT_MAX_TOTAL_WAIT_TIME_MS);
+ }
+
+ /**
+ * Creates an executor with custom timeout configuration (for testing).
+ *
+ * @param scanner the log scanner
+ * @param writer the upsert writer
+ * @param undoComputer the undo computer
+ * @param initialPollTimeoutMs initial poll timeout in milliseconds
+ * @param maxPollTimeoutMs maximum poll timeout after exponential backoff
+ * @param maxTotalWaitTimeMs maximum total wait time before failing
+ */
+ public UndoRecoveryExecutor(
+ LogScanner scanner,
+ UpsertWriter writer,
+ UndoComputer undoComputer,
+ long initialPollTimeoutMs,
+ long maxPollTimeoutMs,
+ long maxTotalWaitTimeMs) {
+ this.scanner = scanner;
+ this.writer = writer;
+ this.undoComputer = undoComputer;
+ this.initialPollTimeoutMs = initialPollTimeoutMs;
+ this.maxPollTimeoutMs = maxPollTimeoutMs;
+ this.maxTotalWaitTimeMs = maxTotalWaitTimeMs;
+ }
+
+ /**
+ * Executes undo recovery for the given bucket contexts.
+ *
+ * @param contexts the bucket recovery contexts (must have target offsets set)
+ * @throws Exception if recovery fails
+ */
+ public void execute(List<BucketRecoveryContext> contexts) throws Exception {
+ // Filter contexts that need recovery
+ List<BucketRecoveryContext> toRecover = filterContextsNeedingRecovery(contexts);
+ if (toRecover.isEmpty()) {
+ LOG.debug("No buckets need recovery after filtering");
+ return;
+ }
+
+ LOG.debug("Executing undo recovery for {} bucket(s)", toRecover.size());
+
+ // Subscribe and read changelog with streaming execution
+ subscribeAll(toRecover);
+ List<CompletableFuture<?>> allFutures = readChangelogAndExecute(toRecover);
+
+ // Wait for all async writes to complete
+ if (!allFutures.isEmpty()) {
+ CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[0])).get();
+ }
+ writer.flush();
+
+ // Log summary at INFO level for visibility
+ int totalUndoOps = 0;
+ for (BucketRecoveryContext ctx : toRecover) {
+ totalUndoOps += ctx.getProcessedKeys().size();
+ }
+
+ LOG.debug(
+ "Undo recovery execution completed: {} bucket(s), {} total undo operation(s)",
+ toRecover.size(),
+ totalUndoOps);
+ }
+
+ private List<BucketRecoveryContext> filterContextsNeedingRecovery(
+ List<BucketRecoveryContext> contexts) {
+ List<BucketRecoveryContext> result = new ArrayList<>();
+ for (BucketRecoveryContext ctx : contexts) {
+ if (ctx.needsRecovery()) {
+ LOG.debug(
+ "Bucket {} needs recovery: checkpoint={}, logEndOffset={}",
+ ctx.getBucket(),
+ ctx.getCheckpointOffset(),
+ ctx.getLogEndOffset());
+ result.add(ctx);
+ } else {
+ LOG.debug("Bucket {} already up-to-date, no recovery needed", ctx.getBucket());
+ }
+ }
+ return result;
+ }
+
+ private void subscribeAll(List<BucketRecoveryContext> contexts) {
+ for (BucketRecoveryContext ctx : contexts) {
+ TableBucket bucket = ctx.getBucket();
+ if (bucket.getPartitionId() != null) {
+ scanner.subscribe(
+ bucket.getPartitionId(), bucket.getBucket(), ctx.getCheckpointOffset());
+ } else {
+ scanner.subscribe(bucket.getBucket(), ctx.getCheckpointOffset());
+ }
+ }
+ }
+
+ private List<CompletableFuture<?>> readChangelogAndExecute(List<BucketRecoveryContext> contexts)
+ throws Exception {
+ List<CompletableFuture<?>> allFutures = new ArrayList<>();
+
+ // Early exit if all contexts are already complete (no records to read)
+ if (allComplete(contexts)) {
+ LOG.debug("All buckets already complete, no changelog reading needed");
+ return allFutures;
+ }
+
+ long currentPollTimeoutMs = initialPollTimeoutMs;
+ long totalWaitTimeMs = 0;
+ long lastProgressLogTime = System.currentTimeMillis();
+
+ while (!allComplete(contexts)) {
+ ScanRecords records = scanner.poll(Duration.ofMillis(currentPollTimeoutMs));
+
+ if (records.isEmpty()) {
+ totalWaitTimeMs += currentPollTimeoutMs;
+
+ // Check if we've exceeded the maximum total wait time
+ if (totalWaitTimeMs >= maxTotalWaitTimeMs) {
+ // Undo recovery failure is fatal - incomplete undo leads to data inconsistency
+ String incompleteBucketsDetail = getIncompleteBucketsDetail(contexts);
+ throw new RuntimeException(
+ String.format(
+ "Undo recovery failed: unable to read all changelog records after "
+ + "%.1f minutes of waiting. Incomplete buckets: %s. "
+ + "This is a fatal error as incomplete undo recovery "
+ + "would lead to data inconsistency.",
+ totalWaitTimeMs / 60000.0, incompleteBucketsDetail));
+ }
+
+ // Log progress periodically during long waits
+ long now = System.currentTimeMillis();
+ if (now - lastProgressLogTime >= PROGRESS_LOG_INTERVAL_MS) {
+ LOG.info(
+ "Undo recovery waiting for changelog records: {} bucket(s) incomplete, "
+ + "waited {} minutes so far (max: {} minutes)",
+ countIncomplete(contexts),
+ String.format("%.1f", totalWaitTimeMs / 60000.0),
+ String.format("%.1f", maxTotalWaitTimeMs / 60000.0));
+ lastProgressLogTime = now;
+ }
+
+ // Exponential backoff: double the timeout up to maxPollTimeoutMs
+ currentPollTimeoutMs = Math.min(currentPollTimeoutMs * 2, maxPollTimeoutMs);
+ LOG.debug(
+ "Empty poll, backing off to {}ms timeout (total wait: {}ms)",
+ currentPollTimeoutMs,
+ totalWaitTimeMs);
+ continue;
+ }
+
+ // Reset timeout on successful poll (but keep tracking total wait time for logging)
+ currentPollTimeoutMs = initialPollTimeoutMs;
+
+ // Process records for each bucket, resetting writer state before each bucket
+ for (BucketRecoveryContext ctx : contexts) {
+ if (ctx.isComplete()) {
+ continue;
+ }
+ List<ScanRecord> bucketRecords = records.records(ctx.getBucket());
+ if (!bucketRecords.isEmpty()) {
+ undoComputer.resetWriterState();
+ processRecords(ctx, bucketRecords, allFutures);
+ }
+ }
+ }
+
+ // Log summary
+ for (BucketRecoveryContext ctx : contexts) {
+ LOG.debug(
+ "Bucket {} read {} records, executed {} undo operations",
+ ctx.getBucket(),
+ ctx.getTotalRecordsProcessed(),
+ ctx.getProcessedKeys().size());
+ }
Review Comment:
Add `if (LOG.isDebugEnabled())` condition for the code block.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add undo recovery support for aggregation tables [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #2545:
URL: https://github.com/apache/fluss/pull/2545#discussion_r2782319757
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryCoordinatorITCase.java:
##########
@@ -374,25 +430,11 @@ void testPartialUpdateMultiBucketComprehensiveRecovery() throws Exception {
// Phase 3: Perform undo recovery
Map<TableBucket, UndoOffsets> offsetRanges =
buildUndoOffsets(checkpointOffsets, tablePath, admin);
- new UndoRecoveryCoordinator(table, targetColumns)
- .performUndoRecovery(offsetRanges, 0, 1);
+ new UndoRecoveryManager(table, targetColumns).performUndoRecovery(offsetRanges, 0, 1);
// Phase 4: Verify results (parallel lookup)
List<CompletableFuture<Void>> verifyFutures = new ArrayList<>();
- for (int id : zeroOffsetBucketKeysAfterCheckpoint) {
- final int keyId = id;
- verifyFutures.add(
- lookuper.lookup(row(keyId))
- .thenAccept(
- r ->
- assertThat(r.getSingletonRow())
- .as(
- "Zero-offset bucket key %d should be deleted",
- keyId)
- .isNull()));
- }
Review Comment:
I'll reintroduce this test in #2609.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add undo recovery support for aggregation tables [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on PR #2545:
URL: https://github.com/apache/fluss/pull/2545#issuecomment-3869017738
@wuchong Thank you for your detailed review. I’ve addressed all the comments above—please take a look when you have a moment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add undo recovery support for aggregation tables [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #2545:
URL: https://github.com/apache/fluss/pull/2545#discussion_r2782263448
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/writer/undo/UndoRecoveryCoordinatorITCase.java:
##########
@@ -374,25 +430,11 @@ void testPartialUpdateMultiBucketComprehensiveRecovery() throws Exception {
// Phase 3: Perform undo recovery
Map<TableBucket, UndoOffsets> offsetRanges =
buildUndoOffsets(checkpointOffsets, tablePath, admin);
- new UndoRecoveryCoordinator(table, targetColumns)
- .performUndoRecovery(offsetRanges, 0, 1);
+ new UndoRecoveryManager(table, targetColumns).performUndoRecovery(offsetRanges, 0, 1);
// Phase 4: Verify results (parallel lookup)
List<CompletableFuture<Void>> verifyFutures = new ArrayList<>();
- for (int id : zeroOffsetBucketKeysAfterCheckpoint) {
- final int keyId = id;
- verifyFutures.add(
- lookuper.lookup(row(keyId))
- .thenAccept(
- r ->
- assertThat(r.getSingletonRow())
- .as(
- "Zero-offset bucket key %d should be deleted",
- keyId)
- .isNull()));
- }
Review Comment:
I think we still need to check the `zeroOffsetBucketKeysAfterCheckpoint` related rows after restoring. Could you add this test back?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add undo recovery support for aggregation tables [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #2545:
URL: https://github.com/apache/fluss/pull/2545
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add undo recovery support for aggregation tables [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #2545:
URL: https://github.com/apache/fluss/pull/2545#issuecomment-3866942286
Btw, it seems the UNDO recovery for aggregate merge engine table relies on the full changelog of the table. However, `table.changelog.image=WAL` will break this and lead to in-correct results. I think we need to forbid such case, a validation check can be done during table creation and throw exception for tables with agg merge engine and WAL mode. We can do that in another issue. What do you think @platinumhamburg ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org