[server][flink][test] polish orphan cleanup and broaden active snapshot set

Author: platinumhamburg

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/OrphanCleanUtils.java

@@ -20,6 +20,7 @@
 import org.apache.fluss.annotation.Internal;
 import org.apache.fluss.client.admin.Admin;
 import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
 import org.apache.fluss.config.cluster.ConfigEntry;
 import org.apache.fluss.fs.FileStatus;
 import org.apache.fluss.fs.FileSystem;
@@ -36,6 +37,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 
@@ -108,12 +110,65 @@ public static String resolveRemoteDataDir(
      */
     @Nullable
     public static String resolveClusterRemoteDataDir(Admin admin) throws Exception {
+        return resolveClusterRemoteDataDir(fetchClusterConfigMap(admin));
+    }
+
+    /** Extracts the single-root {@code remote.data.dir} from a pre-fetched config map. */
+    @Nullable
+    public static String resolveClusterRemoteDataDir(Map<String, String> configMap) {
+        return configMap.get(ConfigOptions.REMOTE_DATA_DIR.key());
+    }
+
+    /**
+     * Resolves all cluster-level remote data directories by querying the coordinator's runtime
+     * configuration. Reads both the single-root {@code remote.data.dir} and the multi-root {@code
+     * remote.data.dirs}, deduplicates by normalized form, and returns the union as the canonical
+     * root list.
+     *
+     * <p>This is the authoritative source for determining what storage roots the cleanup action is
+     * allowed to touch.
+     *
+     * @return list of normalized roots (no trailing slash); never {@code null}, may be empty if the
+     *     cluster has neither config set (which should not happen because the coordinator requires
+     *     at least one remote data dir at startup).
+     */
+    public static List<String> resolveClusterRemoteDataDirs(Admin admin) throws Exception {
+        return resolveClusterRemoteDataDirs(fetchClusterConfigMap(admin));
+    }
+
+    /** Extracts all remote data roots from a pre-fetched config map. */
+    public static List<String> resolveClusterRemoteDataDirs(Map<String, String> configMap) {
+        Configuration conf = Configuration.fromMap(configMap);
+        LinkedHashSet<String> roots = new LinkedHashSet<String>();
+        String singleDir = conf.get(ConfigOptions.REMOTE_DATA_DIR);
+        if (singleDir != null && !singleDir.isEmpty()) {
+            roots.add(normalizeRoot(singleDir));
+        }
+        List<String> multiDirs = conf.get(ConfigOptions.REMOTE_DATA_DIRS);
+        if (multiDirs != null) {
+            for (String dir : multiDirs) {
+                if (dir != null && !dir.isEmpty()) {
+                    roots.add(normalizeRoot(dir));
+                }
+            }
+        }
+        return new ArrayList<String>(roots);
+    }
+
+    /**
+     * Fetches the coordinator's runtime configuration as a key-value map. Use this once and pass
+     * the result to the map-based overloads of {@link #resolveClusterRemoteDataDir(Map)} and {@link
+     * #resolveClusterRemoteDataDirs(Map)} to avoid duplicate RPCs.
+     */
+    public static Map<String, String> fetchClusterConfigMap(Admin admin) throws Exception {
         Collection<ConfigEntry> entries = admin.describeClusterConfigs().get();
         Map<String, String> map = new HashMap<String, String>();
         for (ConfigEntry entry : entries) {
-            map.put(entry.key(), entry.value());
+            if (entry.value() != null) {
+                map.put(entry.key(), entry.value());
+            }
         }
-        return map.get(ConfigOptions.REMOTE_DATA_DIR.key());
+        return map;
     }
 
     /** Constructs a remote sub-directory path, normalizing trailing slashes on the root. */

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/OrphanFilesCleanActionFactory.java

@@ -44,21 +44,21 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
     public String help() {
         return "Usage: orphan_files_clean --bootstrap-server <host:port>\n"
                 + "  (--database <db> [--table <name>] | --all-databases)\n"
-                + "  [--scan-root <remoteDataDir>]...\n"
-                + "  [--older-than 'yyyy-MM-dd HH:mm:ss']\n"
+                + "  [--older-than '<ISO-8601 with offset>']\n"
                 + "  [--delete-rate-limit-per-second 100] [--dry-run]\n"
                 + "  [--allow-delete-manifest]\n"
                 + "  [--allow-clean-orphan-tables]\n"
                 + "  [--allow-clean-orphan-partitions]\n"
                 + "  [--conf <key>=<value>]...\n"
                 + "\n"
                 + "Notes:\n"
-                + "  --older-than is an absolute wall-clock cutoff (server local timezone). Files\n"
-                + "    with mtime strictly less than the cutoff are deletion-eligible. Default:\n"
-                + "    now - 3d, computed once at startup. The cutoff is frozen for the run, so a\n"
-                + "    long scan cannot accidentally pull in files written after the action started.\n"
-                + "    The cutoff must be at least 1d before now (closer cutoffs would race with\n"
-                + "    mid-write files).\n"
+                + "  --older-than is an absolute wall-clock cutoff in ISO-8601 with explicit\n"
+                + "    offset (e.g. '2024-01-01T00:00:00+08:00' or '2024-01-01T00:00:00Z').\n"
+                + "    Files with mtime strictly less than the cutoff are deletion-eligible.\n"
+                + "    Default: now - 3d, computed once at startup. The cutoff is frozen for the\n"
+                + "    run, so a long scan cannot accidentally pull in files written after the\n"
+                + "    action started. The cutoff must be at least 1d before now (closer cutoffs\n"
+                + "    would race with mid-write files).\n"
                 + "  Orphan directory detection (table/partition) relies solely on ID guards\n"
                 + "    (maxKnownTableId / maxKnownPartitionId), not mtime.\n"
                 + "  --table also disables the orphan-table scan (no sibling orphan-table scan in\n"

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/audit/AuditLogger.java

@@ -196,4 +196,15 @@ public void logSkipOrphanPartition(FsPath dir, String reason) {
         AUDIT.info(
                 "action=skip_orphan_partition reason={} path={} ts={}", reason, dir, Instant.now());
     }
+
+    /** Skip a bucket target because its metadata-resolved root is outside cluster config. */
+    public void logSkipBucketOutOfScope(long tableId, Long partitionId, String resolvedRoot) {
+        AUDIT.info(
+                "action=skip_bucket_target reason=out-of-scope-root table_id={} partition_id={}"
+                        + " resolved_root={} ts={}",
+                tableId,
+                partitionId,
+                resolvedRoot,
+                Instant.now());
+    }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/config/OrphanCleanConfig.java

@@ -26,15 +26,11 @@
 import java.io.Serializable;
 import java.time.Duration;
 import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneId;
-import java.time.format.DateTimeFormatter;
+import java.time.OffsetDateTime;
 import java.time.format.DateTimeParseException;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
@@ -57,14 +53,6 @@ public final class OrphanCleanConfig implements Serializable {
 
     private static final long DEFAULT_DELETE_RATE_LIMIT_PER_SECOND = 100L;
 
-    /**
-     * Wall-clock timestamp format accepted on the CLI ({@code yyyy-MM-dd HH:mm:ss}, interpreted in
-     * the server's local time zone). Matches Apache Paimon's {@code orphan_files_clean older_than}
-     * grammar to minimize operator context-switching between systems.
-     */
-    private static final DateTimeFormatter CUTOFF_FORMATTER =
-            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
-
     private final String bootstrapServer;
     private final boolean allDatabases;
     private final @Nullable String database;
@@ -73,7 +61,6 @@ public final class OrphanCleanConfig implements Serializable {
     private final boolean dryRun;
     private final long deleteRateLimitPerSecond;
     private final @Nullable Integer parallelism;
-    private final List<String> scanRoots;
     private final boolean allowDeleteManifest;
     private final boolean allowCleanOrphanTables;
     private final boolean allowCleanOrphanPartitions;
@@ -88,7 +75,6 @@ private OrphanCleanConfig(
             boolean dryRun,
             long deleteRateLimitPerSecond,
             @Nullable Integer parallelism,
-            List<String> scanRoots,
             boolean allowDeleteManifest,
             boolean allowCleanOrphanTables,
             boolean allowCleanOrphanPartitions,
@@ -101,7 +87,6 @@ private OrphanCleanConfig(
         this.dryRun = dryRun;
         this.deleteRateLimitPerSecond = deleteRateLimitPerSecond;
         this.parallelism = parallelism;
-        this.scanRoots = Collections.unmodifiableList(new ArrayList<String>(scanRoots));
         this.allowDeleteManifest = allowDeleteManifest;
         this.allowCleanOrphanTables = allowCleanOrphanTables;
         this.allowCleanOrphanPartitions = allowCleanOrphanPartitions;
@@ -149,7 +134,6 @@ public static OrphanCleanConfig fromParams(MultipleParameterToolAdapter params)
                 params.has("dry-run"),
                 deleteRateLimitPerSecond,
                 parallelism,
-                parseScanRoots(params.getMultiParameter("scan-root")),
                 allowDeleteManifest,
                 allowCleanOrphanTables,
                 allowCleanOrphanPartitions,
@@ -158,27 +142,28 @@ public static OrphanCleanConfig fromParams(MultipleParameterToolAdapter params)
 
     /**
      * Parses a CLI cutoff value into an absolute epoch-ms timestamp. Empty input falls back to
-     * {@code now - defaultGap}. Explicit input must parse as {@code yyyy-MM-dd HH:mm:ss} in the
-     * Flink action JVM's local time zone and must be at least {@link #HARD_LOWER_BOUND} earlier
-     * than {@code now} — closer-to-now cutoffs would race with active writes (see {@code
-     * HARD_LOWER_BOUND} javadoc).
+     * {@code now - defaultGap}. Explicit input must be ISO-8601 with an explicit offset (e.g.
+     * {@code 2024-01-01T00:00:00+08:00} or {@code 2024-01-01T00:00:00Z}) and must be at least
+     * {@link #HARD_LOWER_BOUND} earlier than {@code now} — closer-to-now cutoffs would race with
+     * active writes (see {@code HARD_LOWER_BOUND} javadoc).
      */
     private static long parseCutoff(
             String flag, @Nullable String value, long now, Duration defaultGap) {
         if (StringUtils.isNullOrWhitespaceOnly(value)) {
             return now - defaultGap.toMillis();
         }
-        LocalDateTime parsed;
+        OffsetDateTime parsed;
         try {
-            parsed = LocalDateTime.parse(value, CUTOFF_FORMATTER);
+            parsed = OffsetDateTime.parse(value);
         } catch (DateTimeParseException e) {
             throw new IllegalArgumentException(
                     flag
-                            + " must be a timestamp in 'yyyy-MM-dd HH:mm:ss' (server local TZ), got: "
+                            + " must be an ISO-8601 timestamp with an explicit offset (e.g."
+                            + " '2024-01-01T00:00:00+08:00' or '2024-01-01T00:00:00Z'); got: "
                             + value,
                     e);
         }
-        long parsedMillis = parsed.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+        long parsedMillis = parsed.toInstant().toEpochMilli();
         long maxAllowed = now - HARD_LOWER_BOUND.toMillis();
         if (parsedMillis > maxAllowed) {
             throw new IllegalArgumentException(
@@ -215,21 +200,6 @@ private static Integer parseParallelism(@Nullable String value) {
         return p;
     }
 
-    private static List<String> parseScanRoots(@Nullable Collection<String> values) {
-        if (values == null || values.isEmpty()) {
-            return Collections.emptyList();
-        }
-
-        List<String> scanRoots = new ArrayList<String>(values.size());
-        for (String value : values) {
-            if (StringUtils.isNullOrWhitespaceOnly(value)) {
-                throw new IllegalArgumentException("--scan-root must not be blank");
-            }
-            scanRoots.add(value);
-        }
-        return scanRoots;
-    }
-
     private static Map<String, String> parseExtraConfigs(@Nullable Collection<String> values) {
         if (values == null || values.isEmpty()) {
             return Collections.emptyMap();
@@ -291,11 +261,6 @@ public Optional<Integer> parallelism() {
         return Optional.ofNullable(parallelism);
     }
 
-    /** Returns additional remote.data.dir roots to scan. */
-    public List<String> scanRoots() {
-        return scanRoots;
-    }
-
     /**
      * Opt-in to delete {@code .manifest} files. Default {@code false}: mis-deleting an active
      * manifest leaves the coordinator's manifest pointer dangling and breaks the bucket's metadata

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/CleanStats.java

@@ -24,9 +24,10 @@
 import java.util.List;
 
 /**
- * Aggregatable cleanup statistics emitted by each {@link ScanAndCleanFunction} subtask. The {@code
- * touchedDirs} list is collected by the final aggregator for empty-directory sweeping after all
- * subtasks complete.
+ * Per-task cleanup statistics emitted by each {@link ScanAndCleanFunction} subtask. The scalar
+ * counters are accumulated by {@link StatsAggregateOperator} via simple addition; the short {@code
+ * touchedDirs} list (typically 1–2 entries per task) is inserted into a {@code HashSet} for O(1)
+ * deduplication — no list concatenation or O(n²) merge is needed.
  */
 @Internal
 public final class CleanStats implements Serializable {
@@ -49,11 +50,11 @@ public CleanStats(
         this.deleted = deleted;
         this.deleteFailures = deleteFailures;
         this.bytesReclaimed = bytesReclaimed;
-        this.touchedDirs = new ArrayList<>(touchedDirs);
+        this.touchedDirs = new ArrayList<String>(touchedDirs);
     }
 
     public static CleanStats empty() {
-        return new CleanStats(0L, 0L, 0L, 0L, new ArrayList<String>());
+        return new CleanStats(0L, 0L, 0L, 0L, new ArrayList<String>(0));
     }
 
     public long scanned() {
@@ -75,15 +76,4 @@ public long bytesReclaimed() {
     public List<String> touchedDirs() {
         return touchedDirs;
     }
-
-    public CleanStats merge(CleanStats other) {
-        List<String> mergedDirs = new ArrayList<>(this.touchedDirs);
-        mergedDirs.addAll(other.touchedDirs);
-        return new CleanStats(
-                this.scanned + other.scanned,
-                this.deleted + other.deleted,
-                this.deleteFailures + other.deleteFailures,
-                this.bytesReclaimed + other.bytesReclaimed,
-                mergedDirs);
-    }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/EmptyDirSweeper.java

@@ -54,10 +54,6 @@ public final class EmptyDirSweeper {
     private final RateLimiter rateLimiter;
     private final Set<FsPath> touchedRoots = new HashSet<FsPath>();
 
-    public EmptyDirSweeper(boolean dryRun, AuditLogger audit) {
-        this(dryRun, audit, RateLimiter.create(100.0));
-    }
-
     public EmptyDirSweeper(boolean dryRun, AuditLogger audit, RateLimiter rateLimiter) {
         this.dryRun = dryRun;
         this.audit = audit;

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/OrphanFilesCleanJob.java

@@ -87,7 +87,10 @@ public static CleanStats execute(
                 stats.transform(
                                 "StatsAggregate",
                                 TypeInformation.of(new TypeHint<CleanStats>() {}),
-                                new StatsAggregateOperator(config.dryRun(), config.extraConfigs()))
+                                new StatsAggregateOperator(
+                                        config.dryRun(),
+                                        config.extraConfigs(),
+                                        config.deleteRateLimitPerSecond()))
                         .setParallelism(1)
                         .setMaxParallelism(1);