[lake/flink] Enforce to keep data insistent when commit to lake but fail to commit to fluss (#1067)

Author: luoyuxia

fluss-common/src/main/java/com/alibaba/fluss/lake/committer/CommittedLakeSnapshot.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.lake.committer;
+
+import com.alibaba.fluss.utils.types.Tuple2;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The lake already committed snapshot, containing the lake snapshot id and the bucket end offsets
+ * in this snapshot.
+ */
+public class CommittedLakeSnapshot {
+
+    private final long lakeSnapshotId;
+    // <partition_name, bucket> -> log offset, partition_name will be null if it's not a
+    // partition bucket
+    private final Map<Tuple2<String, Integer>, Long> logEndOffsets = new HashMap<>();
+
+    public CommittedLakeSnapshot(long lakeSnapshotId) {
+        this.lakeSnapshotId = lakeSnapshotId;
+    }
+
+    public long getLakeSnapshotId() {
+        return lakeSnapshotId;
+    }
+
+    public void addBucket(int bucketId, long offset) {
+        logEndOffsets.put(Tuple2.of(null, bucketId), offset);
+    }
+
+    public void addPartitionBucket(String partitionName, int bucketId, long offset) {
+        logEndOffsets.put(Tuple2.of(partitionName, bucketId), offset);
+    }
+
+    public Map<Tuple2<String, Integer>, Long> getLogEndOffsets() {
+        return logEndOffsets;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        CommittedLakeSnapshot that = (CommittedLakeSnapshot) o;
+        return lakeSnapshotId == that.lakeSnapshotId
+                && Objects.equals(logEndOffsets, that.logEndOffsets);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(lakeSnapshotId, logEndOffsets);
+    }
+
+    @Override
+    public String toString() {
+        return "CommittedLakeSnapshot{"
+                + "lakeSnapshotId="
+                + lakeSnapshotId
+                + ", logEndOffsets="
+                + logEndOffsets
+                + '}';
+    }
+}

fluss-common/src/main/java/com/alibaba/fluss/lake/committer/LakeCommitter.java

@@ -18,6 +18,8 @@
 
 import com.alibaba.fluss.annotation.PublicEvolving;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.util.List;
 
@@ -49,4 +51,24 @@ public interface LakeCommitter<WriteResult, CommittableT> extends AutoCloseable
      * @throws IOException if an I/O error occurs
      */
     long commit(CommittableT committable) throws IOException;
+
+    /**
+     * Aborts the given committable object.
+     *
+     * @param committable the committable object
+     * @throws IOException if an I/O error occurs
+     */
+    void abort(CommittableT committable) throws IOException;
+
+    /**
+     * Get missing lake snapshot that has been committed to lake but didn't commit to fluss.
+     *
+     * @param latestLakeSnapshotIdOfFluss the latest lake snapshot id in fluss, used to judge which
+     *     lake snapshot is missing.
+     * @return the missing lake snapshot, returns null if no any missing snapshot found
+     * @throws IOException if an I/O error occurs
+     */
+    @Nullable
+    CommittedLakeSnapshot getMissingLakeSnapshot(@Nullable Long latestLakeSnapshotIdOfFluss)
+            throws IOException;
 }

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/committer/FlussTableLakeSnapshot.java

@@ -18,6 +18,7 @@
 
 import com.alibaba.fluss.metadata.TableBucket;
 
+import java.util.HashMap;
 import java.util.Map;
 
 /** A lake snapshot for a Fluss table. */
@@ -29,6 +30,10 @@ public class FlussTableLakeSnapshot {
 
     private final Map<TableBucket, Long> logEndOffsets;
 
+    public FlussTableLakeSnapshot(long tableId, long lakeSnapshotId) {
+        this(tableId, lakeSnapshotId, new HashMap<>());
+    }
+
     public FlussTableLakeSnapshot(
             long tableId, long lakeSnapshotId, Map<TableBucket, Long> logEndOffsets) {
         this.tableId = tableId;
@@ -48,6 +53,10 @@ public Map<TableBucket, Long> logEndOffsets() {
         return logEndOffsets;
     }
 
+    public void addBucketOffset(TableBucket bucket, long offset) {
+        logEndOffsets.put(bucket, offset);
+    }
+
     @Override
     public String toString() {
         return "FlussTableLakeSnapshot{"

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java

@@ -19,6 +19,7 @@
 import com.alibaba.fluss.client.metadata.MetadataUpdater;
 import com.alibaba.fluss.config.ConfigOptions;
 import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.lake.committer.CommittedLakeSnapshot;
 import com.alibaba.fluss.metadata.TableBucket;
 import com.alibaba.fluss.metrics.registry.MetricRegistry;
 import com.alibaba.fluss.rpc.GatewayClientProxy;
@@ -29,6 +30,9 @@
 import com.alibaba.fluss.rpc.messages.PbLakeTableSnapshotInfo;
 import com.alibaba.fluss.rpc.metrics.ClientMetricGroup;
 import com.alibaba.fluss.utils.ExceptionUtils;
+import com.alibaba.fluss.utils.types.Tuple2;
+
+import javax.annotation.Nullable;
 
 import java.io.IOException;
 import java.util.Map;
@@ -71,6 +75,38 @@ public void commit(FlussTableLakeSnapshot flussTableLakeSnapshot) throws IOExcep
         }
     }
 
+    public void commit(
+            long tableId,
+            @Nullable Map<String, Long> partitionIdByName,
+            CommittedLakeSnapshot committedLakeSnapshot)
+            throws IOException {
+        // construct lake snapshot to commit to Fluss
+        FlussTableLakeSnapshot flussTableLakeSnapshot =
+                new FlussTableLakeSnapshot(tableId, committedLakeSnapshot.getLakeSnapshotId());
+        for (Map.Entry<Tuple2<String, Integer>, Long> entry :
+                committedLakeSnapshot.getLogEndOffsets().entrySet()) {
+            Tuple2<String, Integer> partitionBucket = entry.getKey();
+            TableBucket tableBucket;
+            if (partitionBucket.f0 == null) {
+                tableBucket = new TableBucket(tableId, partitionBucket.f1);
+            } else {
+                String partitionName = partitionBucket.f0;
+                // todo: remove this
+                // in paimon 1.12, we can store this offsets(including partitionId) into snapshot
+                // properties, then, we won't need to get partitionId from partition name
+                Long partitionId = partitionIdByName.get(partitionName);
+                if (partitionId != null) {
+                    tableBucket = new TableBucket(tableId, partitionId, partitionBucket.f1);
+                } else {
+                    // let's skip the bucket
+                    continue;
+                }
+            }
+            flussTableLakeSnapshot.addBucketOffset(tableBucket, entry.getValue());
+        }
+        commit(flussTableLakeSnapshot);
+    }
+
     private CommitLakeTableSnapshotRequest toCommitLakeTableSnapshotRequest(
             FlussTableLakeSnapshot flussTableLakeSnapshot) {
         CommitLakeTableSnapshotRequest commitLakeTableSnapshotRequest =

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/committer/TieringCommitOperator.java

@@ -16,15 +16,24 @@
 
 package com.alibaba.fluss.flink.tiering.committer;
 
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.ConnectionFactory;
+import com.alibaba.fluss.client.admin.Admin;
 import com.alibaba.fluss.config.Configuration;
-import com.alibaba.fluss.flink.tiering.event.FinishTieringEvent;
+import com.alibaba.fluss.exception.LakeTableSnapshotNotExistException;
+import com.alibaba.fluss.flink.tiering.event.FailedTieringEvent;
+import com.alibaba.fluss.flink.tiering.event.FinishedTieringEvent;
 import com.alibaba.fluss.flink.tiering.source.TableBucketWriteResult;
 import com.alibaba.fluss.flink.tiering.source.TieringSource;
+import com.alibaba.fluss.lake.committer.CommittedLakeSnapshot;
 import com.alibaba.fluss.lake.committer.LakeCommitter;
 import com.alibaba.fluss.lake.writer.LakeTieringFactory;
 import com.alibaba.fluss.lake.writer.LakeWriter;
+import com.alibaba.fluss.metadata.PartitionInfo;
 import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TableInfo;
 import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.utils.ExceptionUtils;
 
 import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
 import org.apache.flink.runtime.source.event.SourceEventWrapper;
@@ -67,8 +76,11 @@ public class TieringCommitOperator<WriteResult, Committable>
 
     private static final long serialVersionUID = 1L;
 
+    private final Configuration flussConfig;
     private final LakeTieringFactory<WriteResult, Committable> lakeTieringFactory;
     private final FlussTableLakeSnapshotCommitter flussTableLakeSnapshotCommitter;
+    private Connection connection;
+    private Admin admin;
 
     // gateway to send event to flink source coordinator
     private final OperatorEventGateway operatorEventGateway;
@@ -84,6 +96,7 @@ public TieringCommitOperator(
         this.lakeTieringFactory = lakeTieringFactory;
         this.flussTableLakeSnapshotCommitter = new FlussTableLakeSnapshotCommitter(flussConf);
         this.collectedTableBucketWriteResults = new HashMap<>();
+        this.flussConfig = flussConf;
         this.setup(
                 parameters.getContainingTask(),
                 parameters.getStreamConfig(),
@@ -97,6 +110,8 @@ public TieringCommitOperator(
     @Override
     public void open() {
         flussTableLakeSnapshotCommitter.open();
+        connection = ConnectionFactory.createConnection(flussConfig);
+        admin = connection.getAdmin();
     }
 
     @Override
@@ -112,16 +127,27 @@ public void processElement(StreamRecord<TableBucketWriteResult<WriteResult>> str
                 collectTableAllBucketWriteResult(tableId);
 
         if (committableWriteResults != null) {
-            Committable committable =
-                    commitWriteResults(
-                            tableId, tableBucketWriteResult.tablePath(), committableWriteResults);
-            collectedTableBucketWriteResults.remove(tableId);
-            // notify that the table id has been finished tier
-            operatorEventGateway.sendEventToCoordinator(
-                    new SourceEventWrapper(new FinishTieringEvent(tableId)));
-            // only emit when committable is not-null
-            if (committable != null) {
-                output.collect(new StreamRecord<>(new CommittableMessage<>(committable)));
+            try {
+                Committable committable =
+                        commitWriteResults(
+                                tableId,
+                                tableBucketWriteResult.tablePath(),
+                                committableWriteResults);
+                // only emit when committable is not-null
+                if (committable != null) {
+                    output.collect(new StreamRecord<>(new CommittableMessage<>(committable)));
+                }
+                // notify that the table id has been finished tier
+                operatorEventGateway.sendEventToCoordinator(
+                        new SourceEventWrapper(new FinishedTieringEvent(tableId)));
+            } catch (Exception e) {
+                // if any exception happens, send to source coordinator to mark it as failed
+                operatorEventGateway.sendEventToCoordinator(
+                        new SourceEventWrapper(
+                                new FailedTieringEvent(
+                                        tableId, ExceptionUtils.stringifyException(e))));
+            } finally {
+                collectedTableBucketWriteResults.remove(tableId);
             }
         }
     }
@@ -154,6 +180,8 @@ private Committable commitWriteResults(
                             .collect(Collectors.toList());
             // to committable
             Committable committable = lakeCommitter.toCommitable(writeResults);
+            // before commit to lake, check fluss not missing any lake snapshot commited by fluss
+            checkFlussNotMissingLakeSnapshot(tablePath, lakeCommitter, committable);
             long commitedSnapshotId = lakeCommitter.commit(committable);
             // commit to fluss
             Map<TableBucket, Long> logEndOffsets = new HashMap<>();
@@ -166,6 +194,61 @@ private Committable commitWriteResults(
         }
     }
 
+    private void checkFlussNotMissingLakeSnapshot(
+            TablePath tablePath,
+            LakeCommitter<WriteResult, Committable> lakeCommitter,
+            Committable committable)
+            throws Exception {
+        Long flussCurrentLakeSnapshot;
+        try {
+            flussCurrentLakeSnapshot = admin.getLatestLakeSnapshot(tablePath).get().getSnapshotId();
+        } catch (Exception e) {
+            Throwable throwable = e.getCause();
+            if (throwable instanceof LakeTableSnapshotNotExistException) {
+                // do-nothing
+                flussCurrentLakeSnapshot = null;
+            } else {
+                throw e;
+            }
+        }
+
+        // get Fluss missing lake snapshot in Lake
+        CommittedLakeSnapshot missingCommittedSnapshot =
+                lakeCommitter.getMissingLakeSnapshot(flussCurrentLakeSnapshot);
+
+        // fluss's known snapshot is less than lake snapshot committed by fluss
+        // fail this commit since the data is read from the log end-offset of a invalid fluss
+        // known lake snapshot, which means the data already has been committed to lake,
+        // not to commit to lake to avoid data duplicated
+        if (missingCommittedSnapshot != null) {
+            // commit this missing snapshot to fluss
+            TableInfo tableInfo = admin.getTableInfo(tablePath).get();
+            Map<String, Long> partitionIdByName = null;
+            if (tableInfo.isPartitioned()) {
+                partitionIdByName =
+                        admin.listPartitionInfos(tablePath).get().stream()
+                                .collect(
+                                        Collectors.toMap(
+                                                PartitionInfo::getPartitionName,
+                                                PartitionInfo::getPartitionId));
+            }
+            flussTableLakeSnapshotCommitter.commit(
+                    tableInfo.getTableId(), partitionIdByName, missingCommittedSnapshot);
+            // abort this committable to delete the written files
+            lakeCommitter.abort(committable);
+            throw new IllegalStateException(
+                    String.format(
+                            "The current Fluss's lake snapshot %d is less than"
+                                    + " lake actual snapshot %d committed by Fluss for table: {tablePath=%s, tableId=%d},"
+                                    + " missing snapshot: %s.",
+                            flussCurrentLakeSnapshot,
+                            missingCommittedSnapshot.getLakeSnapshotId(),
+                            tableInfo.getTablePath(),
+                            tableInfo.getTableId(),
+                            missingCommittedSnapshot));
+        }
+    }
+
     private void registerTableBucketWriteResult(
             long tableId, TableBucketWriteResult<WriteResult> tableBucketWriteResult) {
         collectedTableBucketWriteResults
@@ -214,5 +297,11 @@ private List<TableBucketWriteResult<WriteResult>> collectTableAllBucketWriteResu
     @Override
     public void close() throws Exception {
         flussTableLakeSnapshotCommitter.close();
+        if (admin != null) {
+            admin.close();
+        }
+        if (connection != null) {
+            connection.close();
+        }
     }
 }