fix claude review comments

Author: swuferhong

fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotManager.java

@@ -240,6 +240,13 @@ public void downloadSnapshot(long snapshotId) throws Exception {
         }
         CompletedSnapshot completedSnapshot =
                 snapshotContext.getCompletedSnapshotProvider(tableBucket, snapshotId);
+        if (completedSnapshot == null) {
+            LOG.warn(
+                    "Snapshot {} not found for bucket {}, skip downloading.",
+                    snapshotId,
+                    tableBucket);
+            return;
+        }
         incrementalDownloadSnapshot(completedSnapshot);
         standbySnapshotSize = completedSnapshot.getSnapshotSize();
     }
@@ -253,23 +260,26 @@ public void downloadSnapshot(long snapshotId) throws Exception {
      * @return the latest snapshot
      */
     public Optional<CompletedSnapshot> downloadLatestSnapshot() throws Exception {
-        // standbyInitializing is used to prevent concurrent download.
+        // standbyInitializing is used to prevent concurrent download via
+        // downloadSnapshot(snapshotId).
         standbyInitializing = true;
+        try {
+            // Note: no isStandby check here. This method is called from both:
+            // 1. initKvTablet() during leader initialization - isStandby is already false
+            // 2. becomeStandbyAsync() during standby initialization - isStandby is true
+            // The isStandby guard is only needed in downloadSnapshot(snapshotId) which is
+            // called from the notification path exclusively for standby replicas.
+            Optional<CompletedSnapshot> latestSnapshot = getLatestSnapshot();
+            if (latestSnapshot.isPresent()) {
+                CompletedSnapshot completedSnapshot = latestSnapshot.get();
+                incrementalDownloadSnapshot(completedSnapshot);
+                standbySnapshotSize = completedSnapshot.getSnapshotSize();
+            }
 
-        // Note: no isStandby check here. This method is called from both:
-        // 1. initKvTablet() during leader initialization - isStandby is already false
-        // 2. becomeStandbyAsync() during standby initialization - isStandby is true
-        // The isStandby guard is only needed in downloadSnapshot(snapshotId) which is
-        // called from the notification path exclusively for standby replicas.
-        Optional<CompletedSnapshot> latestSnapshot = getLatestSnapshot();
-        if (latestSnapshot.isPresent()) {
-            CompletedSnapshot completedSnapshot = latestSnapshot.get();
-            incrementalDownloadSnapshot(completedSnapshot);
-            standbySnapshotSize = completedSnapshot.getSnapshotSize();
+            return latestSnapshot;
+        } finally {
+            standbyInitializing = false;
         }
-
-        standbyInitializing = false;
-        return latestSnapshot;
     }
 
     private void incrementalDownloadSnapshot(CompletedSnapshot completedSnapshot) throws Exception {

fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java

@@ -1052,7 +1052,6 @@ public void notifyKvSnapshotOffset(
                     () -> {
                         try {
                             replica.downloadSnapshot(snapshotId);
-                            updateMinRetainOffset(replica, minRetainOffset);
                             LOG.debug(
                                     "Successfully downloaded snapshot {} for standby replica {}.",
                                     snapshotId,
@@ -1063,6 +1062,11 @@ public void notifyKvSnapshotOffset(
                                     snapshotId,
                                     tb,
                                     e);
+                        } finally {
+                            // Always update minRetainOffset regardless of download success/failure.
+                            // If we skip this, log segments may never be cleaned up when download
+                            // keeps failing.
+                            updateMinRetainOffset(replica, minRetainOffset);
                         }
                     });
         } else {