platinumhamburg commented on code in PR #3404: URL: https://github.com/apache/fluss/pull/3404#discussion_r3418723138
########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/BucketCleaner.java: ########## @@ -0,0 +1,152 @@ +/* + * 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 org.apache.fluss.utils.FlussPaths; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +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 static final Logger LOG = LoggerFactory.getLogger(BucketCleaner.class); + + 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); + } catch (IOException e) { + LOG.warn("Failed to list directory: {}", dir, e); + continue; + } + if (children == null) { + continue; + } + for (FileStatus child : children) { + FsPath childPath = child.getPath(); + if (child.isDir()) { + if (FlussPaths.REMOTE_KV_SNAPSHOT_SHARED_DIR.equals(childPath.getName())) { + continue; Review Comment: @wuchong Shared SST files are indeed the majority of remote KV data, and failed async snapshots may leave orphaned shared SST files behind. However, I do not think it is safe to simply delete shared SST files from the Flink cleanup action based on active snapshot IDs, or even based on the shared file handles referenced by currently active snapshots. Shared SST files have a different lifecycle from private snapshot files. Private files under `snap-<id>/` are owned by a single snapshot directory, while SST files under `shared/` may be reused by multiple snapshots. More importantly, an externally fetched active-reference set only represents committed and visible snapshots at one point in time. It does not protect against an in-flight snapshot that has already uploaded shared SST files but has not committed yet, or a snapshot whose commit result is temporarily uncertain. For example, the cleanup action could fetch active refs before snapshot N commits, see an uploaded SST as unreferenced, delete it, and then snapshot N may commit successfully with metadata pointing to that SST. The `older-than` cutoff reduces the probability of this race, but it does not provide a correctness guarantee. The safe place to GC shared SST files is the server/engine side, where `CompletedSnapshotStore`, `SharedKvFileRegistry`, still-in-use leases, and snapshot lifecycle state are available under the proper synchronization. If we want the cleanup action to cover this in the future, I think it should trigger a server-side shared-SST GC or rely on a stronger protocol that includes in-flight snapshot protection, bucket-level snapshot locking, or epoch validation. So I agree this is an important gap, but I would avoid deleting shared SST files directly from this action in the current PR. The current `KEEP_ACTIVE` behavior is conservative by design: prefer leaking shared SST files over deleting SST files that may still be referenced by a committed or in-flight snapshot. I created an issue to track it: https://github.com/apache/fluss/issues/3489 -- 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]
