address baiye's fip comments

swuferhong · swuferhong · commit 3249ea48e260 · 2025-12-16T20:15:20.000+08:00
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java
@@ -21,6 +21,7 @@
 import org.apache.fluss.client.metadata.KvSnapshotMetadata;
 import org.apache.fluss.client.metadata.KvSnapshots;
 import org.apache.fluss.client.metadata.LakeSnapshot;
+import org.apache.fluss.client.metadata.RegisterKvSnapshotResult;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.cluster.AlterConfig;
@@ -419,8 +420,9 @@ CompletableFuture<KvSnapshotMetadata> getKvSnapshotMetadata(
      *
      * @param consumerId the consumer id.
      * @param consumeBuckets the tableBuckets to consume, a map from TableBucket to kvSnapshotId.
+     * @return the result of registering kv snapshot consumer.
      */
-    CompletableFuture<Void> registerKvSnapshotConsumer(
+    CompletableFuture<RegisterKvSnapshotResult> registerKvSnapshotConsumer(
             String consumerId, Map<TableBucket, Long> consumeBuckets);
 
     /**
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java
@@ -21,6 +21,7 @@
 import org.apache.fluss.client.metadata.KvSnapshots;
 import org.apache.fluss.client.metadata.LakeSnapshot;
 import org.apache.fluss.client.metadata.MetadataUpdater;
+import org.apache.fluss.client.metadata.RegisterKvSnapshotResult;
 import org.apache.fluss.client.utils.ClientRpcMessageUtils;
 import org.apache.fluss.cluster.Cluster;
 import org.apache.fluss.cluster.ServerNode;
@@ -387,7 +388,7 @@ public CompletableFuture<KvSnapshotMetadata> getKvSnapshotMetadata(
     }
 
     @Override
-    public CompletableFuture<Void> registerKvSnapshotConsumer(
+    public CompletableFuture<RegisterKvSnapshotResult> registerKvSnapshotConsumer(
             String consumerId, Map<TableBucket, Long> consumeBuckets) {
         if (consumeBuckets.isEmpty()) {
             throw new IllegalArgumentException("consumeBuckets is empty");
@@ -400,7 +401,7 @@ public CompletableFuture<Void> registerKvSnapshotConsumer(
         return gateway.registerKvSnapshotConsumer(
                         makeRegisterKvSnapshotConsumerRequest(
                                 consumerId, consumeBuckets, expirationTime))
-                .thenApply(r -> null);
+                .thenApply(ClientRpcMessageUtils::toRegisterKvSnapshotResult);
     }
 
     @Override
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/metadata/RegisterKvSnapshotResult.java b/fluss-client/src/main/java/org/apache/fluss/client/metadata/RegisterKvSnapshotResult.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.fluss.client.metadata;
+
+import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.metadata.TableBucket;
+
+import java.util.Set;
+
+/**
+ * A class to represent the result of registering kv snapshot. It contains:
+ *
+ * <ul>
+ *   <li>An set of failed tableBuckets. Such as the specify snapshotId is not exist for this table
+ *       bucket.
+ * </ul>
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public class RegisterKvSnapshotResult {
+    private final Set<TableBucket> failedTableBucketSet;
+
+    public RegisterKvSnapshotResult(Set<TableBucket> failedTableBucketSet) {
+        this.failedTableBucketSet = failedTableBucketSet;
+    }
+
+    public Set<TableBucket> getFailedTableBucketSet() {
+        return failedTableBucketSet;
+    }
+}
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java
@@ -23,6 +23,7 @@
 import org.apache.fluss.client.metadata.KvSnapshotMetadata;
 import org.apache.fluss.client.metadata.KvSnapshots;
 import org.apache.fluss.client.metadata.LakeSnapshot;
+import org.apache.fluss.client.metadata.RegisterKvSnapshotResult;
 import org.apache.fluss.client.write.KvWriteBatch;
 import org.apache.fluss.client.write.ReadyWriteBatch;
 import org.apache.fluss.config.cluster.AlterConfigOpType;
@@ -65,10 +66,12 @@
 import org.apache.fluss.rpc.messages.PbPutKvReqForBucket;
 import org.apache.fluss.rpc.messages.PbRemotePathAndLocalFile;
 import org.apache.fluss.rpc.messages.PbRenameColumn;
+import org.apache.fluss.rpc.messages.PbTable;
 import org.apache.fluss.rpc.messages.PrefixLookupRequest;
 import org.apache.fluss.rpc.messages.ProduceLogRequest;
 import org.apache.fluss.rpc.messages.PutKvRequest;
 import org.apache.fluss.rpc.messages.RegisterKvSnapshotConsumerRequest;
+import org.apache.fluss.rpc.messages.RegisterKvSnapshotConsumerResponse;
 import org.apache.fluss.rpc.messages.UnregisterKvSnapshotConsumerRequest;
 import org.apache.fluss.utils.json.DataTypeJsonSerde;
 import org.apache.fluss.utils.json.JsonSerdeUtils;
@@ -79,6 +82,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -408,6 +412,23 @@ public static RegisterKvSnapshotConsumerRequest makeRegisterKvSnapshotConsumerRe
         return request;
     }
 
+    public static RegisterKvSnapshotResult toRegisterKvSnapshotResult(
+            RegisterKvSnapshotConsumerResponse response) {
+        Set<TableBucket> failedTableBucketSet = new HashSet<>();
+        for (PbTable failedTable : response.getFailedTablesList()) {
+            long tableId = failedTable.getTableId();
+            for (PbBucket pbBucket : failedTable.getBucketsList()) {
+                TableBucket tableBucket =
+                        new TableBucket(
+                                tableId,
+                                pbBucket.hasPartitionId() ? pbBucket.getPartitionId() : null,
+                                pbBucket.getBucketId());
+                failedTableBucketSet.add(tableBucket);
+            }
+        }
+        return new RegisterKvSnapshotResult(failedTableBucketSet);
+    }
+
     public static UnregisterKvSnapshotConsumerRequest makeUnregisterKvSnapshotConsumerRequest(
             String consumerId, Set<TableBucket> bucketsToUnregister) {
         UnregisterKvSnapshotConsumerRequest request = new UnregisterKvSnapshotConsumerRequest();
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -1462,6 +1462,12 @@ public class ConfigOptions {
                     .withDescription(
                             "The number of threads the server uses to transfer (download and upload) kv snapshot files.");
 
+    public static final ConfigOption<Integer> KV_MAX_RETAINED_SNAPSHOTS =
+            key("kv.snapshot.num-retained")
+                    .intType()
+                    .defaultValue(1)
+                    .withDescription("The maximum number of completed snapshots to retain.");
+
     public static final ConfigOption<Duration> KV_SNAPSHOT_CONSUMER_EXPIRATION_CHECK_INTERVAL =
             key("kv.snapshot.consumer-expiration-check-interval")
                     .durationType()
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/metrics/FlinkMetricsITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/metrics/FlinkMetricsITCase.java
@@ -68,7 +68,11 @@ abstract class FlinkMetricsITCase {
                     .setClusterConf(
                             new org.apache.fluss.config.Configuration()
                                     // set snapshot interval to 1s for testing purposes
-                                    .set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1)))
+                                    .set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1))
+                                    // not to clean snapshots for test purpose
+                                    .set(
+                                            ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS,
+                                            Integer.MAX_VALUE))
                     .setNumOfTabletServers(3)
                     .build();
 
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceFailOverITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceFailOverITCase.java
@@ -75,7 +75,11 @@ abstract class FlinkTableSourceFailOverITCase {
                     .setClusterConf(
                             new org.apache.fluss.config.Configuration()
                                     // set snapshot interval to 1s for testing purposes
-                                    .set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1)))
+                                    .set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1))
+                                    // not to clean snapshots for test purpose
+                                    .set(
+                                            ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS,
+                                            Integer.MAX_VALUE))
                     .setNumOfTabletServers(3)
                     .build();
 
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceITCase.java
@@ -94,7 +94,11 @@ abstract class FlinkTableSourceITCase extends AbstractTestBase {
                     .setClusterConf(
                             new Configuration()
                                     // set snapshot interval to 1s for testing purposes
-                                    .set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1)))
+                                    .set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1))
+                                    // not to clean snapshots for test purpose
+                                    .set(
+                                            ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS,
+                                            Integer.MAX_VALUE))
                     .setNumOfTabletServers(3)
                     .setClock(CLOCK)
                     .build();
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringTestBase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringTestBase.java
@@ -181,6 +181,8 @@ private static Configuration flussClusterConfig() {
         Configuration conf = new Configuration();
         // set snapshot interval to 1s for testing purposes
         conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1));
+        // not to clean snapshots for test purpose
+        conf.set(ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS, Integer.MAX_VALUE);
 
         // enable lake tiering
         conf.set(ConfigOptions.DATALAKE_FORMAT, DataLakeFormat.PAIMON);
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkTestBase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkTestBase.java
@@ -71,7 +71,11 @@ public class FlinkTestBase extends AbstractTestBase {
                     .setClusterConf(
                             new Configuration()
                                     // set snapshot interval to 1s for testing purposes
-                                    .set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1)))
+                                    .set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1))
+                                    // not to clean snapshots for test purpose
+                                    .set(
+                                            ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS,
+                                            Integer.MAX_VALUE))
                     .setNumOfTabletServers(3)
                     .build();
 
diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java
@@ -105,7 +105,8 @@ public class FlinkIcebergTieringTestBase {
 
     private static Configuration initConfig() {
         Configuration conf = new Configuration();
-        conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1));
+        conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1))
+                .set(ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS, Integer.MAX_VALUE);
 
         // Configure the tiering sink to be Iceberg
         conf.set(ConfigOptions.DATALAKE_FORMAT, DataLakeFormat.ICEBERG);
diff --git a/fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/testutils/FlinkLanceTieringTestBase.java b/fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/testutils/FlinkLanceTieringTestBase.java
@@ -78,7 +78,9 @@ public class FlinkLanceTieringTestBase {
 
     private static Configuration initConfig() {
         Configuration conf = new Configuration();
-        conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1));
+        conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1))
+                // not to clean snapshots for test purpose
+                .set(ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS, Integer.MAX_VALUE);
         conf.setString("datalake.format", "lance");
         try {
             warehousePath =
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/FlinkPaimonTieringTestBase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/FlinkPaimonTieringTestBase.java
@@ -88,7 +88,9 @@ public abstract class FlinkPaimonTieringTestBase {
 
     protected static Configuration initConfig() {
         Configuration conf = new Configuration();
-        conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1));
+        conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1))
+                // not to clean snapshots for test purpose
+                .set(ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS, Integer.MAX_VALUE);
         conf.setString("datalake.format", "paimon");
         conf.setString("datalake.paimon.metastore", "filesystem");
         try {
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -370,7 +370,9 @@ message RegisterKvSnapshotConsumerRequest {
   repeated PbConsumeKvSnapshotForTable register_tables = 3;
 }
 
-message RegisterKvSnapshotConsumerResponse {}
+message RegisterKvSnapshotConsumerResponse {
+  repeated PbTable failed_tables = 1;
+}
 
 message UnregisterKvSnapshotConsumerRequest {
   required string consumer_id = 1;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManager.java
@@ -46,6 +46,7 @@
 import java.util.concurrent.Executor;
 import java.util.function.Function;
 
+import static org.apache.fluss.utils.Preconditions.checkArgument;
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
 
 /**
@@ -57,6 +58,7 @@
 public class CompletedSnapshotStoreManager {
 
     private static final Logger LOG = LoggerFactory.getLogger(CompletedSnapshotStoreManager.class);
+    private final int maxNumberOfSnapshotsToRetain;
     private final ZooKeeperClient zooKeeperClient;
     private final ConcurrentHashMap<TableBucket, CompletedSnapshotStore>
             bucketCompletedSnapshotStores;
@@ -67,11 +69,13 @@ public class CompletedSnapshotStoreManager {
     private final CoordinatorMetricGroup coordinatorMetricGroup;
 
     public CompletedSnapshotStoreManager(
+            int maxNumberOfSnapshotsToRetain,
             Executor ioExecutor,
             ZooKeeperClient zooKeeperClient,
             CoordinatorMetricGroup coordinatorMetricGroup,
             CanSubsume canSubsume) {
         this(
+                maxNumberOfSnapshotsToRetain,
                 ioExecutor,
                 zooKeeperClient,
                 ZooKeeperCompletedSnapshotHandleStore::new,
@@ -81,12 +85,16 @@ public CompletedSnapshotStoreManager(
 
     @VisibleForTesting
     CompletedSnapshotStoreManager(
+            int maxNumberOfSnapshotsToRetain,
             Executor ioExecutor,
             ZooKeeperClient zooKeeperClient,
             Function<ZooKeeperClient, CompletedSnapshotHandleStore>
                     makeZookeeperCompletedSnapshotHandleStore,
             CoordinatorMetricGroup coordinatorMetricGroup,
             CanSubsume canSubsume) {
+        checkArgument(
+                maxNumberOfSnapshotsToRetain > 0, "maxNumberOfSnapshotsToRetain must be positive");
+        this.maxNumberOfSnapshotsToRetain = maxNumberOfSnapshotsToRetain;
         this.zooKeeperClient = zooKeeperClient;
         this.bucketCompletedSnapshotStores = MapUtils.newConcurrentHashMap();
         this.ioExecutor = ioExecutor;
@@ -231,6 +239,7 @@ private CompletedSnapshotStore createCompletedSnapshotStore(
         }
 
         return new CompletedSnapshotStore(
+                maxNumberOfSnapshotsToRetain,
                 sharedKvFileRegistry,
                 retrievedSnapshots,
                 completedSnapshotHandleStore,
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -44,6 +44,7 @@
 import org.apache.fluss.rpc.messages.CommitLakeTableSnapshotResponse;
 import org.apache.fluss.rpc.messages.CommitRemoteLogManifestResponse;
 import org.apache.fluss.rpc.messages.ControlledShutdownResponse;
+import org.apache.fluss.rpc.messages.PbBucket;
 import org.apache.fluss.rpc.messages.PbCommitLakeTableSnapshotRespForTable;
 import org.apache.fluss.rpc.messages.RegisterKvSnapshotConsumerResponse;
 import org.apache.fluss.rpc.messages.UnregisterKvSnapshotConsumerResponse;
@@ -210,6 +211,7 @@ public CoordinatorEventProcessor(
 
         this.completedSnapshotStoreManager =
                 new CompletedSnapshotStoreManager(
+                        conf.getInt(ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS),
                         ioExecutor,
                         zooKeeperClient,
                         coordinatorMetricGroup,
@@ -1346,10 +1348,24 @@ private ControlledShutdownResponse tryProcessControlledShutdown(
     private RegisterKvSnapshotConsumerResponse tryProcessRegisterKvSnapshotConsumer(
             RegisterKvSnapshotConsumerEvent event) throws Exception {
         RegisterKvSnapshotConsumerResponse response = new RegisterKvSnapshotConsumerResponse();
-        kvSnapshotConsumerManager.register(
-                event.getConsumerId(),
-                event.getExpirationTime(),
-                event.getTableIdToRegisterBucket());
+        Set<TableBucket> failedTableBuckets =
+                kvSnapshotConsumerManager.register(
+                        event.getConsumerId(),
+                        event.getExpirationTime(),
+                        event.getTableIdToRegisterBucket());
+
+        Map<Long, List<PbBucket>> pbFailedTables = new HashMap<>();
+        for (TableBucket tb : failedTableBuckets) {
+            PbBucket pbBucket = new PbBucket().setBucketId(tb.getBucket());
+            if (tb.getPartitionId() != null) {
+                pbBucket.setPartitionId(tb.getPartitionId());
+            }
+            pbFailedTables.computeIfAbsent(tb.getTableId(), k -> new ArrayList<>()).add(pbBucket);
+        }
+
+        for (Map.Entry<Long, List<PbBucket>> entry : pbFailedTables.entrySet()) {
+            response.addFailedTable().setTableId(entry.getKey()).addAllBuckets(entry.getValue());
+        }
         return response;
     }
 
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
@@ -521,6 +521,12 @@ private static void validateConfigs(Configuration conf) {
                             "Invalid configuration for %s, it must be greater than or equal 1.",
                             ConfigOptions.DEFAULT_REPLICATION_FACTOR.key()));
         }
+        if (conf.get(ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS) < 1) {
+            throw new IllegalConfigurationException(
+                    String.format(
+                            "Invalid configuration for %s, it must be greater than or equal 1.",
+                            ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS.key()));
+        }
 
         if (conf.get(ConfigOptions.COORDINATOR_IO_POOL_SIZE) < 1) {
             throw new IllegalConfigurationException(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/KvSnapshotConsumerManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/KvSnapshotConsumerManager.java
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStore.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStore.java
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CompletedSnapshotStoreManagerTest.java
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/KvSnapshotConsumerManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/KvSnapshotConsumerManagerTest.java
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshotStoreTest.java
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTargetTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvTabletSnapshotTargetTest.java
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotStoreTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/ZooKeeperCompletedSnapshotStoreTest.java
diff --git a/website/docs/maintenance/configuration.md b/website/docs/maintenance/configuration.md
