[log] Follower fetch offset need truncate to leaderEndOffsetSnapshot if we found duplicated batch or out of sequence batch (#731)

swuferhong · web-flow · commit b52701364eac · 2025-04-12T21:14:11.000+08:00
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/log/ListOffsetsParam.java b/fluss-server/src/main/java/com/alibaba/fluss/server/log/ListOffsetsParam.java
@@ -41,6 +41,9 @@ public class ListOffsetsParam {
      */
     public static final int TIMESTAMP_OFFSET_TYPE = 2;
 
+    /** The offset type indicate the end offset snapshot when become leader. */
+    public static final int LEADER_END_OFFSET_SNAPSHOT_TYPE = 3;
+
     private final int followerServerId;
     private final Integer offsetType;
     private @Nullable final Long startTimestamp;
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogTablet.java b/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogTablet.java
@@ -20,6 +20,7 @@
 import com.alibaba.fluss.config.ConfigOptions;
 import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.exception.CorruptRecordException;
+import com.alibaba.fluss.exception.DuplicateSequenceException;
 import com.alibaba.fluss.exception.FlussRuntimeException;
 import com.alibaba.fluss.exception.InvalidTimestampException;
 import com.alibaba.fluss.exception.LogOffsetOutOfRangeException;
@@ -105,6 +106,9 @@ public final class LogTablet {
     @GuardedBy("lock")
     private volatile LogOffsetMetadata highWatermarkMetadata;
 
+    /** The leader end offset snapshot when become leader. */
+    private volatile long leaderEndOffsetSnapshot = -1L;
+
     // The minimum offset that should be retained in the local log. This is used to ensure that,
     // the offset of kv snapshot should be retained, otherwise, kv recovery will fail.
     private volatile long minRetainOffset;
@@ -259,6 +263,10 @@ public LogFormat getLogFormat() {
         return logFormat;
     }
 
+    public long getLeaderEndOffsetSnapshot() {
+        return leaderEndOffsetSnapshot;
+    }
+
     @VisibleForTesting
     public WriterStateManager writerStateManager() {
         return writerStateManager;
@@ -336,6 +344,16 @@ public void registerMetrics(BucketMetricGroup bucketMetricGroup) {
                 MetricNames.LOG_FLUSH_LATENCY_MS, localLog.getFlushLatencyHistogram());
     }
 
+    public void updateLeaderEndOffsetSnapshot() {
+        synchronized (lock) {
+            LOG.info(
+                    "Update leaderEndOffsetSnapshot to {} for tb {} while become leader",
+                    localLogEndOffset(),
+                    localLog.getTableBucket());
+            leaderEndOffsetSnapshot = localLog.getLocalLogEndOffset();
+        }
+    }
+
     /**
      * Append this message set to the active segment of the local log, assigning offsets and Bucket
      * Leader Epochs.
@@ -582,10 +600,10 @@ private LogOffsetMetadata convertToOffsetMetadataOrThrow(long offset) throws IOE
      * segment if necessary.
      *
      * <p>This method will generally be responsible for assigning offsets to the messages, however
-     * if the needAssignOffsetAndTimestamp=false flag is passed we will only check that the existing
-     * offsets are valid.
+     * if the appendAsLeader=false flag is passed we will only check that the existing offsets are
+     * valid.
      */
-    private LogAppendInfo append(MemoryLogRecords records, boolean needAssignOffsetAndTimestamp)
+    private LogAppendInfo append(MemoryLogRecords records, boolean appendAsLeader)
             throws Exception {
         LogAppendInfo appendInfo = analyzeAndValidateRecords(records);
 
@@ -599,7 +617,7 @@ private LogAppendInfo append(MemoryLogRecords records, boolean needAssignOffsetA
 
         synchronized (lock) {
             localLog.checkIfMemoryMappedBufferClosed();
-            if (needAssignOffsetAndTimestamp) {
+            if (appendAsLeader) {
                 long offset = localLog.getLocalLogEndOffset();
                 // assign offsets to the message set.
                 appendInfo.setFirstOffset(offset);
@@ -630,11 +648,24 @@ private LogAppendInfo append(MemoryLogRecords records, boolean needAssignOffsetA
                 // have duplicated batch metadata, skip the append and update append info.
                 WriterStateEntry.BatchMetadata duplicatedBatch = validateResult.left();
                 long startOffset = duplicatedBatch.firstOffset();
-                appendInfo.setFirstOffset(startOffset);
-                appendInfo.setLastOffset(duplicatedBatch.lastOffset);
-                appendInfo.setMaxTimestamp(duplicatedBatch.timestamp);
-                appendInfo.setStartOffsetOfMaxTimestamp(startOffset);
-                appendInfo.setDuplicated(true);
+                if (appendAsLeader) {
+                    appendInfo.setFirstOffset(startOffset);
+                    appendInfo.setLastOffset(duplicatedBatch.lastOffset);
+                    appendInfo.setMaxTimestamp(duplicatedBatch.timestamp);
+                    appendInfo.setStartOffsetOfMaxTimestamp(startOffset);
+                    appendInfo.setDuplicated(true);
+                } else {
+                    String errorMsg =
+                            String.format(
+                                    "Found duplicated batch for table bucket %s, duplicated offset is %s, "
+                                            + "writer id is %s and batch sequence is: %s",
+                                    getTableBucket(),
+                                    duplicatedBatch.lastOffset,
+                                    duplicatedBatch.writerId,
+                                    duplicatedBatch.batchSequence);
+                    LOG.error(errorMsg);
+                    throw new DuplicateSequenceException(errorMsg);
+                }
             } else {
                 // Append the records, and increment the local log end offset immediately after
                 // append because write to the transaction index below may fail, and we want to
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterStateEntry.java b/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterStateEntry.java
@@ -86,7 +86,8 @@ public long lastBatchTimestamp() {
     }
 
     public void addBath(int batchSequence, long lastOffset, int offsetDelta, long timestamp) {
-        addBatchMetadata(new BatchMetadata(batchSequence, lastOffset, offsetDelta, timestamp));
+        addBatchMetadata(
+                new BatchMetadata(writerId, batchSequence, lastOffset, offsetDelta, timestamp));
         this.lastTimestamp = timestamp;
     }
 
@@ -129,12 +130,19 @@ public WriterStateEntry withWriterIdAndBatchMetadata(
 
     /** Metadata of a batch. */
     public static final class BatchMetadata {
+        public final long writerId;
         public final int batchSequence;
         public final long lastOffset;
         public final int offsetDelta;
         public final long timestamp;
 
-        public BatchMetadata(int batchSequence, long lastOffset, int offsetDelta, long timestamp) {
+        public BatchMetadata(
+                long writerId,
+                int batchSequence,
+                long lastOffset,
+                int offsetDelta,
+                long timestamp) {
+            this.writerId = writerId;
             this.batchSequence = batchSequence;
             this.lastOffset = lastOffset;
             this.offsetDelta = offsetDelta;
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterStateManager.java b/fluss-server/src/main/java/com/alibaba/fluss/server/log/WriterStateManager.java
@@ -437,6 +437,7 @@ private static List<WriterStateEntry> readSnapshot(File file) {
                                             snapshotEntry.writerId,
                                             snapshotEntry.lastBatchTimestamp,
                                             new WriterStateEntry.BatchMetadata(
+                                                    snapshotEntry.writerId,
                                                     snapshotEntry.lastBatchSequence,
                                                     snapshotEntry.lastBatchBaseOffset,
                                                     snapshotEntry.lastBatchOffsetDelta,
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/Replica.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/Replica.java
@@ -506,6 +506,8 @@ public LogOffsetSnapshot fetchOffsetSnapshot(boolean fetchOnlyFromLeader) throws
     // -------------------------------------------------------------------------------------------
 
     private void onBecomeNewLeader() {
+        updateLeaderEndOffsetSnapshot();
+
         if (isKvTable()) {
             // if it's become new leader, we must
             // first destroy the old kv tablet
@@ -523,6 +525,11 @@ private void onBecomeNewFollower() {
         }
     }
 
+    @VisibleForTesting
+    public void updateLeaderEndOffsetSnapshot() {
+        logTablet.updateLeaderEndOffsetSnapshot();
+    }
+
     private void createKv() {
         try {
             // create a closeable registry for the closable related to kv
@@ -792,6 +799,10 @@ private void startPeriodicKvSnapshot(@Nullable CompletedSnapshot completedSnapsh
         }
     }
 
+    public long getLeaderEndOffsetSnapshot() {
+        return logTablet.getLeaderEndOffsetSnapshot();
+    }
+
     public LogAppendInfo appendRecordsToLeader(MemoryLogRecords memoryLogRecords, int requiredAcks)
             throws Exception {
         return inReadLock(
@@ -1235,6 +1246,8 @@ public long getOffset(RemoteLogManager remoteLogManager, ListOffsetsParam listOf
                         }
                     } else if (offsetType == ListOffsetsParam.LATEST_OFFSET_TYPE) {
                         return getLatestOffset(listOffsetsParam.getFollowerServerId());
+                    } else if (offsetType == ListOffsetsParam.LEADER_END_OFFSET_SNAPSHOT_TYPE) {
+                        return logTablet.getLeaderEndOffsetSnapshot();
                     } else {
                         throw new IllegalArgumentException(
                                 "Invalid list offset type: " + offsetType);
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/LeaderEndpoint.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/LeaderEndpoint.java
@@ -36,6 +36,8 @@ interface LeaderEndpoint {
     /** Fetches the local log start offset of the given table bucket. */
     CompletableFuture<Long> fetchLocalLogStartOffset(TableBucket tableBucket);
 
+    CompletableFuture<Long> fetchLeaderEndOffsetSnapshot(TableBucket tableBucket);
+
     /**
      * Given a fetchLogRequest, carries out the expected request and returns the results from
      * fetching from the leader.
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/RemoteLeaderEndpoint.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/RemoteLeaderEndpoint.java
@@ -81,6 +81,11 @@ public CompletableFuture<Long> fetchLocalLogStartOffset(TableBucket tableBucket)
         return fetchLogOffset(tableBucket, ListOffsetsParam.EARLIEST_OFFSET_TYPE);
     }
 
+    @Override
+    public CompletableFuture<Long> fetchLeaderEndOffsetSnapshot(TableBucket tableBucket) {
+        return fetchLogOffset(tableBucket, ListOffsetsParam.LEADER_END_OFFSET_SNAPSHOT_TYPE);
+    }
+
     @Override
     public CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> fetchLog(
             FetchLogRequest fetchLogRequest) {
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherThread.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherThread.java
@@ -17,7 +17,10 @@
 package com.alibaba.fluss.server.replica.fetcher;
 
 import com.alibaba.fluss.exception.CorruptRecordException;
+import com.alibaba.fluss.exception.DuplicateSequenceException;
+import com.alibaba.fluss.exception.InvalidOffsetException;
 import com.alibaba.fluss.exception.InvalidRecordException;
+import com.alibaba.fluss.exception.OutOfOrderSequenceException;
 import com.alibaba.fluss.exception.RemoteStorageException;
 import com.alibaba.fluss.exception.StorageException;
 import com.alibaba.fluss.metadata.TableBucket;
@@ -317,6 +320,27 @@ private void handleFetchLogResponseOfSuccessBucket(
                         currentFetchStatus.fetchOffset(),
                         e);
                 removeBucket(tableBucket);
+            } else if (e instanceof DuplicateSequenceException
+                    || e instanceof OutOfOrderSequenceException
+                    || e instanceof InvalidOffsetException) {
+                // TODO this part of logic need to be removed after we introduce leader epoch cache.
+                // Trace by https://github.com/alibaba/fluss/issues/673
+                LOG.error(
+                        "Founding recoverable error while processing data for bucket {} at offset {}, try to "
+                                + "truncate to LeaderEndOffsetSnapshot",
+                        tableBucket,
+                        currentFetchStatus.fetchOffset(),
+                        e);
+                try {
+                    truncateToLeaderEndOffsetSnapshot(tableBucket);
+                } catch (Exception ex) {
+                    LOG.error(
+                            "Error while truncating bucket {} at offset {}",
+                            tableBucket,
+                            currentFetchStatus.fetchOffset(),
+                            ex);
+                    removeBucket(tableBucket);
+                }
             } else {
                 LOG.error(
                         "Unexpected error occurred while processing data for bucket {} at offset {}",
@@ -328,6 +352,24 @@ private void handleFetchLogResponseOfSuccessBucket(
         }
     }
 
+    private void truncateToLeaderEndOffsetSnapshot(TableBucket tableBucket) throws Exception {
+        long leaderLocalEndOffsetWhileBecomeLeader =
+                leader.fetchLeaderEndOffsetSnapshot(tableBucket).get();
+        long localLogEndOffset =
+                replicaManager.getReplicaOrException(tableBucket).getLocalLogEndOffset();
+        if (leaderLocalEndOffsetWhileBecomeLeader != 0L
+                && leaderLocalEndOffsetWhileBecomeLeader < localLogEndOffset) {
+            // truncate to leaderEndOffsetSnapshot to reset follower's WriterState and fetch offset.
+            truncate(tableBucket, leaderLocalEndOffsetWhileBecomeLeader);
+
+            // update fetch status.
+            BucketFetchStatus bucketFetchStatus =
+                    new BucketFetchStatus(
+                            tableBucket.getTableId(), leaderLocalEndOffsetWhileBecomeLeader, null);
+            fairBucketStatusMap.updateAndMoveToEnd(tableBucket, bucketFetchStatus);
+        }
+    }
+
     private boolean handleOutOfRangeError(TableBucket tableBucket, BucketFetchStatus fetchStatus) {
         try {
             BucketFetchStatus newFetchStatus = fetchOffsetAndTruncate(tableBucket);
diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
@@ -74,6 +74,7 @@
 import static com.alibaba.fluss.server.zk.data.LeaderAndIsr.INITIAL_BUCKET_EPOCH;
 import static com.alibaba.fluss.server.zk.data.LeaderAndIsr.INITIAL_LEADER_EPOCH;
 import static com.alibaba.fluss.testutils.DataTestUtils.genMemoryLogRecordsByObject;
+import static com.alibaba.fluss.testutils.DataTestUtils.genMemoryLogRecordsWithWriterId;
 import static com.alibaba.fluss.testutils.common.CommonTestUtils.retry;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -199,6 +200,75 @@ void testFollowerHighWatermarkHigherThanOrEqualToLeader() throws Exception {
         }
     }
 
+    // TODO this test need to be removed after we introduce leader epoch cache. Trace by
+    // https://github.com/alibaba/fluss/issues/673
+    @Test
+    void testAppendAsFollowerThrowDuplicatedBatchException() throws Exception {
+        Replica leaderReplica = leaderRM.getReplicaOrException(tb);
+        Replica followerReplica = followerRM.getReplicaOrException(tb);
+
+        // 1. append same batches to leader and follower with different writer id.
+        CompletableFuture<List<ProduceLogResultForBucket>> future;
+        List<Long> writerIds = Arrays.asList(100L, 101L);
+        long baseOffset = 0L;
+        for (long writerId : writerIds) {
+            for (int i = 0; i < 5; i++) {
+                future = new CompletableFuture<>();
+                leaderRM.appendRecordsToLog(
+                        1000,
+                        1,
+                        Collections.singletonMap(
+                                tb, genMemoryLogRecordsWithWriterId(DATA1, writerId, i, 0)),
+                        future::complete);
+                assertThat(future.get())
+                        .containsOnly(
+                                new ProduceLogResultForBucket(tb, baseOffset, baseOffset + 10L));
+
+                followerReplica.appendRecordsToFollower(
+                        genMemoryLogRecordsWithWriterId(DATA1, writerId, i, baseOffset));
+                assertThat(followerReplica.getLocalLogEndOffset()).isEqualTo(baseOffset + 10L);
+                baseOffset = baseOffset + 10L;
+            }
+        }
+
+        // 2. append one batch to follower with (writerId=100L, batchSequence=5 offset=100L) to mock
+        // follower have one batch ahead of leader.
+        followerReplica.appendRecordsToFollower(
+                genMemoryLogRecordsWithWriterId(DATA1, 100L, 5, 100L));
+        assertThat(followerReplica.getLocalLogEndOffset()).isEqualTo(110L);
+
+        // 3. mock becomeLeaderAndFollower as follower end.
+        leaderReplica.updateLeaderEndOffsetSnapshot();
+        followerFetcher.addBuckets(
+                Collections.singletonMap(
+                        tb, new InitialFetchStatus(DATA1_TABLE_ID, leader.id(), 110L)));
+        followerFetcher.start();
+
+        // 4. mock append to leader with different writer id (writerId=101L, batchSequence=5
+        // offset=100L) to mock leader receive different batch from recovery follower.
+        future = new CompletableFuture<>();
+        leaderRM.appendRecordsToLog(
+                1000,
+                1,
+                Collections.singletonMap(tb, genMemoryLogRecordsWithWriterId(DATA1, 101L, 5, 100L)),
+                future::complete);
+        assertThat(future.get()).containsOnly(new ProduceLogResultForBucket(tb, 100L, 110L));
+
+        // 5. mock append to leader with (writerId=100L, batchSequence=5 offset=110L) to mock
+        // follower fetch duplicated batch from leader. In this case follower will truncate to
+        // LeaderEndOffsetSnapshot and fetch again.
+        future = new CompletableFuture<>();
+        leaderRM.appendRecordsToLog(
+                1000,
+                1,
+                Collections.singletonMap(tb, genMemoryLogRecordsWithWriterId(DATA1, 100L, 5, 110L)),
+                future::complete);
+        assertThat(future.get()).containsOnly(new ProduceLogResultForBucket(tb, 110L, 120L));
+        retry(
+                Duration.ofSeconds(20),
+                () -> assertThat(followerReplica.getLocalLogEndOffset()).isEqualTo(120L));
+    }
+
     private void registerTableInZkClient() throws Exception {
         ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().cleanupRoot();
         zkClient.registerTable(
diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/TestingLeaderEndpoint.java b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/TestingLeaderEndpoint.java
@@ -75,6 +75,12 @@ public CompletableFuture<Long> fetchLocalLogStartOffset(TableBucket tableBucket)
         return CompletableFuture.completedFuture(replica.getLocalLogStartOffset());
     }
 
+    @Override
+    public CompletableFuture<Long> fetchLeaderEndOffsetSnapshot(TableBucket tableBucket) {
+        Replica replica = replicaManager.getReplicaOrException(tableBucket);
+        return CompletableFuture.completedFuture(replica.getLeaderEndOffsetSnapshot());
+    }
+
     @Override
     public CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> fetchLog(
             FetchLogRequest fetchLogRequest) {
