wuchong commented on code in PR #3404: URL: https://github.com/apache/fluss/pull/3404#discussion_r3408280226
########## fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/action/orphan/OrphanFilesCleanITCase.java: ########## @@ -0,0 +1,1310 @@ +/* + * 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.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.config.OrphanCleanConfig; +import org.apache.fluss.flink.adapter.MultipleParameterToolAdapter; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.metadata.DatabaseDescriptor; +import org.apache.fluss.metadata.PartitionInfo; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.PhysicalTablePath; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.server.testutils.FlussClusterExtension; +import org.apache.fluss.server.zk.ZooKeeperClient; +import org.apache.fluss.server.zk.data.BucketSnapshot; +import org.apache.fluss.server.zk.data.RemoteLogManifestHandle; +import org.apache.fluss.server.zk.data.ZkData.BucketSnapshotIdZNode; +import org.apache.fluss.server.zk.data.ZkData.PartitionZNode; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.utils.FlussPaths; + +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.LoggerContext; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.LoggerConfig; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.attribute.FileTime; +import java.time.Duration; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CopyOnWriteArrayList; + +import static org.assertj.core.api.Assertions.assertThat; + +/** End-to-end tests for orphan files cleanup safety scenarios. */ +class OrphanFilesCleanITCase { Review Comment: `extends AbstractTestBase` to reuse Flink minicluster resources. ########## fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java: ########## @@ -796,4 +798,32 @@ CompletableFuture<RegisterResult> registerProducerOffsets( * @since 1.0 */ CompletableFuture<ClusterHealth> getClusterHealth(); + + /** + * 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 + */ + @PublicEvolving + default CompletableFuture<List<RemoteLogManifestInfo>> listRemoteLogManifests( + long tableId, @Nullable Long partitionId) { + throw new UnsupportedOperationException( Review Comment: no default implementation ########## 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: Why not clean up the `shared` directory? It stores all `.sst` files, which constitute the majority of remote KV data. Orphaned SST files are highly likely to remain if an async KV snapshot fails. ########## fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java: ########## @@ -796,4 +798,32 @@ CompletableFuture<RegisterResult> registerProducerOffsets( * @since 1.0 */ CompletableFuture<ClusterHealth> getClusterHealth(); + + /** + * 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 + */ + @PublicEvolving + default CompletableFuture<List<RemoteLogManifestInfo>> 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 ids grouped by bucket + */ + @PublicEvolving + default CompletableFuture<ActiveKvSnapshots> listKvSnapshots( + long tableId, @Nullable Long partitionId) { + throw new UnsupportedOperationException( Review Comment: no default implementation ########## 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 + * the engine's {@code SharedKvFileRegistry}; orphan cleanup has no read path into that registry, so + * any deletion here would be a guess. Per the action's hard constraint "prefer leak over + * mis-delete," the rule never deletes, and as a consequence orphan PK-table / orphan-partition + * directories permanently retain their {@code shared/} subtree as accepted residue (recovering that + * residue would require a registry-backed GC channel that is out of scope for this action). + */ +@Internal +public final class KvSharedSstRule implements FileRule { + + @Override + public RuleId id() { + return RuleId.KV_SHARED_SST; + } + + @Override + public Decision evaluate(FileMeta file, BucketActiveRefs activeRefs, long cutoffMillis) { + FsPath parent = file.path().getParent(); + if (parent == null || !FlussPaths.REMOTE_KV_SNAPSHOT_SHARED_DIR.equals(parent.getName())) { + return Decision.SKIP_UNKNOWN; + } + if (!file.path().getName().endsWith(".sst")) { + return Decision.SKIP_UNKNOWN; + } + return Decision.KEEP_ACTIVE; Review Comment: Why never delete SST files? ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/OrphanFilesCleanActionFactory.java: ########## @@ -0,0 +1,76 @@ +/* + * 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.flink.action.Action; +import org.apache.fluss.flink.action.ActionFactory; +import org.apache.fluss.flink.action.orphan.config.OrphanCleanConfig; +import org.apache.fluss.flink.adapter.MultipleParameterToolAdapter; + +import java.util.Optional; + +/** Factory for the shell-mode orphan files cleanup action. */ +@Internal +public class OrphanFilesCleanActionFactory implements ActionFactory { + + @Override + public String identifier() { + return "orphan_files_clean"; + } + + @Override + public Optional<Action> create(MultipleParameterToolAdapter params) { + return Optional.<Action>of( + new OrphanFilesCleanAction(OrphanCleanConfig.fromParams(params))); + } + + @Override + public String help() { + return "Usage: orphan_files_clean --bootstrap-server <host:port>\n" Review Comment: Please consider renaming this action to `remove_orphan_files` to align with industry standards like Apache Iceberg and Paimon. This consistency will be particularly beneficial as we introduce the CALL procedure in the near future. References: https://iceberg.apache.org/docs/latest/spark-procedures/#remove_orphan_files https://paimon.apache.org/docs/master/spark/procedures/ ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/build/ActiveRefsFetcher.java: ########## @@ -0,0 +1,359 @@ +/* + * 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.build; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.client.metadata.ActiveKvSnapshots; +import org.apache.fluss.client.metadata.RemoteLogManifestInfo; +import org.apache.fluss.flink.action.orphan.RpcErrorClassifier; +import org.apache.fluss.flink.action.orphan.rule.BucketActiveRefs; +import org.apache.fluss.fs.FSDataInputStream; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.fluss.utils.FlussPaths; +import org.apache.fluss.utils.IOUtils; +import org.apache.fluss.utils.RetryUtils; + +import javax.annotation.Nullable; + +import java.io.ByteArrayOutputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; + +import static org.apache.fluss.utils.Preconditions.checkArgument; + +/** + * Builds the active reference set for a single {@code (tableId, partitionId|null)} target, sourced + * from coordinator metadata via RPC (not from filesystem listing). + * + * <p>Log path: discovers each bucket's current remote log manifest path via {@code + * LIST_REMOTE_LOG_MANIFESTS}, then second-reads the manifest file from object storage. The + * per-target RPC is retried with exponential backoff via {@link RetryUtils}; per-bucket + * second-reads make a single attempt — a {@link FileNotFoundException} (manifest upserted between + * RPC and read) or any other IO failure immediately marks the bucket as {@link + * LogActiveRefsFetchResult.ManifestReadStatus#READ_FAILED} and recovery is left to the next cleanup + * round, avoiding {@code N × retries × IO} blow-up on cluster-wide turbulence. + * + * <p>KV path: {@code LIST_KV_SNAPSHOTS} returns snapshot ids directly (no second-read), so the + * per-target RPC retry alone is sufficient symmetry with the log path. + */ +@Internal +public final class ActiveRefsFetcher { + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + private static final String REMOTE_LOG_SEGMENTS_FIELD = "remote_log_segments"; + private static final String SEGMENT_ID_FIELD = "segment_id"; + private static final String START_OFFSET_FIELD = "start_offset"; + private static final String END_OFFSET_FIELD = "end_offset"; + + /** + * Retry backoff base used by {@link RetryUtils} for per-target RPCs. With the default 3 retries + * and exponential backoff (200 → 400 → cap) this caps total retry delay at ~600ms — negligible + * vs the smoothing it gives over server jitter. + */ + private static final long DEFAULT_BACKOFF_MILLIS = 200L; + + private static final long MAX_BACKOFF_MILLIS = 2000L; + + private static final MetadataReader DEFAULT_METADATA_READER = + new MetadataReader() { + @Override + public byte[] read(FsPath path) throws IOException { + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + try (FSDataInputStream inputStream = path.getFileSystem().open(path)) { + IOUtils.copyBytes(inputStream, outputStream); + } + return outputStream.toByteArray(); + } + }; + + private final AdminFacade admin; + private final MetadataReader metadataReader; + private final int maxRetries; + private final long backoffMillis; + + public ActiveRefsFetcher(Admin admin, int maxRetries) { + this(wrap(admin), DEFAULT_METADATA_READER, maxRetries, DEFAULT_BACKOFF_MILLIS); + } + + public ActiveRefsFetcher(Admin admin, MetadataReader metadataReader, int maxRetries) { + this(wrap(admin), metadataReader, maxRetries, DEFAULT_BACKOFF_MILLIS); + } + + /** Test constructor: defaults backoff to 0 so unit tests don't pay retry sleep. */ + @VisibleForTesting + ActiveRefsFetcher(AdminFacade admin, MetadataReader metadataReader, int maxRetries) { + this(admin, metadataReader, maxRetries, 0L); + } + + @VisibleForTesting + ActiveRefsFetcher( + AdminFacade admin, MetadataReader metadataReader, int maxRetries, long backoffMillis) { + checkArgument(maxRetries >= 1, "maxRetries must be >= 1, got %s", maxRetries); + checkArgument(backoffMillis >= 0L, "backoffMillis must be >= 0, got %s", backoffMillis); + this.admin = admin; + this.metadataReader = metadataReader; + this.maxRetries = maxRetries; + this.backoffMillis = backoffMillis; + } + + private static AdminFacade wrap(Admin admin) { + return new AdminFacade() { + @Override + public CompletableFuture<List<RemoteLogManifestInfo>> listRemoteLogManifests( + long tableId, @Nullable Long partitionId) { + return admin.listRemoteLogManifests(tableId, partitionId); + } + + @Override + public CompletableFuture<ActiveKvSnapshots> listKvSnapshots( + long tableId, @Nullable Long partitionId) { + return admin.listKvSnapshots(tableId, partitionId); + } + }; + } + + /** + * Fetches per-bucket log active refs for a single {@code (tableId, partitionId|null)} target. + * Each bucket whose remote manifest is returned by the RPC is second-read in a single attempt; + * a {@link FileNotFoundException} or any other IO failure marks the bucket as {@link + * LogActiveRefsFetchResult.ManifestReadStatus#READ_FAILED} without affecting siblings. + * Per-target RPC failure (after retries) is reported via {@link + * LogActiveRefsFetchResult#listOk()}. + */ + public LogActiveRefsFetchResult fetchLogActiveRefsByBucket( + long tableId, @Nullable Long partitionId) { + List<RemoteLogManifestInfo> manifests; + try { + manifests = + RetryUtils.executeWithRetry( + () -> admin.listRemoteLogManifests(tableId, partitionId).get(), + "listRemoteLogManifests", + maxRetries, + backoffMillis, + MAX_BACKOFF_MILLIS, + e -> + RpcErrorClassifier.classify(e) + != RpcErrorClassifier.Category.NOT_FOUND); + } catch (IOException e) { + return LogActiveRefsFetchResult.listFailed( + formatRpcFailureReason(tableId, partitionId, e.getCause())); + } + + Map<Integer, List<RemoteLogManifestInfo>> entriesByBucket = new HashMap<>(); + for (RemoteLogManifestInfo entry : manifests) { + int bucketId = entry.getTableBucket().getBucket(); + entriesByBucket.computeIfAbsent(bucketId, id -> new ArrayList<>()).add(entry); + } + + Map<Integer, BucketActiveRefs> resolved = new HashMap<>(); + Map<Integer, String> readFailures = new HashMap<>(); + for (Map.Entry<Integer, List<RemoteLogManifestInfo>> bucketEntries : + entriesByBucket.entrySet()) { + int bucketId = bucketEntries.getKey(); + try { + resolved.put(bucketId, buildBucketActiveRefs(bucketEntries.getValue())); + } catch (FileNotFoundException e) { + readFailures.put( + bucketId, + formatBucketReadFailureReason( + "Manifest not found (likely upserted concurrently)", + tableId, + partitionId, + bucketId, + e)); + } catch (ManifestParseException | JsonProcessingException e) { + // Manifest payload is unreadable as JSON or violates the expected shape — corrupt + // or schema-skewed, not a transient FS hiccup. Distinct reason so operators triage + // separately (re-running the action will not recover). + readFailures.put( + bucketId, + formatBucketReadFailureReason( + "Manifest parse failure (corrupt or unexpected schema)", + tableId, + partitionId, + bucketId, + e)); + } catch (IOException e) { + readFailures.put( + bucketId, + formatBucketReadFailureReason( + "IO error reading manifest", tableId, partitionId, bucketId, e)); + } + } + return LogActiveRefsFetchResult.ofPerBucket(resolved, readFailures); + } + + /** + * Fetches the per-bucket active snapshot directories ({@code snap-{id}} names) for one {@code + * (tableId, partitionId|null)} target. The set per bucket is the union of RETAINED and + * STILL_IN_USE entries returned by {@link Admin#listKvSnapshots(long, Long)}. Per-target RPC + * failure (after retries) is reported via {@link KvActiveRefsFetchResult#listOk()}, symmetric + * with the log path. + */ + public KvActiveRefsFetchResult fetchKvActiveSnapDirs(long tableId, @Nullable Long partitionId) { + ActiveKvSnapshots activeKvSnapshots; + try { + activeKvSnapshots = + RetryUtils.executeWithRetry( + () -> admin.listKvSnapshots(tableId, partitionId).get(), + "listKvSnapshots", + maxRetries, + backoffMillis, + MAX_BACKOFF_MILLIS, + e -> + RpcErrorClassifier.classify(e) + != RpcErrorClassifier.Category.NOT_FOUND); + } catch (IOException e) { + return KvActiveRefsFetchResult.listFailed( + formatRpcFailureReason(tableId, partitionId, e.getCause())); + } + Map<Integer, Set<String>> dirsByBucket = new HashMap<>(); + for (Map.Entry<Integer, Set<Long>> entry : + activeKvSnapshots.getSnapshotIdsByBucket().entrySet()) { + int bucketId = entry.getKey(); + Set<String> dirNames = new HashSet<>(); + for (Long snapshotId : entry.getValue()) { + dirNames.add(FlussPaths.REMOTE_KV_SNAPSHOT_DIR_PREFIX + snapshotId); + } + dirsByBucket.put(bucketId, dirNames); + } + return KvActiveRefsFetchResult.ok(dirsByBucket); + } + + private static String formatRpcFailureReason( + long tableId, @Nullable Long partitionId, @Nullable Throwable cause) { + String reason = + String.format("RPC failure for tableId=%s partitionId=%s", tableId, partitionId); + if (cause != null && cause.getMessage() != null) { + reason = reason + ": " + cause.getMessage(); + } + return reason; + } + + private static String formatBucketReadFailureReason( + String prefix, + long tableId, + @Nullable Long partitionId, + int bucketId, + Throwable cause) { + String reason = + String.format( + "%s for tableId=%s partitionId=%s bucketId=%s", + prefix, tableId, partitionId, bucketId); + if (cause != null && cause.getMessage() != null) { + reason = reason + ": " + cause.getMessage(); + } + return reason; + } + + private BucketActiveRefs buildBucketActiveRefs(List<RemoteLogManifestInfo> entries) + throws IOException { + Set<String> manifestPaths = new HashSet<>(); + Set<String> segmentRelpaths = new HashSet<>(); + for (RemoteLogManifestInfo entry : entries) { + String path = entry.getRemoteLogManifestPath(); + manifestPaths.add(path); + byte[] manifestBytes = metadataReader.read(new FsPath(path)); + segmentRelpaths.addAll(parseLogSegmentRelativePaths(manifestBytes)); + } + return new BucketActiveRefs(segmentRelpaths, Collections.emptySet(), manifestPaths); + } + + private Set<String> parseLogSegmentRelativePaths(byte[] manifestBytes) throws IOException { Review Comment: Currently, we manually parse the manifest JSON and maintain a duplicate expected shape in this class. This is error-prone (e.g., missing required fields or version-dependent fields) and adds significant maintenance overhead. Ideally, we should move `RemoteLogManifest` and its JSON serde and tests to flink-common to enable reuse across both the client and server. Additionally, this is also much safer when the server uses a newer manifest schema which changes the fields meaning or a different file layout. ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java: ########## @@ -0,0 +1,239 @@ +/* + * 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.Arrays; +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 a single {@link CleanStats} containing scalar counters and the short list of + * directories walked. 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; + private transient RateLimiter rateLimiter; + + public ScanAndCleanFunction(long deleteRateLimitPerSecond, Map<String, String> extraConfigs) { + this.deleteRateLimitPerSecond = deleteRateLimitPerSecond; + this.extraConfigs = extraConfigs; + } + + @Override + public void open(org.apache.flink.api.common.functions.OpenContext openContext) + throws Exception { + super.open(openContext); + if (!extraConfigs.isEmpty()) { + FileSystem.initialize(Configuration.fromMap(extraConfigs), null); + } + audit = new AuditLogger(); + int parallelism = getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks(); + int subtaskIndex = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask(); + // Distribute the configured rate as base + 1 extra for the first `remainder` subtasks so + // that the per-subtask rates sum back to the configured aggregate. Each subtask gets at + // least 1/s (hard floor) — when parallelism exceeds the configured rate, the aggregate + // may theoretically exceed it; in practice Batch scheduling limits actual concurrency. + long base = deleteRateLimitPerSecond / parallelism; + long remainder = deleteRateLimitPerSecond % parallelism; + long quota = base + (subtaskIndex < remainder ? 1L : 0L); + rateLimiter = RateLimiter.create(Math.max(1.0, (double) quota)); + } + + @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>(2); + 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 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()) { + stack.push(childPath); + continue; + } + if (childPath.getName().startsWith(".")) { Review Comment: Why not clean up `.` files? This may cause some orphan dirs can never be cleanup. ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/rule/LogSegmentRule.java: ########## @@ -0,0 +1,98 @@ +/* + * 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; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.regex.Pattern; + +/** + * Rule for log-segment files under a remote log bucket. + * + * <p>{@code .writer_snapshot} files are only eligible for deletion in orphan-directory mode. In + * active-bucket mode the engine's own TTL cleanup handles them; the orphan tool conservatively + * keeps them to avoid any risk of racing a concurrent write. + */ +@Internal +public final class LogSegmentRule implements FileRule { + + private static final Pattern SEGMENT_DIR_PATTERN = + Pattern.compile( + "[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}" + + "-[0-9a-fA-F]{12}"); + + private static final Set<String> KNOWN_SUFFIXES = + new HashSet<String>(Arrays.asList(".log", ".index", ".timeindex", ".writer_snapshot")); + + private final boolean orphanDirMode; + + public LogSegmentRule() { + this(false); + } + + public LogSegmentRule(boolean orphanDirMode) { + this.orphanDirMode = orphanDirMode; + } + + @Override + public RuleId id() { + return RuleId.LOG_SEGMENT; + } + + @Override + public Decision evaluate(FileMeta file, BucketActiveRefs activeRefs, long cutoffMillis) { + FsPath path = file.path(); + FsPath parent = path.getParent(); + if (parent == null || !isSegmentDir(parent.getName()) || !hasKnownSuffix(path.getName())) { + return Decision.SKIP_UNKNOWN; + } + + String relativePath = parent.getName() + "/" + path.getName(); + if (activeRefs.logSegmentRelativePaths().contains(relativePath)) { + return Decision.KEEP_ACTIVE; + } + + if (path.getName().endsWith(FlussPaths.WRITER_SNAPSHOT_FILE_SUFFIX) && !orphanDirMode) { + return Decision.KEEP_ACTIVE; + } Review Comment: Why keep writer_snapshot files? ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java: ########## @@ -0,0 +1,239 @@ +/* + * 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.Arrays; +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 a single {@link CleanStats} containing scalar counters and the short list of + * directories walked. 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; + private transient RateLimiter rateLimiter; + + public ScanAndCleanFunction(long deleteRateLimitPerSecond, Map<String, String> extraConfigs) { + this.deleteRateLimitPerSecond = deleteRateLimitPerSecond; + this.extraConfigs = extraConfigs; + } + + @Override + public void open(org.apache.flink.api.common.functions.OpenContext openContext) + throws Exception { + super.open(openContext); + if (!extraConfigs.isEmpty()) { + FileSystem.initialize(Configuration.fromMap(extraConfigs), null); + } + audit = new AuditLogger(); + int parallelism = getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks(); + int subtaskIndex = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask(); + // Distribute the configured rate as base + 1 extra for the first `remainder` subtasks so + // that the per-subtask rates sum back to the configured aggregate. Each subtask gets at + // least 1/s (hard floor) — when parallelism exceeds the configured rate, the aggregate + // may theoretically exceed it; in practice Batch scheduling limits actual concurrency. + long base = deleteRateLimitPerSecond / parallelism; + long remainder = deleteRateLimitPerSecond % parallelism; + long quota = base + (subtaskIndex < remainder ? 1L : 0L); + rateLimiter = RateLimiter.create(Math.max(1.0, (double) quota)); + } + + @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>(2); + 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 { Review Comment: If we can definitively identify an orphan directory (e.g., orphan table or partition), we should delete the remote directory directly instead of performing a deep traversal of every file. Deep traversal generates significant QPS, which can cause backpressure and increase code complexity. Direct deletion should be safe in this scenario. Additionally, the current orphan detection logic is too simplistic. Relying solely on `activeTableIds` carries a risk of accidental deletion if there are bugs in that set. I recommend adding a secondary verification step: fetch the latest table/partition info based on the name and confirm deletion only if there is an ID mismatch (can be added in a follow-up issue/PR). ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java: ########## @@ -0,0 +1,239 @@ +/* + * 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.Arrays; +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 a single {@link CleanStats} containing scalar counters and the short list of + * directories walked. 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; + private transient RateLimiter rateLimiter; + + public ScanAndCleanFunction(long deleteRateLimitPerSecond, Map<String, String> extraConfigs) { + this.deleteRateLimitPerSecond = deleteRateLimitPerSecond; + this.extraConfigs = extraConfigs; + } + + @Override + public void open(org.apache.flink.api.common.functions.OpenContext openContext) + throws Exception { + super.open(openContext); + if (!extraConfigs.isEmpty()) { + FileSystem.initialize(Configuration.fromMap(extraConfigs), null); + } + audit = new AuditLogger(); + int parallelism = getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks(); + int subtaskIndex = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask(); + // Distribute the configured rate as base + 1 extra for the first `remainder` subtasks so + // that the per-subtask rates sum back to the configured aggregate. Each subtask gets at + // least 1/s (hard floor) — when parallelism exceeds the configured rate, the aggregate + // may theoretically exceed it; in practice Batch scheduling limits actual concurrency. + long base = deleteRateLimitPerSecond / parallelism; + long remainder = deleteRateLimitPerSecond % parallelism; + long quota = base + (subtaskIndex < remainder ? 1L : 0L); + rateLimiter = RateLimiter.create(Math.max(1.0, (double) quota)); + } + + @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>(2); + 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 e) { + LOG.warn("Failed to list directory: {}", dir, e); + continue; + } + if (children == null) { Review Comment: I suggest we remove a directory if it has no children (i.e., the children list is empty) and has existed for a sufficient duration. To implement this, we can push the parent directory onto the stack before processing its child paths. This allows us to revisit and evaluate the parent directory for deletion after all its children have been processed. My primary goal is to eliminate `EmptyDirSweeper` from `StatsAggregateOperator`. As its name implies, `StatsAggregateOperator` should focus solely on collecting statistics. The current `EmptyDirSweeper` approach is too aggressive and operates at a coarse granularity, leading to significant redundant detection overhead during cleanup. For instance, `BucketCleaner` should only clean up explicitly expired log segment or snapshot directories. It should not need to rely on `EmptyDirSweeper` to traverse the entire `logTabletDir` or `kvTabletDir`. Similarly, `OrphanDirCleanTask` should directly remove identified orphan directories without performing a deep traversal of every file within them. ########## fluss-flink/fluss-flink-common/src/main/resources/META-INF/services/org.apache.fluss.flink.action.ActionFactory: ########## @@ -0,0 +1,19 @@ +# +# 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. +# + +org.apache.fluss.flink.action.orphan.OrphanFilesCleanActionFactory Review Comment: Move the factory SPI files to specific flink version `fluss-flink-xxx` modules, like how we did for the flink table factory SPI. There is some issues when putting it in `fluss-flink-common`, but I can't clearly remember it. ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScopeEnumeratorFunction.java: ########## @@ -0,0 +1,535 @@ +/* + * 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.fetchClusterConfigMap; +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.normalizeRoot; +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.resolveClusterRemoteDataDirs; +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()) { Review Comment: I recommend checking whether the server supports the required new RPC here first. If not, we should fail fast immediately. This is crucial because the action jar might run on older clusters, so we need to ensure proper protection. Performing this check at compile time or early in the runtime phase is a safer approach. ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/StatsAggregateOperator.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.config.Configuration; +import org.apache.fluss.flink.action.orphan.audit.AuditLogger; +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.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.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * Stage 3 of the orphan files cleanup job. Runs at parallelism=1 to aggregate per-subtask {@link + * CleanStats} records 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. + * + * <p>Scalar counters are accumulated into four longs; directory paths from each incoming {@link + * CleanStats#touchedDirs()} are inserted into a {@link HashSet} for O(1) deduplication. The final + * empty-dir sweep happens in {@link #endInput()}. + */ +@Internal +public final class StatsAggregateOperator extends AbstractStreamOperator<CleanStats> + implements OneInputStreamOperator<CleanStats, CleanStats>, BoundedOneInput { + + private static final long serialVersionUID = 2L; + private static final Logger LOG = LoggerFactory.getLogger(StatsAggregateOperator.class); + + private final boolean dryRun; + private final Map<String, String> extraConfigs; + private final long deleteRateLimitPerSecond; + + private transient long scanned; + private transient long deleted; + private transient long deleteFailures; + private transient long bytesReclaimed; + private transient Set<String> touchedDirs; + private transient RateLimiter sweepRateLimiter; + + public StatsAggregateOperator( + boolean dryRun, Map<String, String> extraConfigs, long deleteRateLimitPerSecond) { + this.dryRun = dryRun; + this.extraConfigs = extraConfigs; + this.deleteRateLimitPerSecond = deleteRateLimitPerSecond; + } + + @Override + public void open() throws Exception { + super.open(); + if (!extraConfigs.isEmpty()) { + FileSystem.initialize(Configuration.fromMap(extraConfigs), null); + } + scanned = 0L; + deleted = 0L; + deleteFailures = 0L; + bytesReclaimed = 0L; + touchedDirs = new HashSet<String>(); + sweepRateLimiter = RateLimiter.create((double) deleteRateLimitPerSecond); + } + + @Override + public void processElement(StreamRecord<CleanStats> element) { + CleanStats stats = element.getValue(); + scanned += stats.scanned(); + deleted += stats.deleted(); + deleteFailures += stats.deleteFailures(); + bytesReclaimed += stats.bytesReclaimed(); + touchedDirs.addAll(stats.touchedDirs()); + } + + @Override + public void endInput() { + long emptyDirsRemoved = sweepEmptyDirs(touchedDirs); + long totalDeleted = deleted + emptyDirsRemoved; + + CleanStats finalStats = + new CleanStats( + scanned, + totalDeleted, + deleteFailures, + bytesReclaimed, + new ArrayList<String>(0)); + + LOG.info( Review Comment: Use `AuditLogger` to log this as well. So the most important information can be easily queried in audit logs. ########## fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScopeEnumeratorFunction.java: ########## @@ -0,0 +1,535 @@ +/* + * 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.fetchClusterConfigMap; +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.normalizeRoot; +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.resolveClusterRemoteDataDirs; +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(); + Map<String, String> clusterConfigMap = fetchClusterConfigMap(admin); + String clusterRemoteDataDir = resolveClusterRemoteDataDir(clusterConfigMap); + List<String> clusterRoots = + normalizeRoots(resolveClusterRemoteDataDirs(clusterConfigMap)); + + Map<String, DbScanState> dbStates = enumerateActiveScope(admin, audit, tracker); + + for (DbScanState dbState : dbStates.values()) { Review Comment: Currently, all cleanup tasks are driven by active databases. If a table has orphan files but its parent database has been deleted, those orphan files will never be cleaned up. We should consider adding a scan for orphan files within deleted (orphan) databases. Please create an issue to track this enhancement for future support. -- 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]
