[flink-action][common][server] Support KV shared SST orphan file cleanup

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java

@@ -57,6 +57,8 @@
 import org.apache.fluss.exception.TooManyPartitionsException;
 import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.fs.FsPathAndFileName;
+import org.apache.fluss.kv.snapshot.CompletedSnapshot;
+import org.apache.fluss.kv.snapshot.KvSnapshotHandle;
 import org.apache.fluss.metadata.AggFunctions;
 import org.apache.fluss.metadata.DatabaseChange;
 import org.apache.fluss.metadata.DatabaseDescriptor;
@@ -77,8 +79,6 @@
 import org.apache.fluss.rpc.messages.ListOffsetsRequest;
 import org.apache.fluss.rpc.messages.ListOffsetsResponse;
 import org.apache.fluss.server.coordinator.CoordinatorContext;
-import org.apache.fluss.server.kv.snapshot.CompletedSnapshot;
-import org.apache.fluss.server.kv.snapshot.KvSnapshotHandle;
 import org.apache.fluss.server.log.LogTablet;
 import org.apache.fluss.server.metadata.ServerInfo;
 import org.apache.fluss.server.replica.Replica;

fluss-server/src/main/java/org/apache/fluss/server/kv/autoinc/AutoIncIDRange.java → fluss-common/src/main/java/org/apache/fluss/kv/autoinc/AutoIncIDRange.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.fluss.server.kv.autoinc;
+package org.apache.fluss.kv.autoinc;
 
 import java.util.Objects;
 

fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/CompletedSnapshot.java → fluss-common/src/main/java/org/apache/fluss/kv/snapshot/CompletedSnapshot.java

@@ -15,13 +15,13 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.server.kv.snapshot;
+package org.apache.fluss.kv.snapshot;
 
 import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.fs.FileSystem;
 import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.kv.autoinc.AutoIncIDRange;
 import org.apache.fluss.metadata.TableBucket;
-import org.apache.fluss.server.kv.autoinc.AutoIncIDRange;
 import org.apache.fluss.utils.concurrent.FutureUtils;
 
 import javax.annotation.Nullable;
@@ -106,7 +106,7 @@ public CompletedSnapshot(
     }
 
     @VisibleForTesting
-    CompletedSnapshot(
+    public CompletedSnapshot(
             TableBucket tableBucket,
             long snapshotID,
             FsPath snapshotLocation,
@@ -152,22 +152,11 @@ public long getSnapshotSize() {
         return kvSnapshotHandle.getSnapshotSize();
     }
 
-    /**
-     * Register all shared kv files in the given registry. This method is called before the snapshot
-     * is added into the store.
-     *
-     * @param sharedKvFileRegistry The registry where shared kv files are registered
-     */
-    public void registerSharedKvFilesAfterRestored(SharedKvFileRegistry sharedKvFileRegistry) {
-        sharedKvFileRegistry.registerAllAfterRestored(this);
-    }
-
     public CompletableFuture<Void> discardAsync(Executor ioExecutor) {
-        // it'll discard the snapshot files for kv, it'll always discard
-        // the private files; for shared files, only if they're not be registered in
-        // SharedKvRegistry, can the files be deleted.
+        // Discard private files only; shared files are managed by the SharedKvFileRegistry
+        // on the server side.
         CompletableFuture<Void> discardKvFuture =
-                FutureUtils.runAsync(kvSnapshotHandle::discard, ioExecutor);
+                FutureUtils.runAsync(kvSnapshotHandle::discardPrivateFiles, ioExecutor);
 
         CompletableFuture<Void> discardMetaFileFuture =
                 FutureUtils.runAsync(this::disposeMetadata, ioExecutor);

fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvFileHandle.java → fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvFileHandle.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.server.kv.snapshot;
+package org.apache.fluss.kv.snapshot;
 
 import org.apache.fluss.fs.FileSystem;
 import org.apache.fluss.fs.FsPath;

fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvFileHandleAndLocalPath.java → fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvFileHandleAndLocalPath.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.server.kv.snapshot;
+package org.apache.fluss.kv.snapshot;
 
 import java.util.Objects;
 

fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/KvSnapshotHandle.java → fluss-common/src/main/java/org/apache/fluss/kv/snapshot/KvSnapshotHandle.java

@@ -15,9 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.server.kv.snapshot;
-
-import org.apache.fluss.server.utils.SnapshotUtil;
+package org.apache.fluss.kv.snapshot;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -26,9 +24,6 @@
 import java.util.Objects;
 import java.util.stream.Collectors;
 
-import static org.apache.fluss.utils.Preconditions.checkNotNull;
-import static org.apache.fluss.utils.Preconditions.checkState;
-
 /**
  * A handle to the snapshot of a kv tablet. It contains the share file handles and the private file
  * handles from which we can rebuild the kv tablet.
@@ -45,16 +40,6 @@ public class KvSnapshotHandle {
     /** The size of the incremental snapshot. */
     private final long incrementalSize;
 
-    /**
-     * Once the shared states are registered, it is the {@link SharedKvFileRegistry}'s
-     * responsibility to cleanup those shared states. But in the cases where the state handle is
-     * discarded before performing the registration, the handle should delete all the shared states
-     * created by it.
-     *
-     * <p>This variable is not null iff the handles was registered.
-     */
-    private transient SharedKvFileRegistry sharedKvFileRegistry;
-
     public KvSnapshotHandle(
             List<KvFileHandleAndLocalPath> sharedFileHandles,
             List<KvFileHandleAndLocalPath> privateFileHandles,
@@ -97,43 +82,35 @@ public long getSnapshotSize() {
         return snapshotSize;
     }
 
-    public void discard() {
-        SharedKvFileRegistry registry = this.sharedKvFileRegistry;
-        final boolean isRegistered = (registry != null);
-
+    /**
+     * Discards only the private files of this snapshot. Use this when shared files are managed
+     * externally by a registry.
+     */
+    public void discardPrivateFiles() {
         try {
             SnapshotUtil.bestEffortDiscardAllKvFiles(
                     privateFileHandles.stream()
                             .map(KvFileHandleAndLocalPath::getKvFileHandle)
                             .collect(Collectors.toList()));
         } catch (Exception e) {
-            LOG.warn("Could not properly discard misc file states.", e);
-        }
-
-        if (!isRegistered) {
-            try {
-                SnapshotUtil.bestEffortDiscardAllKvFiles(
-                        sharedFileHandles.stream()
-                                .map(KvFileHandleAndLocalPath::getKvFileHandle)
-                                .collect(Collectors.toSet()));
-            } catch (Exception e) {
-                LOG.warn("Could not properly discard new sst file states.", e);
-            }
+            LOG.warn("Could not properly discard private file states.", e);
         }
     }
 
-    public void registerKvFileHandles(SharedKvFileRegistry registry, long snapshotID) {
-        checkState(
-                sharedKvFileRegistry != registry,
-                "The kv file handle has already registered its shared kv files to the given registry.");
-
-        sharedKvFileRegistry = checkNotNull(registry);
+    /**
+     * Discards all files (private + shared). Use this for failed or aborted snapshots that were
+     * never registered with a shared file registry.
+     */
+    public void discardAll() {
+        discardPrivateFiles();
 
-        for (KvFileHandleAndLocalPath handleAndLocalPath : sharedFileHandles) {
-            registry.registerReference(
-                    SharedKvFileRegistryKey.fromKvFileHandle(handleAndLocalPath.getKvFileHandle()),
-                    handleAndLocalPath.getKvFileHandle(),
-                    snapshotID);
+        try {
+            SnapshotUtil.bestEffortDiscardAllKvFiles(
+                    sharedFileHandles.stream()
+                            .map(KvFileHandleAndLocalPath::getKvFileHandle)
+                            .collect(Collectors.toSet()));
+        } catch (Exception e) {
+            LOG.warn("Could not properly discard shared sst file states.", e);
         }
     }
 

fluss-server/src/main/java/org/apache/fluss/server/utils/SnapshotUtil.java → fluss-common/src/main/java/org/apache/fluss/kv/snapshot/SnapshotUtil.java

@@ -15,9 +15,8 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.server.utils;
+package org.apache.fluss.kv.snapshot;
 
-import org.apache.fluss.server.kv.snapshot.KvFileHandle;
 import org.apache.fluss.utils.ExceptionUtils;
 import org.apache.fluss.utils.function.ThrowingConsumer;
 
@@ -44,6 +43,7 @@ public static void bestEffortDiscardAllKvFiles(
         applyToAllWhileSuppressingExceptions(handlesToDiscard, KvFileHandle::discard);
     }
 
+    /** Quietly discards a single kv file handle, logging any exception. */
     public static void discardKvFileQuietly(KvFileHandle kvFileHandle) {
         if (kvFileHandle == null) {
             return;