luoyuxia commented on code in PR #3404: URL: https://github.com/apache/fluss/pull/3404#discussion_r3331611730
########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/config/OrphanCleanConfig.java: ########## @@ -0,0 +1,335 @@ +/* + * 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.config; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.flink.adapter.MultipleParameterToolAdapter; +import org.apache.fluss.utils.StringUtils; + +import javax.annotation.Nullable; + +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.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; + +/** Parsed command-line options for the orphan files cleanup action. */ +@Internal +public final class OrphanCleanConfig implements Serializable { + + private static final long serialVersionUID = 1L; + + /** + * Minimum gap between any user-supplied cutoff and {@code now}. A cutoff closer to {@code now} + * would risk classifying files that are mid-write (committed file written, snapshot/manifest + * not yet visible to {@code ListRemoteLogManifests} / {@code ListKvSnapshots}) as orphan and + * deleting them. + */ + private static final Duration HARD_LOWER_BOUND = Duration.ofDays(1); + + /** Default file-level cutoff: files written before {@code now - 3d} are deletion-eligible. */ + private static final Duration DEFAULT_OLDER_THAN = Duration.ofDays(3); + + 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; + private final @Nullable String table; + private final long olderThanMillis; + 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; + private final Map<String, String> extraConfigs; + + private OrphanCleanConfig( + String bootstrapServer, + boolean allDatabases, + @Nullable String database, + @Nullable String table, + long olderThanMillis, + boolean dryRun, + long deleteRateLimitPerSecond, + @Nullable Integer parallelism, + List<String> scanRoots, + boolean allowDeleteManifest, + boolean allowCleanOrphanTables, + boolean allowCleanOrphanPartitions, + Map<String, String> extraConfigs) { + this.bootstrapServer = bootstrapServer; + this.allDatabases = allDatabases; + this.database = database; + this.table = table; + this.olderThanMillis = olderThanMillis; + 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; + this.extraConfigs = Collections.unmodifiableMap(new HashMap<>(extraConfigs)); + } + + /** Parses a cleanup config from CLI parameters. */ + public static OrphanCleanConfig fromParams(MultipleParameterToolAdapter params) { + String bootstrapServer = params.get("bootstrap-server"); + if (StringUtils.isNullOrWhitespaceOnly(bootstrapServer)) { + throw new IllegalArgumentException("--bootstrap-server is required"); + } + + boolean allDatabases = params.has("all-databases"); + String database = params.get("database"); + if (allDatabases && !StringUtils.isNullOrWhitespaceOnly(database)) { + throw new IllegalArgumentException( + "--database and --all-databases are mutually exclusive"); + } + if (!allDatabases && StringUtils.isNullOrWhitespaceOnly(database)) { + throw new IllegalArgumentException( + "Either --database or --all-databases must be provided"); + } + if (allDatabases && !StringUtils.isNullOrWhitespaceOnly(params.get("table"))) { + throw new IllegalArgumentException( + "--table requires --database and cannot be used with --all-databases"); + } + + long now = System.currentTimeMillis(); + long olderThanMillis = + parseCutoff("--older-than", params.get("older-than"), now, DEFAULT_OLDER_THAN); + long deleteRateLimitPerSecond = + parseDeleteRateLimit(params.get("delete-rate-limit-per-second")); + Integer parallelism = parseParallelism(params.get("parallelism")); + boolean allowDeleteManifest = params.has("allow-delete-manifest"); + boolean allowCleanOrphanTables = params.has("allow-clean-orphan-tables"); + boolean allowCleanOrphanPartitions = params.has("allow-clean-orphan-partitions"); + + return new OrphanCleanConfig( + bootstrapServer, + allDatabases, + database, + params.get("table"), + olderThanMillis, + params.has("dry-run"), + deleteRateLimitPerSecond, + parallelism, + parseScanRoots(params.getMultiParameter("scan-root")), + allowDeleteManifest, + allowCleanOrphanTables, + allowCleanOrphanPartitions, + parseExtraConfigs(params.getMultiParameter("conf"))); + } + + /** + * 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 + * server's local time zone and must be at least {@link #HARD_LOWER_BOUND} earlier than {@code Review Comment: nit: server make me feel like it's fluss server timezone, may clarify this as `Flink action JVM local timezone` ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/OrphanFilesCleanJob.java: ########## @@ -0,0 +1,109 @@ +/* + * 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.job; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.flink.action.orphan.config.OrphanCleanConfig; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.api.common.typeinfo.TypeHint; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; + +import java.util.Collections; +import java.util.Iterator; +import java.util.List; + +/** + * Builds and executes the 3-stage Flink Batch DAG for orphan files cleanup. + * + * <pre> + * Stage 1: ScopeEnumerator (p=1) — coordinator RPCs, emits CleanTask + * Stage 2: ScanAndClean (p=N) — FS scan + rate-limited delete, emits CleanStats + * Stage 3: StatsAggregate (p=1) — merge stats + empty-dir sweep, emits final CleanStats + * </pre> + */ +@Internal +public final class OrphanFilesCleanJob { + + private OrphanFilesCleanJob() {} + + /** + * Builds the DAG, executes it in batch mode, and returns the final aggregated cleanup + * statistics. + * + * @param env the Flink execution environment (caller configures classpath, etc.) + * @param config parsed orphan cleanup configuration + * @param parallelism the parallelism for Stage 2 (ScanAndClean); null uses env default + * @return the final cleanup statistics + */ + public static CleanStats execute( + StreamExecutionEnvironment env, OrphanCleanConfig config, Integer parallelism) + throws Exception { + env.setRuntimeMode(RuntimeExecutionMode.BATCH); + + // Stage 1: ScopeEnumerator (parallelism=1) + DataStream<Integer> trigger = + env.fromCollection(Collections.singletonList(1), TypeInformation.of(Integer.class)); + + SingleOutputStreamOperator<CleanTask> tasks = + trigger.process(new ScopeEnumeratorFunction(config)) + .returns(TypeInformation.of(new TypeHint<CleanTask>() {})) + .setParallelism(1) + .name("ScopeEnumerator"); + + // Stage 2: ScanAndClean (parallelism=N) + SingleOutputStreamOperator<CleanStats> stats = + tasks.rebalance() + .process( + new ScanAndCleanFunction( + config.deleteRateLimitPerSecond(), config.extraConfigs())) + .returns(TypeInformation.of(new TypeHint<CleanStats>() {})) + .name("ScanAndClean"); + if (parallelism != null) { + stats = stats.setParallelism(parallelism); + } + + // Stage 3: StatsAggregate + EmptyDirSweep (parallelism=1) + SingleOutputStreamOperator<CleanStats> result = + stats.transform( + "StatsAggregate", + TypeInformation.of(new TypeHint<CleanStats>() {}), + new StatsAggregateOperator(config.dryRun())) Review Comment: Should pass `config.extraConfigs()` to `StatsAggregateOperator` since seems it will also delete empty dir which may require may ak/sk ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/OrphanFilesCleanJob.java: ########## @@ -0,0 +1,109 @@ +/* + * 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.job; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.flink.action.orphan.config.OrphanCleanConfig; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.api.common.typeinfo.TypeHint; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; + +import java.util.Collections; +import java.util.Iterator; +import java.util.List; + +/** + * Builds and executes the 3-stage Flink Batch DAG for orphan files cleanup. + * + * <pre> + * Stage 1: ScopeEnumerator (p=1) — coordinator RPCs, emits CleanTask + * Stage 2: ScanAndClean (p=N) — FS scan + rate-limited delete, emits CleanStats + * Stage 3: StatsAggregate (p=1) — merge stats + empty-dir sweep, emits final CleanStats + * </pre> + */ +@Internal +public final class OrphanFilesCleanJob { + + private OrphanFilesCleanJob() {} + + /** + * Builds the DAG, executes it in batch mode, and returns the final aggregated cleanup + * statistics. + * + * @param env the Flink execution environment (caller configures classpath, etc.) + * @param config parsed orphan cleanup configuration + * @param parallelism the parallelism for Stage 2 (ScanAndClean); null uses env default + * @return the final cleanup statistics + */ + public static CleanStats execute( + StreamExecutionEnvironment env, OrphanCleanConfig config, Integer parallelism) + throws Exception { + env.setRuntimeMode(RuntimeExecutionMode.BATCH); + + // Stage 1: ScopeEnumerator (parallelism=1) + DataStream<Integer> trigger = + env.fromCollection(Collections.singletonList(1), TypeInformation.of(Integer.class)); + + SingleOutputStreamOperator<CleanTask> tasks = + trigger.process(new ScopeEnumeratorFunction(config)) + .returns(TypeInformation.of(new TypeHint<CleanTask>() {})) + .setParallelism(1) + .name("ScopeEnumerator"); + + // Stage 2: ScanAndClean (parallelism=N) + SingleOutputStreamOperator<CleanStats> stats = + tasks.rebalance() + .process( + new ScanAndCleanFunction( + config.deleteRateLimitPerSecond(), config.extraConfigs())) + .returns(TypeInformation.of(new TypeHint<CleanStats>() {})) + .name("ScanAndClean"); + if (parallelism != null) { + stats = stats.setParallelism(parallelism); + } + + // Stage 3: StatsAggregate + EmptyDirSweep (parallelism=1) + SingleOutputStreamOperator<CleanStats> result = + stats.transform( + "StatsAggregate", + TypeInformation.of(new TypeHint<CleanStats>() {}), + new StatsAggregateOperator(config.dryRun())) + .setParallelism(1); Review Comment: dito: also setMaxParallelism ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/StatsAggregateOperator.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.job; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.flink.action.orphan.audit.AuditLogger; +import org.apache.fluss.fs.FsPath; + +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.BoundedOneInput; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +/** + * Stage 3 of the orphan files cleanup job. Runs at parallelism=1 to aggregate {@link CleanStats} + * from all Stage 2 subtasks and perform the final empty-directory sweep. + * + * <p>Implemented as a custom operator (not ProcessFunction) because {@code ProcessOperator} does + * not implement {@link BoundedOneInput} — the {@code endInput()} callback would never fire. This + * operator accumulates all incoming stats and performs the empty-dir sweep in {@code endInput()}. + */ +@Internal +public final class StatsAggregateOperator extends AbstractStreamOperator<CleanStats> + implements OneInputStreamOperator<CleanStats, CleanStats>, BoundedOneInput { + + private static final long serialVersionUID = 1L; + private static final Logger LOG = LoggerFactory.getLogger(StatsAggregateOperator.class); + + private final boolean dryRun; + + private transient CleanStats accumulated; + + public StatsAggregateOperator(boolean dryRun) { + this.dryRun = dryRun; Review Comment: should init filesystem with arguments in here? ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java: ########## @@ -0,0 +1,222 @@ +/* + * 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.job; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.flink.action.orphan.audit.AuditLogger; +import org.apache.fluss.flink.action.orphan.fs.SafeDeleter; +import org.apache.fluss.flink.action.orphan.rule.BucketActiveRefs; +import org.apache.fluss.flink.action.orphan.rule.Decision; +import org.apache.fluss.flink.action.orphan.rule.FileMeta; +import org.apache.fluss.flink.action.orphan.rule.FileRule; +import org.apache.fluss.flink.action.orphan.rule.RuleDispatcher; +import org.apache.fluss.fs.FileStatus; +import org.apache.fluss.fs.FileSystem; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.shaded.guava32.com.google.common.util.concurrent.RateLimiter; + +import org.apache.flink.streaming.api.functions.ProcessFunction; +import org.apache.flink.util.Collector; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Map; + +/** + * Stage 2 of the orphan files cleanup job. Runs at user-configured parallelism (N) and performs + * pure FS operations — no coordinator RPC interaction. + * + * <p>Each subtask processes assigned {@link CleanTask} items serially: + * + * <ul> + * <li>{@link BucketCleanTask}: second-reads manifests from object storage to build the active + * reference set, then walks log/kv directories and deletes orphan files. + * <li>{@link OrphanDirCleanTask}: recursively walks the orphan directory and deletes all files + * older than the cutoff. + * </ul> + * + * <p>Each task emits its own {@link CleanStats} immediately upon completion. Delete rate is limited + * per-subtask: {@code configuredRate / runtimeParallelism}. The serial processing within each + * subtask guarantees no concurrent throttler access. + */ +@Internal +public final class ScanAndCleanFunction extends ProcessFunction<CleanTask, CleanStats> { + + private static final long serialVersionUID = 1L; + private static final Logger LOG = LoggerFactory.getLogger(ScanAndCleanFunction.class); + + private final long deleteRateLimitPerSecond; + private final Map<String, String> extraConfigs; + + private transient AuditLogger audit; + + public ScanAndCleanFunction(long deleteRateLimitPerSecond, Map<String, String> extraConfigs) { + this.deleteRateLimitPerSecond = deleteRateLimitPerSecond; + this.extraConfigs = extraConfigs; + } + + @Override + public void open(org.apache.flink.configuration.Configuration parameters) { + if (!extraConfigs.isEmpty()) { + FileSystem.initialize(Configuration.fromMap(extraConfigs), null); + } + audit = new AuditLogger(); + } + + @Override + public void processElement(CleanTask task, Context ctx, Collector<CleanStats> out) + throws Exception { + if (task instanceof BucketCleanTask) { + out.collect(processBucketTask((BucketCleanTask) task)); + } else if (task instanceof OrphanDirCleanTask) { + out.collect(processOrphanDirTask((OrphanDirCleanTask) task)); + } + } + + // ------------------------------------------------------------------------- + // BucketCleanTask processing + // ------------------------------------------------------------------------- + + private CleanStats processBucketTask(BucketCleanTask task) throws IOException { + FsPath logDir = task.logTabletDir() != null ? new FsPath(task.logTabletDir()) : null; + FsPath kvDir = task.kvTabletDir() != null ? new FsPath(task.kvTabletDir()) : null; + + FsPath anyDir = logDir != null ? logDir : kvDir; + if (anyDir == null) { + return CleanStats.empty(); + } + + BucketActiveRefs activeRefs = + new BucketActiveRefs( + task.logSegmentRelativePaths(), + task.kvActiveSnapDirs(), + task.logActiveManifestPaths()); + RuleDispatcher dispatcher = new RuleDispatcher(task.allowDeleteManifest()); + SafeDeleter safeDeleter = createSafeDeleter(anyDir.getFileSystem(), task.dryRun()); + BucketCleaner cleaner = + new BucketCleaner(dispatcher, safeDeleter, audit, task.cutoffMillis()); + + BucketCleaner.BucketCleanStats bucketStats = cleaner.clean(activeRefs, logDir, kvDir); + + List<String> touchedDirs = new ArrayList<String>(); + if (logDir != null) { + touchedDirs.add(logDir.toString()); + } + if (kvDir != null) { + touchedDirs.add(kvDir.toString()); + } + + return new CleanStats( + bucketStats.scanned, + bucketStats.deleted, + bucketStats.deleteFailures, + bucketStats.bytesReclaimed, + touchedDirs); + } + + // ------------------------------------------------------------------------- + // OrphanDirCleanTask processing + // ------------------------------------------------------------------------- + + private CleanStats processOrphanDirTask(OrphanDirCleanTask task) throws IOException { + FsPath dirPath = new FsPath(task.dirPath()); + FileSystem fs = dirPath.getFileSystem(); + if (!fs.exists(dirPath)) { + return CleanStats.empty(); + } + + SafeDeleter safeDeleter = createSafeDeleter(fs, task.dryRun()); + RuleDispatcher dispatcher = new RuleDispatcher(task.allowDeleteManifest(), true); + + long scanned = 0L; + long deleted = 0L; + long deleteFailures = 0L; + long bytesReclaimed = 0L; + + Deque<FsPath> stack = new ArrayDeque<FsPath>(); + stack.push(dirPath); + while (!stack.isEmpty()) { + FsPath dir = stack.pop(); + FileStatus[] children; + try { + children = fs.listStatus(dir); + } catch (IOException ignored) { + continue; Review Comment: nit: info or warn it? ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java: ########## @@ -776,6 +788,139 @@ public CompletableFuture<MetadataResponse> metadata(MetadataRequest request) { return metadataResponseAccessContextEvent.getResultFuture(); } + @Override + public CompletableFuture<ListRemoteLogManifestsResponse> listRemoteLogManifests( + ListRemoteLogManifestsRequest request) { + long tableId = request.getTableId(); + // Capture session before entering async block since currentSession() is thread-local + Session session = authorizer != null ? currentSession() : null; + return CompletableFuture.supplyAsync( + () -> { + if (authorizer != null) { + authorizeTableWithSession(session, OperationType.DESCRIBE, tableId); + } + Long partitionId = request.hasPartitionId() ? request.getPartitionId() : null; + ListRemoteLogManifestsResponse response = new ListRemoteLogManifestsResponse(); + try { + List<ZooKeeperClient.TableBucketAndManifest> entries = + zkClient.listRemoteLogManifestHandles(tableId, partitionId); + for (ZooKeeperClient.TableBucketAndManifest entry : entries) { + PbRemoteLogManifestEntry pb = response.addManifest(); + PbTableBucket pbTb = pb.setTableBucket(); + pbTb.setTableId(entry.getTableBucket().getTableId()); + if (entry.getTableBucket().getPartitionId() != null) { + pbTb.setPartitionId(entry.getTableBucket().getPartitionId()); + } + pbTb.setBucketId(entry.getTableBucket().getBucket()); + pb.setRemoteLogManifestPath( + entry.getManifestHandle() + .getRemoteLogManifestPath() + .toString()); + pb.setRemoteLogEndOffset( + entry.getManifestHandle().getRemoteLogEndOffset()); + } + } catch (ApiException e) { + throw e; + } catch (Exception e) { + throw new UnknownServerException( + "Failed to list remote log manifests for tableId=" + tableId, e); + } + return response; + }, + ioExecutor); + } + + @Override + public CompletableFuture<ListKvSnapshotsResponse> listKvSnapshots( + ListKvSnapshotsRequest request) { + long tableId = request.getTableId(); + Session session = authorizer != null ? currentSession() : null; + return CompletableFuture.supplyAsync( + () -> { + if (authorizer != null) { + authorizeTableWithSession(session, OperationType.DESCRIBE, tableId); + } + Long partitionId = request.hasPartitionId() ? request.getPartitionId() : null; + + // Resolve table metadata: prefer in-memory context, fallback to ZK assignment + // (single getData) to handle the race between createTable and context sync. + CoordinatorContext context = coordinatorContextSupplier.get(); + TablePath tablePath = context.getTablePathById(tableId); Review Comment: not sure concurrent access in here should be considered as a problem, one thread(event thread) may put table to the map, but this thread get table from map. ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/rule/KvSharedSstRule.java: ########## @@ -0,0 +1,53 @@ +/* + * 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.rule; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.utils.FlussPaths; + +/** + * Rule for shared SST files under the {@code shared/} KV directory. + * + * <p>Always returns {@link Decision#KEEP_ACTIVE}. The true active set for shared SSTs lives inside Review Comment: I think we could theoretically clean `shared/*.sst` more precisely: after getting the active snapshots, read each snapshot's metadata file and build the set of live SST files referenced by those snapshots, then delete only shared SSTs that are not in that live set. But I’m also fine with keeping all shared SSTs in this first version for simplicity and safety, especially if truly orphaned shared SSTs are expected to be relatively rare compared to log segment ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/OrphanFilesCleanJob.java: ########## @@ -0,0 +1,109 @@ +/* + * 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.job; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.flink.action.orphan.config.OrphanCleanConfig; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.api.common.typeinfo.TypeHint; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; + +import java.util.Collections; +import java.util.Iterator; +import java.util.List; + +/** + * Builds and executes the 3-stage Flink Batch DAG for orphan files cleanup. + * + * <pre> + * Stage 1: ScopeEnumerator (p=1) — coordinator RPCs, emits CleanTask + * Stage 2: ScanAndClean (p=N) — FS scan + rate-limited delete, emits CleanStats + * Stage 3: StatsAggregate (p=1) — merge stats + empty-dir sweep, emits final CleanStats + * </pre> + */ +@Internal +public final class OrphanFilesCleanJob { + + private OrphanFilesCleanJob() {} + + /** + * Builds the DAG, executes it in batch mode, and returns the final aggregated cleanup + * statistics. + * + * @param env the Flink execution environment (caller configures classpath, etc.) + * @param config parsed orphan cleanup configuration + * @param parallelism the parallelism for Stage 2 (ScanAndClean); null uses env default + * @return the final cleanup statistics + */ + public static CleanStats execute( + StreamExecutionEnvironment env, OrphanCleanConfig config, Integer parallelism) + throws Exception { + env.setRuntimeMode(RuntimeExecutionMode.BATCH); + + // Stage 1: ScopeEnumerator (parallelism=1) + DataStream<Integer> trigger = + env.fromCollection(Collections.singletonList(1), TypeInformation.of(Integer.class)); + + SingleOutputStreamOperator<CleanTask> tasks = + trigger.process(new ScopeEnumeratorFunction(config)) + .returns(TypeInformation.of(new TypeHint<CleanTask>() {})) + .setParallelism(1) Review Comment: also set maxParallelism(1) to enfore the parallelism to be 1 since user may overwrite the parallelism ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/BucketCleaner.java: ########## @@ -0,0 +1,142 @@ +/* + * 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.job; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.flink.action.orphan.audit.AuditLogger; +import org.apache.fluss.flink.action.orphan.fs.SafeDeleter; +import org.apache.fluss.flink.action.orphan.rule.BucketActiveRefs; +import org.apache.fluss.flink.action.orphan.rule.Decision; +import org.apache.fluss.flink.action.orphan.rule.FileMeta; +import org.apache.fluss.flink.action.orphan.rule.FileRule; +import org.apache.fluss.flink.action.orphan.rule.RuleDispatcher; +import org.apache.fluss.fs.FileStatus; +import org.apache.fluss.fs.FileSystem; +import org.apache.fluss.fs.FsPath; + +import java.io.IOException; +import java.util.ArrayDeque; +import java.util.Deque; + +/** + * Per-bucket orphan cleanup for live buckets: walks the provided bucket directories and dispatches + * each file to the appropriate {@link FileRule} using the caller-supplied active reference set. + * + * <p>All deletions go through {@link SafeDeleter} (no recursive deletes). Unknown file types are + * skipped with an audit warning per the design's "unknown-types-not-deleted" principle. + */ +@Internal +public final class BucketCleaner { + + private final RuleDispatcher dispatcher; + private final SafeDeleter safeDeleter; + private final AuditLogger audit; + private final long cutoffMillis; + + public BucketCleaner( + RuleDispatcher dispatcher, + SafeDeleter safeDeleter, + AuditLogger audit, + long cutoffMillis) { + this.dispatcher = dispatcher; + this.safeDeleter = safeDeleter; + this.audit = audit; + this.cutoffMillis = cutoffMillis; + } + + /** Cleans one bucket's log/kv subtrees using the caller-supplied active reference set. */ + public BucketCleanStats clean(BucketActiveRefs activeRefs, FsPath... bucketDirs) + throws IOException { + BucketCleanStats stats = BucketCleanStats.empty(); + for (FsPath bucketDir : bucketDirs) { + if (bucketDir != null) { + walkAndCleanDir(bucketDir, activeRefs, stats); + } + } + return stats; + } + + private void walkAndCleanDir(FsPath root, BucketActiveRefs activeRefs, BucketCleanStats stats) + throws IOException { + FileSystem fs = root.getFileSystem(); + if (!fs.exists(root)) { + return; + } + Deque<FsPath> stack = new ArrayDeque<FsPath>(); + stack.push(root); + while (!stack.isEmpty()) { + FsPath dir = stack.pop(); + FileStatus[] children; + try { + children = fs.listStatus(dir); Review Comment: nit: since shared sst will never be deleted, can we skill the list for shared sst path? ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/OrphanCleanUtils.java: ########## @@ -0,0 +1,151 @@ +/* + * 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; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.cluster.ConfigEntry; +import org.apache.fluss.fs.FileStatus; +import org.apache.fluss.fs.FileSystem; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.metadata.PartitionInfo; +import org.apache.fluss.metadata.PhysicalTablePath; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** Shared utility methods for the orphan files cleanup action. */ +@Internal +public final class OrphanCleanUtils { + + private OrphanCleanUtils() {} + + /** + * Constructs a {@link PhysicalTablePath} from a table path and an optional partition. Returns + * the non-partitioned form when {@code partitionInfo} is null. + */ + public static PhysicalTablePath physicalPath( + TablePath tablePath, @Nullable PartitionInfo partitionInfo) { + if (partitionInfo == null) { + return PhysicalTablePath.of(tablePath); + } + return PhysicalTablePath.of(tablePath, partitionInfo.getPartitionName()); + } + + /** + * Enumerates all {@link TableBucket} instances for a table (or a single partition of that + * table). + */ + public static List<TableBucket> enumerateBuckets( + TableInfo tableInfo, @Nullable PartitionInfo partitionInfo) { + int n = tableInfo.getNumBuckets(); + List<TableBucket> buckets = new ArrayList<TableBucket>(n); + long tableId = tableInfo.getTableId(); + for (int b = 0; b < n; b++) { + if (partitionInfo == null) { + buckets.add(new TableBucket(tableId, b)); + } else { + buckets.add(new TableBucket(tableId, partitionInfo.getPartitionId(), b)); + } + } + return buckets; + } + + /** + * Resolves the effective remote data directory for a table/partition target using the + * three-level fallback: partition-level → table-level → cluster-level. + */ + @Nullable + public static String resolveRemoteDataDir( + TableInfo tableInfo, + @Nullable PartitionInfo partitionInfo, + @Nullable String clusterRemoteDataDir) { Review Comment: nit: just curious about when the clusterRemoteDataDir will be null ########## fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java: ########## @@ -371,6 +375,30 @@ public CompletableFuture<MetadataResponse> metadata(MetadataRequest request) { return CompletableFuture.completedFuture(metadataResponse); } + @Override + public CompletableFuture<ListRemoteLogManifestsResponse> listRemoteLogManifests( Review Comment: can we just move `listRemoteLogManifests` and `listKvSnapshots` to AdminGateway? ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScopeEnumeratorFunction.java: ########## @@ -0,0 +1,538 @@ +/* + * 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.job; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.client.Connection; +import org.apache.fluss.client.ConnectionFactory; +import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.flink.action.orphan.OrphanCleanUtils; +import org.apache.fluss.flink.action.orphan.RpcErrorClassifier; +import org.apache.fluss.flink.action.orphan.audit.AuditLogger; +import org.apache.fluss.flink.action.orphan.build.ActiveRefsFetcher; +import org.apache.fluss.flink.action.orphan.build.KvActiveRefsFetchResult; +import org.apache.fluss.flink.action.orphan.build.LogActiveRefsFetchResult; +import org.apache.fluss.flink.action.orphan.build.MaxKnownIdsTracker; +import org.apache.fluss.flink.action.orphan.config.OrphanCleanConfig; +import org.apache.fluss.flink.action.orphan.rule.OrphanDirDetector; +import org.apache.fluss.fs.FileStatus; +import org.apache.fluss.fs.FileSystem; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.metadata.PartitionInfo; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.FlussPaths; + +import org.apache.flink.streaming.api.functions.ProcessFunction; +import org.apache.flink.util.Collector; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Consumer; +import java.util.function.Predicate; + +import static org.apache.fluss.flink.action.orphan.OrphanCleanUtils.enumerateBuckets; +import static org.apache.fluss.flink.action.orphan.OrphanCleanUtils.getFileSystemIfExists; +import static org.apache.fluss.flink.action.orphan.OrphanCleanUtils.listStatuses; +import static org.apache.fluss.flink.action.orphan.OrphanCleanUtils.physicalPath; +import static org.apache.fluss.flink.action.orphan.OrphanCleanUtils.remoteSubDir; +import static org.apache.fluss.flink.action.orphan.OrphanCleanUtils.resolveClusterRemoteDataDir; +import static org.apache.fluss.flink.action.orphan.OrphanCleanUtils.resolveRemoteDataDir; + +/** + * Stage 1 of the orphan files cleanup job. Runs at parallelism=1 and concentrates all coordinator + * RPC interaction in a single subtask. + * + * <p>For each live bucket, emits a {@link BucketCleanTask} containing the FS paths and manifest + * locations needed for Stage 2 to execute cleanup without coordinator access. For each detected + * orphan directory, emits an {@link OrphanDirCleanTask}. + */ +@Internal +public final class ScopeEnumeratorFunction extends ProcessFunction<Integer, CleanTask> { + + private static final long serialVersionUID = 1L; + private static final Logger LOG = LoggerFactory.getLogger(ScopeEnumeratorFunction.class); + private static final String[] TOP_LEVEL_DIRS = { + FlussPaths.REMOTE_LOG_DIR_NAME, FlussPaths.REMOTE_KV_DIR_NAME + }; + + private final OrphanCleanConfig config; + + public ScopeEnumeratorFunction(OrphanCleanConfig config) { + this.config = config; + } + + @Override + public void processElement(Integer trigger, Context ctx, Collector<CleanTask> out) + throws Exception { + if (!config.extraConfigs().isEmpty()) { + FileSystem.initialize(Configuration.fromMap(config.extraConfigs()), null); + } + + Configuration flussConfig = new Configuration(); + flussConfig.setString(ConfigOptions.BOOTSTRAP_SERVERS.key(), config.bootstrapServer()); + + try (Connection connection = ConnectionFactory.createConnection(flussConfig); + Admin admin = connection.getAdmin()) { + AuditLogger audit = new AuditLogger(); + audit.logCutoff(config.olderThanMillis()); + + ActiveRefsFetcher fetcher = new ActiveRefsFetcher(admin, 3); + MaxKnownIdsTracker tracker = new MaxKnownIdsTracker(); + String clusterRemoteDataDir = resolveClusterRemoteDataDir(admin); + + Map<String, DbScanState> dbStates = enumerateActiveScope(admin, audit, tracker); + + for (DbScanState dbState : dbStates.values()) { + for (LiveTableScope liveTable : dbState.liveTables) { + emitBucketTasks(liveTable, fetcher, audit, clusterRemoteDataDir, out); + emitOrphanPartitionDirTasks( + liveTable, tracker, clusterRemoteDataDir, audit, out); + } + emitOrphanTableDirTasks(dbState, tracker, clusterRemoteDataDir, audit, out); + } + } + } + + // ------------------------------------------------------------------------- + // Scope enumeration (coordinator RPCs only) + // ------------------------------------------------------------------------- + + private Map<String, DbScanState> enumerateActiveScope( + Admin admin, AuditLogger audit, MaxKnownIdsTracker tracker) { + List<String> dbs = resolveDatabasesToScan(admin, audit); + Map<String, DbScanState> result = new LinkedHashMap<String, DbScanState>(); + for (String dbName : dbs) { + DbScanState dbState = new DbScanState(dbName); + result.put(dbName, dbState); + if (config.table().isPresent()) { + dbState.tableInfosComplete = false; + resolveTable(admin, audit, tracker, dbState, config.table().get(), true); + continue; + } + List<String> tableNames; + try { + tableNames = admin.listTables(dbName).get(); + } catch (Exception e) { + audit.logSkipDb(dbName, classifyName(e)); + dbState.tableInfosComplete = false; + continue; + } + for (String tableName : tableNames) { + resolveTable(admin, audit, tracker, dbState, tableName, false); + } + } + return result; + } + + private List<String> resolveDatabasesToScan(Admin admin, AuditLogger audit) { + if (config.allDatabases()) { + try { + return admin.listDatabases().get(); + } catch (Exception e) { + audit.logSkipDb("*", classifyName(e)); + return Collections.emptyList(); + } + } + String databaseName = config.database().get(); + try { + if (admin.databaseExists(databaseName).get()) { + return Collections.singletonList(databaseName); + } + } catch (Exception e) { + audit.logSkipDb(databaseName, classifyName(e)); + return Collections.emptyList(); + } + audit.logSkipDb(databaseName, RpcErrorClassifier.Category.NOT_FOUND.name()); + return Collections.emptyList(); + } + + private void resolveTable( + Admin admin, + AuditLogger audit, + MaxKnownIdsTracker tracker, + DbScanState dbState, + String tableName, + boolean explicitTableTarget) { + TablePath tablePath = TablePath.of(dbState.dbName, tableName); + TableInfo tableInfo; + try { + tableInfo = admin.getTableInfo(tablePath).get(); + } catch (Exception e) { + RpcErrorClassifier.Category category = RpcErrorClassifier.classify(e); + if (category != RpcErrorClassifier.Category.NOT_FOUND || explicitTableTarget) { + audit.logSkipTable(dbState.dbName, tableName, category.name()); + dbState.tableInfosComplete = false; + } + return; + } + tracker.observeTableId(tableInfo.getTableId()); + dbState.activeTableIds.add(tableInfo.getTableId()); + + LiveTableScope liveTable = new LiveTableScope(dbState.dbName, tableName, tableInfo); + dbState.liveTables.add(liveTable); + if (!tableInfo.isPartitioned()) { + return; + } + try { + List<PartitionInfo> partitions = admin.listPartitionInfos(tablePath).get(); + TableInfo confirm = admin.getTableInfo(tablePath).get(); + if (confirm.getTableId() != tableInfo.getTableId()) { + audit.logSkipTable(dbState.dbName, tableName, "ABA"); Review Comment: nit: `ABA` is pretty terse for an audit reason. This log is likely to be read by operators, and `ABA` does not explain what happened unless the reader knows the concurrency term. Could we use something more explicit, e.g. `table-recreated-during-enumeration` or include the old/new table ids? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
