address baiye's comments

swuferhong · swuferhong · commit 72f9e6165805 · 2026-01-28T15:00:47.000+08:00
diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
@@ -73,6 +73,13 @@ public class MetricNames {
     public static final String SERVER_LOGICAL_STORAGE_KV_SIZE = "kvSize";
     public static final String SERVER_PHYSICAL_STORAGE_LOCAL_SIZE = "localSize";
     public static final String SERVER_PHYSICAL_STORAGE_REMOTE_LOG_SIZE = "remoteLogSize";
+
+    /**
+     * Total physical storage size of standby replica snapshots on this tablet server.
+     *
+     * <p>Standby replicas maintain downloaded KV snapshots for fast leader failover. This metric
+     * tracks the total size of all KV snapshot data downloaded by standby replicas on this server.
+     */
     public static final String SERVER_PHYSICAL_STORAGE_STANDBY_SIZE = "standbySize";
 
     // --------------------------------------------------------------------------------------------
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotManager.java
@@ -137,6 +137,8 @@ public void becomeLeader() {
 
     public void becomeFollower() {
         isLeader = false;
+        // Clear standby download cache when leaving standby role
+        clearStandbyDownloadCache();
     }
 
     public void becomeStandby() {
@@ -153,6 +155,24 @@ public void becomeStandby() {
         return downloadedMiscFiles;
     }
 
+    /**
+     * Clear the standby download cache.
+     *
+     * <p>This method should be called when a replica leaves the standby role (becomes a regular
+     * follower or leader). It clears the cached state of downloaded SST files, misc files, and
+     * snapshot size. This ensures that if the replica becomes standby again later, it will perform
+     * a fresh download based on the actual local files, rather than reusing stale cache that
+     * references deleted files.
+     */
+    private void clearStandbyDownloadCache() {
+        downloadedSstFiles = null;
+        downloadedMiscFiles = null;
+        standbySnapshotSize = 0;
+        LOG.info(
+                "Cleared standby download cache for table bucket {}, will reload from local files on next standby promotion",
+                tableBucket);
+    }
+
     /**
      * The guardedExecutor is an executor that uses to trigger upload snapshot.
      *
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
@@ -85,6 +85,7 @@
 import org.apache.fluss.server.zk.data.LeaderAndIsr;
 import org.apache.fluss.server.zk.data.ZkData;
 import org.apache.fluss.utils.CloseableRegistry;
+import org.apache.fluss.utils.FileUtils;
 import org.apache.fluss.utils.FlussPaths;
 import org.apache.fluss.utils.IOUtils;
 import org.apache.fluss.utils.MapUtils;
@@ -98,6 +99,7 @@
 import javax.annotation.concurrent.ThreadSafe;
 
 import java.io.Closeable;
+import java.io.File;
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.ArrayList;
@@ -249,7 +251,8 @@ public Replica(
             this.kvSnapshotManager =
                     KvSnapshotManager.create(
                             tableBucket,
-                            kvManager.getOrCreateTabletDir(physicalPath, tableBucket),
+                            kvManager.getOrCreateTabletDir(
+                                    physicalPath, tableBucket), // kv tablet dir always built.
                             snapshotContext,
                             clock);
         }
@@ -583,10 +586,14 @@ private void onBecomeNewFollower(int standbyReplica) {
                 // to be new follower.
                 kvSnapshotManager.becomeFollower();
                 if (wasStandby || wasLeader) {
-                    // standby -> leader or leader -> leader
+                    // standby -> follower or leader -> follower
                     dropKv();
                 }
 
+                if (wasStandby) {
+                    isStandbyReplica = false;
+                }
+
                 // follower -> follower: do nothing.
             }
         }
@@ -692,6 +699,25 @@ private void dropKv() {
             checkNotNull(kvManager);
             kvManager.dropKv(tableBucket);
             kvTablet = null;
+        } else {
+            // For standby replicas, kvTablet is null, so we need to manually delete the
+            // downloaded snapshot files and the tablet directory.
+            // For follower replicas, we need to delete the empty directory.
+            checkNotNull(kvSnapshotManager);
+            File tabletDir = kvSnapshotManager.getTabletDir();
+            try {
+                FileUtils.deleteDirectory(tabletDir);
+                LOG.info(
+                        "Deleted standby replica snapshot directory {} for table bucket {}",
+                        tabletDir,
+                        tableBucket);
+            } catch (IOException e) {
+                LOG.warn(
+                        "Failed to delete standby replica snapshot directory {} for table bucket {}",
+                        tabletDir,
+                        tableBucket,
+                        e);
+            }
         }
     }
 
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
@@ -788,35 +788,48 @@ public void notifyRemoteLogOffsets(
     public void notifyKvSnapshotOffset(
             NotifyKvSnapshotOffsetData notifyKvSnapshotOffsetData,
             Consumer<NotifyKvSnapshotOffsetResponse> responseCallback) {
-        inLock(
-                replicaStateChangeLock,
-                () -> {
-                    // check or apply coordinator epoch.
-                    validateAndApplyCoordinatorEpoch(
-                            notifyKvSnapshotOffsetData.getCoordinatorEpoch(),
-                            "notifyKvSnapshotOffset");
-                    // update the snapshot offset.
-                    TableBucket tb = notifyKvSnapshotOffsetData.getTableBucket();
-                    Replica replica = getReplicaOrException(tb);
-                    if (notifyKvSnapshotOffsetData.getSnapshotId() != null) {
-                        // try to download the snapshot for standby replica.
+        TableBucket tb = notifyKvSnapshotOffsetData.getTableBucket();
+        Long snapshotId = notifyKvSnapshotOffsetData.getSnapshotId();
+        long minRetainOffset = notifyKvSnapshotOffsetData.getMinRetainOffset();
+
+        // Validate and get replica under lock
+        Replica replica =
+                inLock(
+                        replicaStateChangeLock,
+                        () -> {
+                            // check or apply coordinator epoch.
+                            validateAndApplyCoordinatorEpoch(
+                                    notifyKvSnapshotOffsetData.getCoordinatorEpoch(),
+                                    "notifyKvSnapshotOffset");
+                            return getReplicaOrException(tb);
+                        });
+
+        // Respond immediately to avoid blocking coordinator
+        responseCallback.accept(new NotifyKvSnapshotOffsetResponse());
+
+        // Download snapshot asynchronously outside the lock to avoid blocking
+        // leader/follower transitions and other state changes
+        if (snapshotId != null) {
+            ioExecutor.execute(
+                    () -> {
                         try {
-                            replica.downloadSnapshot(notifyKvSnapshotOffsetData.getSnapshotId());
-                            updateMinRetainOffset(
-                                    replica, notifyKvSnapshotOffsetData.getMinRetainOffset());
+                            replica.downloadSnapshot(snapshotId);
+                            updateMinRetainOffset(replica, minRetainOffset);
+                            LOG.info(
+                                    "Successfully downloaded snapshot {} for standby replica {}.",
+                                    snapshotId,
+                                    tb);
                         } catch (Exception e) {
                             LOG.error(
                                     "Error downloading snapshot id {} for standby replica {}.",
-                                    notifyKvSnapshotOffsetData.getSnapshotId(),
+                                    snapshotId,
                                     tb,
                                     e);
                         }
-                    } else {
-                        updateMinRetainOffset(
-                                replica, notifyKvSnapshotOffsetData.getMinRetainOffset());
-                    }
-                    responseCallback.accept(new NotifyKvSnapshotOffsetResponse());
-                });
+                    });
+        } else {
+            updateMinRetainOffset(replica, minRetainOffset);
+        }
     }
 
     private void updateMinRetainOffset(Replica replica, long minRetainOffset) {
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvSnapshotITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvSnapshotITCase.java
@@ -31,6 +31,7 @@
 import org.apache.fluss.server.kv.snapshot.ZooKeeperCompletedSnapshotHandleStore;
 import org.apache.fluss.server.tablet.TabletServer;
 import org.apache.fluss.server.testutils.FlussClusterExtension;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
 import org.apache.fluss.utils.FlussPaths;
 import org.apache.fluss.utils.types.Tuple2;
 
@@ -41,6 +42,7 @@
 import java.io.File;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -435,6 +437,187 @@ void testStandbyIncrementalSnapshotDownload() throws Exception {
         assertThat(standbySnapshotManager.getStandbySnapshotSize()).isGreaterThan(0);
     }
 
+    @Test
+    void testStandbyDemotionAndReStandby() throws Exception {
+        TablePath tablePath = TablePath.of("test_db", "test_table_standby_demotion");
+        long tableId = createTable(FLUSS_CLUSTER_EXTENSION, tablePath, DATA1_TABLE_DESCRIPTOR_PK);
+        TableBucket tb0 = new TableBucket(tableId, 0);
+
+        FLUSS_CLUSTER_EXTENSION.waitAndGetLeaderReplica(tb0);
+        int leaderServer = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb0);
+
+        // put some data and wait for snapshot
+        KvRecordBatch kvRecordBatch =
+                genKvRecordBatch(
+                        Tuple2.of("k1", new Object[] {1, "k1"}),
+                        Tuple2.of("k2", new Object[] {2, "k2"}));
+        PutKvRequest putKvRequest = newPutKvRequest(tableId, 0, -1, kvRecordBatch);
+        TabletServerGateway leaderGateway =
+                FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leaderServer);
+        leaderGateway.putKv(putKvRequest).get();
+
+        // wait for snapshot
+        waitValue(
+                () -> completedSnapshotHandleStore.get(tb0, 0),
+                Duration.ofMinutes(2),
+                "Fail to wait for snapshot 0");
+
+        // verify standby replica has downloaded snapshot
+        int standbyServer = FLUSS_CLUSTER_EXTENSION.waitAndGetStandby(tb0);
+        TabletServer standbyTs = FLUSS_CLUSTER_EXTENSION.getTabletServerById(standbyServer);
+        Replica standbyReplica = standbyTs.getReplicaManager().getReplicaOrException(tb0);
+        assertThat(standbyReplica.isStandby()).isTrue();
+
+        KvSnapshotManager standbySnapshotManager = standbyReplica.getKvSnapshotManager();
+        retry(
+                Duration.ofMinutes(1),
+                () -> assertThat(standbySnapshotManager.getDownloadedSstFiles()).isNotEmpty());
+
+        // get current leader and isr
+        LeaderAndIsr leaderAndIsr = FLUSS_CLUSTER_EXTENSION.waitLeaderAndIsrReady(tb0);
+        List<Integer> replicas = new ArrayList<>(leaderAndIsr.isr());
+
+        // demote standby to regular follower by sending notifyLeaderAndIsr without standby
+        LeaderAndIsr newLeaderAndIsr =
+                new LeaderAndIsr(
+                        leaderAndIsr.leader(),
+                        leaderAndIsr.leaderEpoch() + 1,
+                        leaderAndIsr.isr(),
+                        Collections.emptyList(), // no standby replicas
+                        leaderAndIsr.coordinatorEpoch(),
+                        leaderAndIsr.bucketEpoch());
+
+        FLUSS_CLUSTER_EXTENSION.notifyLeaderAndIsr(
+                standbyServer, tablePath, tb0, newLeaderAndIsr, replicas);
+
+        // verify the replica is no longer standby
+        retry(
+                Duration.ofMinutes(1),
+                () -> {
+                    Replica replica = standbyTs.getReplicaManager().getReplicaOrException(tb0);
+                    assertThat(replica.isStandby()).isFalse();
+                    // verify kv tablet is dropped when demoted from standby
+                    assertThat(replica.getKvTablet()).isNull();
+                    // verify standby download cache is cleared
+                    KvSnapshotManager snapshotManager = replica.getKvSnapshotManager();
+                    assertThat(snapshotManager.getDownloadedSstFiles()).isNull();
+                    assertThat(snapshotManager.getDownloadedMiscFiles()).isNull();
+                    assertThat(snapshotManager.getStandbySnapshotSize()).isEqualTo(0);
+                });
+
+        // put more data and create another snapshot
+        kvRecordBatch =
+                genKvRecordBatch(
+                        Tuple2.of("k3", new Object[] {3, "k3"}),
+                        Tuple2.of("k4", new Object[] {4, "k4"}));
+        putKvRequest = newPutKvRequest(tableId, 0, 1, kvRecordBatch);
+        leaderGateway.putKv(putKvRequest).get();
+
+        waitValue(
+                () -> completedSnapshotHandleStore.get(tb0, 1),
+                Duration.ofMinutes(2),
+                "Fail to wait for snapshot 1");
+
+        // re-promote the replica back to standby by sending notifyLeaderAndIsr with standby
+        LeaderAndIsr reStandbyLeaderAndIsr =
+                new LeaderAndIsr(
+                        newLeaderAndIsr.leader(),
+                        newLeaderAndIsr.leaderEpoch() + 1,
+                        newLeaderAndIsr.isr(),
+                        Collections.singletonList(standbyServer),
+                        newLeaderAndIsr.coordinatorEpoch(),
+                        newLeaderAndIsr.bucketEpoch()); // re-add as standby
+
+        FLUSS_CLUSTER_EXTENSION.notifyLeaderAndIsr(
+                standbyServer, tablePath, tb0, reStandbyLeaderAndIsr, replicas);
+
+        // verify the replica is standby again and downloads the latest snapshot
+        retry(
+                Duration.ofMinutes(1),
+                () -> {
+                    Replica replica = standbyTs.getReplicaManager().getReplicaOrException(tb0);
+                    assertThat(replica.isStandby()).isTrue();
+                    KvSnapshotManager snapshotManager = replica.getKvSnapshotManager();
+                    assertThat(snapshotManager).isNotNull();
+                    assertThat(snapshotManager.getDownloadedSstFiles()).isNotEmpty();
+                    // verify it has the latest snapshot data
+                    assertThat(snapshotManager.getStandbySnapshotSize()).isGreaterThan(0);
+                });
+    }
+
+    @Test
+    void testStandbySnapshotDownloadFailureAndRecovery() throws Exception {
+        TablePath tablePath = TablePath.of("test_db", "test_table_download_failure");
+        long tableId = createTable(FLUSS_CLUSTER_EXTENSION, tablePath, DATA1_TABLE_DESCRIPTOR_PK);
+        TableBucket tb0 = new TableBucket(tableId, 0);
+
+        FLUSS_CLUSTER_EXTENSION.waitAndGetLeaderReplica(tb0);
+        int leaderServer = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb0);
+
+        // put data and create snapshot
+        KvRecordBatch kvRecordBatch =
+                genKvRecordBatch(
+                        Tuple2.of("k1", new Object[] {1, "k1"}),
+                        Tuple2.of("k2", new Object[] {2, "k2"}));
+        PutKvRequest putKvRequest = newPutKvRequest(tableId, 0, -1, kvRecordBatch);
+        TabletServerGateway leaderGateway =
+                FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leaderServer);
+        leaderGateway.putKv(putKvRequest).get();
+
+        // wait for first snapshot
+        CompletedSnapshot snapshot0 =
+                waitValue(
+                                () -> completedSnapshotHandleStore.get(tb0, 0),
+                                Duration.ofMinutes(2),
+                                "Fail to wait for snapshot 0")
+                        .retrieveCompleteSnapshot();
+
+        // get standby server and verify it's marked as standby even if download fails initially
+        int standbyServer = FLUSS_CLUSTER_EXTENSION.waitAndGetStandby(tb0);
+        TabletServer standbyTs = FLUSS_CLUSTER_EXTENSION.getTabletServerById(standbyServer);
+        Replica standbyReplica = standbyTs.getReplicaManager().getReplicaOrException(tb0);
+
+        // verify replica is marked as standby (even if download might fail)
+        retry(Duration.ofMinutes(1), () -> assertThat(standbyReplica.isStandby()).isTrue());
+
+        // wait for standby to eventually download snapshot (may retry on failures)
+        KvSnapshotManager standbySnapshotManager = standbyReplica.getKvSnapshotManager();
+        assertThat(standbySnapshotManager).isNotNull();
+
+        // verify snapshot is eventually downloaded successfully
+        retry(
+                Duration.ofMinutes(1),
+                () -> {
+                    assertThat(standbySnapshotManager.getDownloadedSstFiles()).isNotEmpty();
+                    assertThat(standbySnapshotManager.getStandbySnapshotSize())
+                            .isEqualTo(snapshot0.getSnapshotSize());
+                });
+
+        // create another snapshot to verify recovery continues to work
+        kvRecordBatch =
+                genKvRecordBatch(
+                        Tuple2.of("k3", new Object[] {3, "k3"}),
+                        Tuple2.of("k4", new Object[] {4, "k4"}));
+        putKvRequest = newPutKvRequest(tableId, 0, 1, kvRecordBatch);
+        leaderGateway.putKv(putKvRequest).get();
+
+        CompletedSnapshot snapshot1 =
+                waitValue(
+                                () -> completedSnapshotHandleStore.get(tb0, 1),
+                                Duration.ofMinutes(2),
+                                "Fail to wait for snapshot 1")
+                        .retrieveCompleteSnapshot();
+
+        // verify standby continues to download new snapshots after recovery
+        retry(
+                Duration.ofMinutes(1),
+                () -> {
+                    assertThat(standbySnapshotManager.getDownloadedSstFiles()).isNotEmpty();
+                    assertThat(standbySnapshotManager.getStandbySnapshotSize())
+                            .isEqualTo(snapshot1.getSnapshotSize());
+                });
+    }
+
     private void checkDirsDeleted(Set<File> bucketDirs, Map<Long, TablePath> tablePathMap) {
         for (File bucketDir : bucketDirs) {
             retry(Duration.ofMinutes(1), () -> assertThat(bucketDir.exists()).isFalse());
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
