[server] Fix ReplicaFetcherThread keeps throwing OutOfOrderSequenceException because of writer id expire

Author: LiebingYu

fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java

@@ -41,6 +41,8 @@ public class MetricNames {
     public static final String TABLE_COUNT = "tableCount";
     public static final String BUCKET_COUNT = "bucketCount";
     public static final String REPLICAS_TO_DELETE_COUNT = "replicasToDeleteCount";
+    public static final String PARTITION_COUNT = "partitionCount";
+    public static final String TOTAL_PARTITION_COUNT = "totalPartitionCount";
 
     // for coordinator event processor
     public static final String EVENT_QUEUE_SIZE = "eventQueueSize";

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java

@@ -104,6 +104,8 @@ public class CoordinatorContext {
     private ServerInfo coordinatorServerInfo = null;
     private int coordinatorEpoch = INITIAL_COORDINATOR_EPOCH;
 
+    private Runnable partitionCountUpdateCallback = null;
+
     public CoordinatorContext() {}
 
     public int getCoordinatorEpoch() {
@@ -232,6 +234,17 @@ public void putTableInfo(TableInfo tableInfo) {
     public void putPartition(long partitionId, PhysicalTablePath physicalTablePath) {
         this.pathByPartitionId.put(partitionId, physicalTablePath);
         this.partitionIdByPath.put(physicalTablePath, partitionId);
+        updatePartitionCountMetrics();
+    }
+
+    public void setPartitionCountUpdateCallback(Runnable callback) {
+        this.partitionCountUpdateCallback = callback;
+    }
+
+    private void updatePartitionCountMetrics() {
+        if (partitionCountUpdateCallback != null) {
+            partitionCountUpdateCallback.run();
+        }
     }
 
     public TableInfo getTableInfoById(long tableId) {
@@ -551,6 +564,23 @@ public Set<TablePartition> getPartitionsToBeDeleted() {
         return partitionsToBeDeleted;
     }
 
+    public int getTotalPartitionCount() {
+        return partitionAssignments.size();
+    }
+
+    public int getPartitionCountForTable(long tableId) {
+        return (int)
+                partitionAssignments.keySet().stream()
+                        .filter(partition -> partition.getTableId() == tableId)
+                        .count();
+    }
+
+    public Set<TablePartition> getPartitionsForTable(long tableId) {
+        return partitionAssignments.keySet().stream()
+                .filter(partition -> partition.getTableId() == tableId)
+                .collect(Collectors.toSet());
+    }
+
     public boolean isToBeDeleted(TableBucket tableBucket) {
         if (tableBucket.getPartitionId() == null) {
             return isTableQueuedForDeletion(tableBucket.getTableId());
@@ -614,6 +644,8 @@ public void removePartition(TablePartition tablePartition) {
         if (physicalTablePath != null) {
             partitionIdByPath.remove(physicalTablePath);
         }
+        // Update partition count metrics when partition is removed
+        updatePartitionCountMetrics();
     }
 
     private void clearTablesState() {

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java

@@ -176,6 +176,16 @@ protected void startServices() throws Exception {
             this.coordinatorContext = new CoordinatorContext();
             this.metadataCache = new CoordinatorMetadataCache();
 
+            coordinatorContext.setPartitionCountUpdateCallback(
+                    () -> {
+                        if (serverMetricGroup != null) {
+                            int totalPartitionCount = coordinatorContext.getTotalPartitionCount();
+                            ((CoordinatorMetricGroup.MutableGauge<Integer>)
+                                            serverMetricGroup.totalPartitionCount())
+                                    .setValue(totalPartitionCount);
+                        }
+                    });
+
             this.authorizer = AuthorizerLoader.createAuthorizer(conf, zkClient, pluginManager);
             if (authorizer != null) {
                 authorizer.startup();

fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java

@@ -643,7 +643,7 @@ private LogAppendInfo append(MemoryLogRecords records, boolean appendAsLeader)
             // now that we have valid records, offsets assigned, we need to validate the idempotent
             // state of the writers and collect some metadata.
             Either<WriterStateEntry.BatchMetadata, Collection<WriterAppendInfo>> validateResult =
-                    analyzeAndValidateWriterState(validRecords);
+                    analyzeAndValidateWriterState(validRecords, appendAsLeader);
 
             if (validateResult.isLeft()) {
                 // have duplicated batch metadata, skip the append and update append info.
@@ -1006,7 +1006,7 @@ private LogAppendInfo analyzeAndValidateRecords(MemoryLogRecords records) {
 
     /** Returns either the duplicated batch metadata (left) or the updated writers (right). */
     private Either<WriterStateEntry.BatchMetadata, Collection<WriterAppendInfo>>
-            analyzeAndValidateWriterState(MemoryLogRecords records) {
+            analyzeAndValidateWriterState(MemoryLogRecords records, boolean isAppendAsLeader) {
         Map<Long, WriterAppendInfo> updatedWriters = new HashMap<>();
 
         for (LogRecordBatch batch : records.batches()) {
@@ -1023,14 +1023,15 @@ private LogAppendInfo analyzeAndValidateRecords(MemoryLogRecords records) {
                 }
 
                 // update write append info.
-                updateWriterAppendInfo(writerStateManager, batch, updatedWriters);
+                updateWriterAppendInfo(writerStateManager, batch, updatedWriters, isAppendAsLeader);
             }
         }
 
         return Either.right(updatedWriters.values());
     }
 
-    void removeExpiredWriter(long currentTimeMs) {
+    @VisibleForTesting
+    public void removeExpiredWriter(long currentTimeMs) {
         synchronized (lock) {
             writerStateManager.removeExpiredWriters(currentTimeMs);
         }
@@ -1110,14 +1111,16 @@ private void deleteSegments(List<LogSegment> deletableSegments, SegmentDeletionR
     private static void updateWriterAppendInfo(
             WriterStateManager writerStateManager,
             LogRecordBatch batch,
-            Map<Long, WriterAppendInfo> writers) {
+            Map<Long, WriterAppendInfo> writers,
+            boolean isAppendAsLeader) {
         long writerId = batch.writerId();
         // update writers.
         WriterAppendInfo appendInfo =
                 writers.computeIfAbsent(writerId, id -> writerStateManager.prepareUpdate(writerId));
         appendInfo.append(
                 batch,
-                writerStateManager.isWriterInBatchExpired(System.currentTimeMillis(), batch));
+                writerStateManager.isWriterInBatchExpired(System.currentTimeMillis(), batch),
+                isAppendAsLeader);
     }
 
     static void rebuildWriterState(
@@ -1230,7 +1233,7 @@ private static void loadWritersFromRecords(
         Map<Long, WriterAppendInfo> loadedWriters = new HashMap<>();
         for (LogRecordBatch batch : records.batches()) {
             if (batch.hasWriterId()) {
-                updateWriterAppendInfo(writerStateManager, batch, loadedWriters);
+                updateWriterAppendInfo(writerStateManager, batch, loadedWriters, true);
             }
         }
         loadedWriters.values().forEach(writerStateManager::update);

fluss-server/src/main/java/org/apache/fluss/server/log/WriterAppendInfo.java

@@ -44,13 +44,15 @@ public long writerId() {
         return writerId;
     }
 
-    public void append(LogRecordBatch batch, boolean isWriterInBatchExpired) {
+    public void append(
+            LogRecordBatch batch, boolean isWriterInBatchExpired, boolean isAppendAsLeader) {
         LogOffsetMetadata firstOffsetMetadata = new LogOffsetMetadata(batch.baseLogOffset());
         appendDataBatch(
                 batch.batchSequence(),
                 firstOffsetMetadata,
                 batch.lastLogOffset(),
                 isWriterInBatchExpired,
+                isAppendAsLeader,
                 batch.commitTimestamp());
     }
 
@@ -59,8 +61,9 @@ public void appendDataBatch(
             LogOffsetMetadata firstOffsetMetadata,
             long lastOffset,
             boolean isWriterInBatchExpired,
+            boolean isAppendAsLeader,
             long batchTimestamp) {
-        maybeValidateDataBatch(batchSequence, isWriterInBatchExpired, lastOffset);
+        maybeValidateDataBatch(batchSequence, isWriterInBatchExpired, lastOffset, isAppendAsLeader);
         updatedEntry.addBath(
                 batchSequence,
                 lastOffset,
@@ -69,13 +72,16 @@ public void appendDataBatch(
     }
 
     private void maybeValidateDataBatch(
-            int appendFirstSeq, boolean isWriterInBatchExpired, long lastOffset) {
+            int appendFirstSeq,
+            boolean isWriterInBatchExpired,
+            long lastOffset,
+            boolean isAppendAsLeader) {
         int currentLastSeq =
                 !updatedEntry.isEmpty()
                         ? updatedEntry.lastBatchSequence()
                         : currentEntry.lastBatchSequence();
         // must be in sequence, even for the first batch should start from 0
-        if (!inSequence(currentLastSeq, appendFirstSeq, isWriterInBatchExpired)) {
+        if (!inSequence(currentLastSeq, appendFirstSeq, isWriterInBatchExpired, isAppendAsLeader)) {
             throw new OutOfOrderSequenceException(
                     String.format(
                             "Out of order batch sequence for writer %s at offset %s in "
@@ -93,16 +99,53 @@ public WriterStateEntry toEntry() {
      * three scenarios will be judged as in sequence:
      *
      * <ul>
-     *   <li>If lastBatchSeq equals NO_BATCH_SEQUENCE, we need to check whether the committed
-     *       timestamp of the next batch under the current writerId has expired. If it has expired,
-     *       we consider this a special case caused by writerId expiration, for this case, to ensure
-     *       the correctness of follower sync, we still treat it as in sequence.
+     *   <li>If lastBatchSeq equals NO_BATCH_SEQUENCE, the following two scenarios will be judged as
+     *       in sequence:
+     *       <ul>
+     *         <li>If the committed timestamp of the next batch under the current writerId has
+     *             expired, we consider this a special case caused by writerId expiration, for this
+     *             case, to ensure the correctness of follower sync, we still treat it as in
+     *             sequence.
+     *         <li>If the append request is from the follower, we consider this is a special case
+     *             caused by inconsistent expiration of writerId between the leader and follower. To
+     *             prevent continuous fetch failures on the follower side, we still treat it as in
+     *             sequence. Here is a detailed example: The expiration of a writer is triggered
+     *             asynchronously by the {@code PeriodicWriterIdExpirationCheck} thread at intervals
+     *             defined by {@code server.writer-id.expiration-check-interval}, which can result
+     *             in slight differences in the actual expiration times of the same writer on the
+     *             leader replica and follower replicas. This slight difference leads to a dreadful
+     *             corner case. Imagine the following scenario(set {@code
+     *             server.writer-id.expiration-check-interval}: 10min, {@code
+     *             server.writer-id.expiration-time}: 12h):
+     *             <pre>{@code
+     * Step     Time         Action of Leader                  Action of Follower
+     * 1        00:03:38     receive batch 0 of writer 101
+     * 2        00:03:38                                       fetch batch 0 of writer 101
+     * 3        12:05:00                                       remove state of writer 101
+     * 4        12:10:02     receive batch 1 of writer 101
+     * 5        12:10:02                                       fetch batch 0 of writer 101
+     * 6        12:11:00     remove state of writer 101
+     *
+     * }</pre>
+     *             In step 3, the follower removes the state of writer 101 first, since it has been
+     *             more than 12 hours since writer 101's last batch write, making it safe to remove.
+     *             However, since the expiration of writer 101 has not yet occurred on the leader,
+     *             and a new batch 1 is received at this time, it is successfully written on the
+     *             leader. At this point, the fetcher pulls batch 1 from the leader, but since the
+     *             state of writer 101 has already been cleaned up, an {@link
+     *             OutOfOrderSequenceException} will occur during to write if we don't treat it as
+     *             in sequence.
+     *       </ul>
      *   <li>nextBatchSeq == lastBatchSeq + 1L
      *   <li>lastBatchSeq reaches its maximum value
      * </ul>
      */
-    private boolean inSequence(int lastBatchSeq, int nextBatchSeq, boolean isWriterInBatchExpired) {
-        return (lastBatchSeq == NO_BATCH_SEQUENCE && isWriterInBatchExpired)
+    private boolean inSequence(
+            int lastBatchSeq,
+            int nextBatchSeq,
+            boolean isWriterInBatchExpired,
+            boolean isAppendAsLeader) {
+        return (lastBatchSeq == NO_BATCH_SEQUENCE && (isWriterInBatchExpired || !isAppendAsLeader))
                 || nextBatchSeq == lastBatchSeq + 1L
                 || (nextBatchSeq == 0 && lastBatchSeq == Integer.MAX_VALUE);
     }

fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java

@@ -25,6 +25,8 @@
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.server.coordinator.MetadataManager;
+import org.apache.fluss.server.metrics.group.PhysicalTableMetricGroup;
+import org.apache.fluss.server.metrics.group.TabletServerMetricGroup;
 import org.apache.fluss.server.tablet.TabletServer;
 import org.apache.fluss.server.zk.ZooKeeperClient;
 
@@ -68,11 +70,16 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
 
     private final MetadataManager metadataManager;
     private final ZooKeeperClient zkClient;
+    private final TabletServerMetricGroup metricGroup;
 
-    public TabletServerMetadataCache(MetadataManager metadataManager, ZooKeeperClient zkClient) {
+    public TabletServerMetadataCache(
+            MetadataManager metadataManager,
+            ZooKeeperClient zkClient,
+            TabletServerMetricGroup metricGroup) {
         this.serverMetadataSnapshot = ServerMetadataSnapshot.empty();
         this.metadataManager = metadataManager;
         this.zkClient = zkClient;
+        this.metricGroup = metricGroup;
     }
 
     @Override
@@ -269,9 +276,36 @@ public void updateClusterMetadata(ClusterMetadata clusterMetadata) {
                                     partitionIdByPath,
                                     bucketMetadataMapForTables,
                                     bucketMetadataMapForPartitions);
+
+                    // Update partition count metrics
+                    updatePartitionCountMetrics(partitionIdByPath);
                 });
     }
 
+    private void updatePartitionCountMetrics(Map<PhysicalTablePath, Long> partitionIdByPath) {
+        if (metricGroup != null) {
+            Map<TablePath, Integer> partitionCountByTable = new HashMap<>();
+            for (PhysicalTablePath physicalTablePath : partitionIdByPath.keySet()) {
+                TablePath tablePath = physicalTablePath.getTablePath();
+                partitionCountByTable.merge(tablePath, 1, Integer::sum);
+            }
+
+            for (Map.Entry<TablePath, Integer> entry : partitionCountByTable.entrySet()) {
+                TablePath tablePath = entry.getKey();
+                int partitionCount = entry.getValue();
+
+                PhysicalTablePath physicalTablePath = PhysicalTablePath.of(tablePath, null);
+                PhysicalTableMetricGroup tableMetricGroup =
+                        metricGroup.getPhysicalTableMetricGroup(physicalTablePath);
+                if (tableMetricGroup != null) {
+                    ((PhysicalTableMetricGroup.MutableGauge<Integer>)
+                                    tableMetricGroup.partitionCount())
+                            .setValue(partitionCount);
+                }
+            }
+        }
+    }
+
     @VisibleForTesting
     public void clearTableMetadata() {
         inLock(

fluss-server/src/main/java/org/apache/fluss/server/metrics/group/CoordinatorMetricGroup.java

@@ -18,6 +18,8 @@
 package org.apache.fluss.server.metrics.group;
 
 import org.apache.fluss.metrics.CharacterFilter;
+import org.apache.fluss.metrics.Gauge;
+import org.apache.fluss.metrics.MetricNames;
 import org.apache.fluss.metrics.groups.AbstractMetricGroup;
 import org.apache.fluss.metrics.registry.MetricRegistry;
 
@@ -32,12 +34,17 @@ public class CoordinatorMetricGroup extends AbstractMetricGroup {
     protected final String hostname;
     protected final String serverId;
 
+    private final MutableGauge<Integer> totalPartitionCount;
+
     public CoordinatorMetricGroup(
             MetricRegistry registry, String clusterId, String hostname, String serverId) {
         super(registry, new String[] {clusterId, hostname, NAME}, null);
         this.clusterId = clusterId;
         this.hostname = hostname;
         this.serverId = serverId;
+
+        this.totalPartitionCount = new MutableGauge<>(0);
+        gauge(MetricNames.TOTAL_PARTITION_COUNT, totalPartitionCount);
     }
 
     @Override
@@ -51,4 +58,25 @@ protected final void putVariables(Map<String, String> variables) {
         variables.put("host", hostname);
         variables.put("server_id", serverId);
     }
+
+    public Gauge<Integer> totalPartitionCount() {
+        return totalPartitionCount;
+    }
+
+    public static class MutableGauge<T> implements Gauge<T> {
+        private volatile T value;
+
+        public MutableGauge(T initialValue) {
+            this.value = initialValue;
+        }
+
+        @Override
+        public T getValue() {
+            return value;
+        }
+
+        public void setValue(T value) {
+            this.value = value;
+        }
+    }
 }