[flink-action][server][client] add orphan files cleanup action for remote storage

fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java

@@ -17,6 +17,7 @@
 
 package org.apache.fluss.client.admin;
 
+import org.apache.fluss.annotation.Internal;
 import org.apache.fluss.annotation.PublicEvolving;
 import org.apache.fluss.client.metadata.KvSnapshotMetadata;
 import org.apache.fluss.client.metadata.KvSnapshots;
@@ -67,6 +68,8 @@
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.metadata.TableStats;
+import org.apache.fluss.rpc.messages.ListKvSnapshotsResponse;
+import org.apache.fluss.rpc.messages.ListRemoteLogManifestsResponse;
 import org.apache.fluss.security.acl.AclBinding;
 import org.apache.fluss.security.acl.AclBindingFilter;
 
@@ -770,4 +773,32 @@ CompletableFuture<RegisterResult> registerProducerOffsets(
      * @since 0.9
      */
     CompletableFuture<Void> deleteProducerOffsets(String producerId);
+
+    /**
+     * List per-bucket remote log manifest entries for a table or partition scope.
+     *
+     * @param tableId the table to query
+     * @param partitionId optional partition id (null for non-partitioned tables)
+     * @return per-bucket manifest paths and end offsets
+     */
+    @Internal
+    default CompletableFuture<ListRemoteLogManifestsResponse> listRemoteLogManifests(
+            long tableId, @Nullable Long partitionId) {
+        throw new UnsupportedOperationException(
+                "listRemoteLogManifests is not supported by this Admin implementation");
+    }
+
+    /**
+     * List per-bucket active KV snapshot ids for a table or partition scope.
+     *
+     * @param tableId the table to query
+     * @param partitionId optional partition id (null for non-partitioned tables)
+     * @return per-bucket active snapshot entries
+     */
+    @Internal
+    default CompletableFuture<ListKvSnapshotsResponse> listKvSnapshots(
+            long tableId, @Nullable Long partitionId) {
+        throw new UnsupportedOperationException(
+                "listKvSnapshots is not supported by this Admin implementation");
+    }
 }

fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java

@@ -79,10 +79,14 @@
 import org.apache.fluss.rpc.messages.ListAclsRequest;
 import org.apache.fluss.rpc.messages.ListDatabasesRequest;
 import org.apache.fluss.rpc.messages.ListDatabasesResponse;
+import org.apache.fluss.rpc.messages.ListKvSnapshotsRequest;
+import org.apache.fluss.rpc.messages.ListKvSnapshotsResponse;
 import org.apache.fluss.rpc.messages.ListOffsetsRequest;
 import org.apache.fluss.rpc.messages.ListOffsetsResponse;
 import org.apache.fluss.rpc.messages.ListPartitionInfosRequest;
 import org.apache.fluss.rpc.messages.ListRebalanceProgressRequest;
+import org.apache.fluss.rpc.messages.ListRemoteLogManifestsRequest;
+import org.apache.fluss.rpc.messages.ListRemoteLogManifestsResponse;
 import org.apache.fluss.rpc.messages.ListTablesRequest;
 import org.apache.fluss.rpc.messages.ListTablesResponse;
 import org.apache.fluss.rpc.messages.PbAlterConfig;
@@ -367,6 +371,38 @@ public CompletableFuture<List<PartitionInfo>> listPartitionInfos(
                 .thenApply(ClientRpcMessageUtils::toPartitionInfos);
     }
 
+    /**
+     * Returns per-bucket remote log manifest path for the given table or partition.
+     *
+     * <p>Used by the orphan cleanup action to construct the active manifest path set without
+     * relying on FS LIST + mtime selection.
+     */
+    @Override
+    public CompletableFuture<ListRemoteLogManifestsResponse> listRemoteLogManifests(
+            long tableId, @Nullable Long partitionId) {
+        ListRemoteLogManifestsRequest request = new ListRemoteLogManifestsRequest();
+        request.setTableId(tableId);
+        if (partitionId != null) {
+            request.setPartitionId(partitionId);
+        }
+        return gateway.listRemoteLogManifests(request);
+    }
+
+    /**
+     * Returns per-bucket active KV snapshot dirs (retained_N + still-in-use) for the given table or
+     * partition. Used by the orphan cleanup action to construct the complete KV active set.
+     */
+    @Override
+    public CompletableFuture<ListKvSnapshotsResponse> listKvSnapshots(
+            long tableId, @Nullable Long partitionId) {
+        ListKvSnapshotsRequest request = new ListKvSnapshotsRequest();
+        request.setTableId(tableId);
+        if (partitionId != null) {
+            request.setPartitionId(partitionId);
+        }
+        return gateway.listKvSnapshots(request);
+    }
+
     @Override
     public CompletableFuture<Void> createPartition(
             TablePath tablePath, PartitionSpec partitionSpec, boolean ignoreIfExists) {

fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java

@@ -533,6 +533,17 @@ public static ReleaseKvSnapshotLeaseRequest makeReleaseKvSnapshotLeaseRequest(
         return request;
     }
 
+    public static PbTableBucket toPbTableBucket(TableBucket tableBucket) {
+        PbTableBucket pbTableBucket =
+                new PbTableBucket()
+                        .setTableId(tableBucket.getTableId())
+                        .setBucketId(tableBucket.getBucket());
+        if (tableBucket.getPartitionId() != null) {
+            pbTableBucket.setPartitionId(tableBucket.getPartitionId());
+        }
+        return pbTableBucket;
+    }
+
     public static Optional<RebalanceProgress> toRebalanceProgress(
             ListRebalanceProgressResponse response) {
         if (!response.hasRebalanceId()) {

fluss-common/src/main/java/org/apache/fluss/fs/FileStatus.java

@@ -46,4 +46,18 @@ public interface FileStatus {
      * @return the corresponding Path to the FileStatus
      */
     FsPath getPath();
+
+    /**
+     * Returns the modification time of the file in milliseconds since the epoch.
+     *
+     * <p>The default implementation returns {@link Long#MAX_VALUE}, which is interpreted by
+     * time-based filters (e.g. orphan-files cleanup) as "always fresh" - effectively a fail-closed
+     * default that prevents deletion when modification time is unavailable. File system
+     * implementations that can expose modification time SHOULD override this.
+     *
+     * @return the modification time in epoch millis, or {@link Long#MAX_VALUE} when unavailable
+     */
+    default long getModificationTime() {
+        return Long.MAX_VALUE;
+    }
 }

fluss-common/src/main/java/org/apache/fluss/fs/local/LocalFileStatus.java

@@ -67,6 +67,11 @@ public FsPath getPath() {
         return this.path;
     }
 
+    @Override
+    public long getModificationTime() {
+        return this.file.lastModified();
+    }
+
     public File getFile() {
         return this.file;
     }

fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java

@@ -74,7 +74,7 @@ public class FlussPaths {
     public static final String REMOTE_LOG_DIR_NAME = "log";
 
     /** The directory name for storing metadata files (e.g., manifest) for a log tablet. */
-    private static final String REMOTE_LOG_METADATA_DIR_NAME = "metadata";
+    public static final String REMOTE_LOG_METADATA_DIR_NAME = "metadata";
 
     /** Suffix of a manifest file. */
     private static final String REMOTE_LOG_MANIFEST_FILE_SUFFIX = ".manifest";

fluss-common/src/test/java/org/apache/fluss/fs/FileStatusTest.java

@@ -0,0 +1,54 @@
+/*
+ * 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.fs;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for default methods of {@link FileStatus}. */
+class FileStatusTest {
+
+    /**
+     * An implementation that does not override {@link FileStatus#getModificationTime()} must
+     * inherit the fail-safe default of {@link Long#MAX_VALUE}, so time-based filters treat the file
+     * as "always fresh" and never delete it when modification time is unavailable.
+     */
+    @Test
+    void defaultModificationTimeIsMaxValueFailSafe() {
+        FileStatus status =
+                new FileStatus() {
+                    @Override
+                    public long getLen() {
+                        return 0L;
+                    }
+
+                    @Override
+                    public boolean isDir() {
+                        return false;
+                    }
+
+                    @Override
+                    public FsPath getPath() {
+                        return new FsPath("/tmp/x");
+                    }
+                };
+
+        assertThat(status.getModificationTime()).isEqualTo(Long.MAX_VALUE);
+    }
+}

fluss-filesystems/fluss-fs-hadoop/src/main/java/org/apache/fluss/fs/hdfs/HadoopFileStatus.java

@@ -52,6 +52,11 @@ public boolean isDir() {
         return fileStatus.isDirectory();
     }
 
+    @Override
+    public long getModificationTime() {
+        return fileStatus.getModificationTime();
+    }
+
     // ------------------------------------------------------------------------
 
     /**

fluss-flink/fluss-flink-1.19/src/test/java/org/apache/fluss/flink/action/orphan/Flink19OrphanFilesCleanITCase.java

@@ -0,0 +1,22 @@
+/*
+ * 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.flink.action.orphan;
+
+/** The IT case for orphan files cleanup in Flink 1.19. */
+class Flink19OrphanFilesCleanITCase extends OrphanFilesCleanITCase {}

fluss-flink/fluss-flink-1.20/src/test/java/org/apache/fluss/flink/action/orphan/Flink20OrphanFilesCleanITCase.java

@@ -0,0 +1,22 @@
+/*
+ * 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.flink.action.orphan;
+
+/** The IT case for orphan files cleanup in Flink 1.20. */
+class Flink20OrphanFilesCleanITCase extends OrphanFilesCleanITCase {}

fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/MultipleParameterToolAdapter.java

@@ -19,6 +19,9 @@
 
 import org.apache.flink.util.MultipleParameterTool;
 
+import javax.annotation.Nullable;
+
+import java.util.Collection;
 import java.util.Map;
 
 /**
@@ -43,4 +46,23 @@ public static MultipleParameterToolAdapter fromArgs(String[] args) {
     public Map<String, String> toMap() {
         return this.multipleParameterTool.toMap();
     }
+
+    /** Returns whether the given key is present in the parsed arguments. */
+    public boolean has(String key) {
+        return this.multipleParameterTool.has(key);
+    }
+
+    /** Returns the value for the given key, or {@code null} if the key is not found. */
+    @Nullable
+    public String get(String key) {
+        return this.multipleParameterTool.get(key);
+    }
+
+    /**
+     * Returns all values associated with the given key, or {@code null} if the key is not found.
+     */
+    @Nullable
+    public Collection<String> getMultiParameter(String key) {
+        return this.multipleParameterTool.getMultiParameter(key);
+    }
 }

fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/action/orphan/Flink22OrphanFilesCleanITCase.java

@@ -0,0 +1,22 @@
+/*
+ * 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.flink.action.orphan;
+
+/** The IT case for orphan files cleanup in Flink 2.2. */
+class Flink22OrphanFilesCleanITCase extends OrphanFilesCleanITCase {}