swuferhong commented on code in PR #3404:
URL: https://github.com/apache/fluss/pull/3404#discussion_r3332493980
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java:
##########
@@ -275,6 +283,10 @@ public CoordinatorService(
() ->
coordinatorEventProcessorSupplier.get().getCoordinatorEventManager();
this.coordinatorEpochSupplier =
() ->
coordinatorEventProcessorSupplier.get().getCoordinatorEpoch();
+ this.snapshotStoreManagerSupplier =
+ () ->
coordinatorEventProcessorSupplier.get().completedSnapshotStoreManager();
+ this.coordinatorContextSupplier =
+ () ->
coordinatorEventProcessorSupplier.get().getCoordinatorContext();
Review Comment:
`CoordinatorContext` is annotated `@NotThreadSafe`. Every other method in
`CoordinatorService` accesses it through an `AccessContextEvent` queued to the
coordinator event thread. This code bypasses that safety mechanism, introducing
a data race under concurrent table creation/deletion.
##########
fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java:
##########
@@ -367,6 +371,38 @@ public CompletableFuture<List<PartitionInfo>>
listPartitionInfos(
.thenApply(ClientRpcMessageUtils::toPartitionInfos);
}
+ /**
+ * Returns per-bucket remote log manifest path for the given table or
partition.
+ *
+ * <p>Used by the orphan cleanup action to construct the active manifest
path set without
+ * relying on FS LIST + mtime selection.
+ */
+ @Override
+ public CompletableFuture<ListRemoteLogManifestsResponse>
listRemoteLogManifests(
Review Comment:
Why we directly response the class `ListRemoteLogManifestsResponse` to user?
##########
fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java:
##########
@@ -770,4 +773,32 @@ CompletableFuture<RegisterResult> registerProducerOffsets(
* @since 0.9
*/
CompletableFuture<Void> deleteProducerOffsets(String producerId);
+
+ /**
+ * List per-bucket remote log manifest entries for a table or partition
scope.
+ *
+ * @param tableId the table to query
+ * @param partitionId optional partition id (null for non-partitioned
tables)
+ * @return per-bucket manifest paths and end offsets
+ */
+ @Internal
Review Comment:
This `@Internal` annotation seems wired — the `API` key is marked as
`PUBLIC`, but the method itself is `@Internal`. Shouldn't one of them be
changed to make them consistent?
##########
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
Review Comment:
Neither RPC verifies that the supplied `partitionId` actually belongs to the
supplied `tableId`:
- listRemoteLogManifests → ZooKeeperClient.listRemoteLogManifestHandles
reads directly from PartitionIdZNode.path(partitionId)/buckets and builds
TableBucket(tableId, partitionId, bucketId) using the request's tableId, not
the partition's actual owner in ZK.
- listKvSnapshots → resolves numBuckets via
zkClient.getPartitionAssignment(partitionId) without checking the partition's
owning table.
Impact: a client request (tableId=A, partitionId=B) where B actually belongs
to table C will silently succeed and return C's manifest / snapshot data
labelled with tableId=A in the response. Although this does not cause
cross-table deletion in the current cleanup flow (the enumerator never mixes
partitions across tables), it does open a permission-bypass path: authorization
only checks DESCRIBE on tableId=A, while the data actually returned belongs to
table C. Any user authorized on A who can guess a partition ID under C can read
C's metadata.
Suggested fix: in both handlers, fetch the partition's owning table id from
ZK and reject when it does not equal the request's tableId, treating the
mismatch as NOT_FOUND (do not leak existence of the partition under another
table). For example:
```
Optional<PartitionAssignment> pa =
zkClient.getPartitionAssignment(partitionId);
if (!pa.isPresent() || pa.get().getTableId() != tableId) {
throw new ApiException("Partition " + partitionId + " does not exist
under table " + tableId);
}
```
##########
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;
Review Comment:
ditto. RPC doesn't verifies that the supplied `partitionId` actually belongs
to the supplied `tableId`
##########
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) {
Review Comment:
The underlying `getChildren(bucketsPath)` call throws `NoNodeException`
whenever the path does not exist — e.g. an invalid tableId, a partition that is
being deleted, or a table whose buckets have never had any remote log
registered. All of these are currently swallowed by the catch (Exception)
branch and re-thrown as `UnknownServerException`, which the client side treats
as a non-retriable SERVER_ERROR.
Consequences:
- The client cannot distinguish "table/partition does not exist" from a
genuine server failure.
- Upstream `ActiveRefsFetcher` then applies its real-failure retry policy to
what is actually a NOT_FOUND, issuing several extra rounds of ZK reads — wasted
RPCs that also slow down Stage 1.
Suggested fix: explicitly map NoNodeException to `TableNotExistException` /
`PartitionNotExistException` (depending on whether partitionId == null) before
falling through to the generic `UnknownServerException ` wrapper, so the client
can short-circuit retries on a clear NOT_FOUND signal.
##########
fluss-flink/fluss-flink-action/src/main/java/org/apache/fluss/flink/action/FlussFlinkActionEntrypoint.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * 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;
+
+import java.util.Optional;
+
+/** Main entrypoint for the Fluss Flink action jar. Delegates to {@link
ActionLoader}. */
+public class FlussFlinkActionEntrypoint {
Review Comment:
Maybe we can named as Paimon, just call it `FlinkActions`? — it's shorter
and stays product-neutral. If a Fluss prefix is desirable, at least drop the
Entrypoint suffix; either `FlussActions` or `FlussFlinkActions `reads much
cleaner than the current `FlussFlinkActionEntrypoint`.
##########
fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java:
##########
@@ -1341,6 +1341,96 @@ public Optional<RemoteLogManifestHandle>
getRemoteLogManifestHandle(TableBucket
return getOrEmpty(path).map(BucketRemoteLogsZNode::decode);
}
+ /**
+ * Lists all remote log manifest handles for buckets of the given table
(or partition when
+ * {@code partitionId != null}). Reads the {@code BucketRemoteLogsZNode}
subtree under either
+ * {@code /tabletservers/tables/{tableId}/buckets/} or {@code
+ * /tabletservers/partitions/{partitionId}/buckets/} as a single ZK
getChildren call followed by
+ * per-bucket getData. Returns each bucket's currently-published manifest
path from coordinator
+ * metadata; callers must read the manifest file separately.
+ */
+ public List<TableBucketAndManifest> listRemoteLogManifestHandles(
+ long tableId, @Nullable Long partitionId) throws Exception {
+ String bucketsPath =
+ partitionId == null
+ ? TableIdZNode.path(tableId) + "/buckets"
+ : PartitionIdZNode.path(partitionId) + "/buckets";
+ List<String> bucketIdStrs = getChildren(bucketsPath);
+ List<TableBucketAndManifest> result = new
ArrayList<>(bucketIdStrs.size());
+ for (String bucketIdStr : bucketIdStrs) {
+ int bucketId = Integer.parseInt(bucketIdStr);
+ TableBucket tb =
+ partitionId == null
+ ? new TableBucket(tableId, bucketId)
+ : new TableBucket(tableId, partitionId, bucketId);
+ Optional<RemoteLogManifestHandle> handle =
getRemoteLogManifestHandle(tb);
+ handle.ifPresent(h -> result.add(new TableBucketAndManifest(tb,
h)));
+ }
+ return result;
+ }
+
+ /** Tuple of a table bucket and its current remote log manifest handle. */
+ public static final class TableBucketAndManifest {
+ private final TableBucket tableBucket;
+ private final RemoteLogManifestHandle manifestHandle;
+
+ public TableBucketAndManifest(
+ TableBucket tableBucket, RemoteLogManifestHandle
manifestHandle) {
+ this.tableBucket = tableBucket;
+ this.manifestHandle = manifestHandle;
+ }
+
+ public TableBucket getTableBucket() {
+ return tableBucket;
+ }
+
+ public RemoteLogManifestHandle getManifestHandle() {
+ return manifestHandle;
+ }
+ }
+
+ /**
+ * Lists all completed bucket snapshots for the given {@link TableBucket}.
Reads the {@code
+ * BucketSnapshotsZNode} subtree as a single ZK getChildren followed by
per-snapshot getData.
+ * Snapshot ids are returned ordered ascending.
+ */
+ public List<BucketSnapshotIdAndData> listBucketSnapshots(TableBucket
tableBucket)
+ throws Exception {
+ String path = BucketSnapshotsZNode.path(tableBucket);
+ List<String> snapshotIdStrs = getChildren(path);
+ List<BucketSnapshotIdAndData> result = new
ArrayList<>(snapshotIdStrs.size());
+ for (String snapshotIdStr : snapshotIdStrs) {
+ long snapshotId = Long.parseLong(snapshotIdStr);
+ String childPath = BucketSnapshotIdZNode.path(tableBucket,
snapshotId);
+ Optional<byte[]> data = getOrEmpty(childPath);
+ if (data.isPresent()) {
+ BucketSnapshot bs = BucketSnapshotIdZNode.decode(data.get());
+ result.add(new BucketSnapshotIdAndData(snapshotId, bs));
+ }
+ }
+ result.sort((a, b) -> Long.compare(a.snapshotId, b.snapshotId));
+ return result;
+ }
+
+ /** Tuple of (snapshotId, BucketSnapshot json data) for {@link
#listBucketSnapshots}. */
+ public static final class BucketSnapshotIdAndData {
+ private final long snapshotId;
+ private final BucketSnapshot bucketSnapshot;
+
+ public BucketSnapshotIdAndData(long snapshotId, BucketSnapshot
bucketSnapshot) {
+ this.snapshotId = snapshotId;
+ this.bucketSnapshot = bucketSnapshot;
+ }
+
+ public long getSnapshotId() {
+ return snapshotId;
+ }
+
+ public BucketSnapshot getBucketSnapshot() {
Review Comment:
This method is never used.
##########
fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java:
##########
@@ -367,6 +371,38 @@ public CompletableFuture<List<PartitionInfo>>
listPartitionInfos(
.thenApply(ClientRpcMessageUtils::toPartitionInfos);
}
+ /**
+ * Returns per-bucket remote log manifest path for the given table or
partition.
+ *
+ * <p>Used by the orphan cleanup action to construct the active manifest
path set without
+ * relying on FS LIST + mtime selection.
+ */
+ @Override
+ public CompletableFuture<ListRemoteLogManifestsResponse>
listRemoteLogManifests(
+ long tableId, @Nullable Long partitionId) {
+ ListRemoteLogManifestsRequest request = new
ListRemoteLogManifestsRequest();
+ request.setTableId(tableId);
+ if (partitionId != null) {
+ request.setPartitionId(partitionId);
+ }
+ return gateway.listRemoteLogManifests(request);
+ }
+
+ /**
+ * Returns per-bucket active KV snapshot dirs (retained_N + still-in-use)
for the given table or
+ * partition. Used by the orphan cleanup action to construct the complete
KV active set.
+ */
+ @Override
+ public CompletableFuture<ListKvSnapshotsResponse> listKvSnapshots(
Review Comment:
ditto
##########
fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java:
##########
@@ -770,4 +773,32 @@ CompletableFuture<RegisterResult> registerProducerOffsets(
* @since 0.9
*/
CompletableFuture<Void> deleteProducerOffsets(String producerId);
+
+ /**
+ * List per-bucket remote log manifest entries for a table or partition
scope.
+ *
+ * @param tableId the table to query
+ * @param partitionId optional partition id (null for non-partitioned
tables)
+ * @return per-bucket manifest paths and end offsets
+ */
+ @Internal
+ default CompletableFuture<ListRemoteLogManifestsResponse>
listRemoteLogManifests(
+ long tableId, @Nullable Long partitionId) {
Review Comment:
Both `listRemoteLogManifests` and `listKvSnapshots` are scoped to a single
(tableId, partitionId) target, so a partitioned table with P partitions forces
P sequential RPCs from the Stage-1 enumerator (parallelism=1) per table.
For tables with hundreds or thousands of partitions this concentrates a
large RPC burst on a single coordinator-side hot path, and Stage-1 becomes the
cleanup job's bottleneck.
##########
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);
+ int numBuckets;
+ if (tablePath != null) {
+ TableInfo tableInfo =
context.getTableInfoById(tableId);
+ if (tableInfo == null) {
+ throw new ApiException(
+ "Table metadata not yet available for
table " + tableId);
+ }
+ numBuckets = tableInfo.getNumBuckets();
+ } else {
+ try {
+ if (partitionId != null) {
+ Optional<PartitionAssignment> pAssignment =
+
zkClient.getPartitionAssignment(partitionId);
+ if (!pAssignment.isPresent()) {
+ throw new ApiException(
+ "Partition " + partitionId + "
does not exist");
+ }
+ numBuckets =
pAssignment.get().getBuckets().size();
+ } else {
+ Optional<TableAssignment> assignment =
+ zkClient.getTableAssignment(tableId);
+ if (!assignment.isPresent()) {
+ throw new ApiException("Table " + tableId
+ " does not exist");
+ }
+ numBuckets =
assignment.get().getBuckets().size();
+ }
+ } catch (ApiException e) {
+ throw e;
+ } catch (Exception e) {
Review Comment:
ditto
##########
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(
+ ListRemoteLogManifestsRequest request) {
Review Comment:
`AdminReadOnlyGateway` is the shared read-only admin contract implemented by
both CoordinatorService and TabletService (via RpcServiceBase). Every other
method on this interface (listDatabases, getTableInfo, metadata,
getFileSystemSecurityToken, …) is genuinely served by either side. These two
new methods break that invariant? Maybe u can move both methods out of
`AdminReadOnlyGateway` to a coordinator-scoped contract, so only the
coordinator gateway exposes them?
##########
fluss-flink/fluss-flink-1.20/src/test/java/org/apache/fluss/flink/action/orphan/Flink20OrphanFilesCleanITCase.java:
##########
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.action.orphan;
+
+/** The IT case for orphan files cleanup in Flink 1.20. */
+class Flink20OrphanFilesCleanITCase extends OrphanFilesCleanITCase {}
Review Comment:
ditto. `Flink120OrphanFilesCleanITCase`?
##########
fluss-flink/fluss-flink-1.19/src/test/java/org/apache/fluss/flink/action/orphan/Flink19OrphanFilesCleanITCase.java:
##########
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.action.orphan;
+
+/** The IT case for orphan files cleanup in Flink 1.19. */
+class Flink19OrphanFilesCleanITCase extends OrphanFilesCleanITCase {}
Review Comment:
Why does `Flink19OrphanFilesCleanITCase` not follow the same naming
convention as the other Flink-version IT cases (e.g.
`Flink119OrphanFilesCleanITCase`)?
##########
fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java:
##########
@@ -533,6 +533,17 @@ public static ReleaseKvSnapshotLeaseRequest
makeReleaseKvSnapshotLeaseRequest(
return request;
}
+ public static PbTableBucket toPbTableBucket(TableBucket tableBucket) {
Review Comment:
Can be removed as this methos is not used.
##########
fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java:
##########
@@ -1341,6 +1341,96 @@ public Optional<RemoteLogManifestHandle>
getRemoteLogManifestHandle(TableBucket
return getOrEmpty(path).map(BucketRemoteLogsZNode::decode);
}
+ /**
+ * Lists all remote log manifest handles for buckets of the given table
(or partition when
+ * {@code partitionId != null}). Reads the {@code BucketRemoteLogsZNode}
subtree under either
+ * {@code /tabletservers/tables/{tableId}/buckets/} or {@code
+ * /tabletservers/partitions/{partitionId}/buckets/} as a single ZK
getChildren call followed by
+ * per-bucket getData. Returns each bucket's currently-published manifest
path from coordinator
+ * metadata; callers must read the manifest file separately.
+ */
+ public List<TableBucketAndManifest> listRemoteLogManifestHandles(
+ long tableId, @Nullable Long partitionId) throws Exception {
+ String bucketsPath =
+ partitionId == null
+ ? TableIdZNode.path(tableId) + "/buckets"
+ : PartitionIdZNode.path(partitionId) + "/buckets";
+ List<String> bucketIdStrs = getChildren(bucketsPath);
+ List<TableBucketAndManifest> result = new
ArrayList<>(bucketIdStrs.size());
+ for (String bucketIdStr : bucketIdStrs) {
+ int bucketId = Integer.parseInt(bucketIdStr);
+ TableBucket tb =
+ partitionId == null
+ ? new TableBucket(tableId, bucketId)
+ : new TableBucket(tableId, partitionId, bucketId);
+ Optional<RemoteLogManifestHandle> handle =
getRemoteLogManifestHandle(tb);
+ handle.ifPresent(h -> result.add(new TableBucketAndManifest(tb,
h)));
+ }
+ return result;
+ }
+
+ /** Tuple of a table bucket and its current remote log manifest handle. */
+ public static final class TableBucketAndManifest {
+ private final TableBucket tableBucket;
+ private final RemoteLogManifestHandle manifestHandle;
+
+ public TableBucketAndManifest(
+ TableBucket tableBucket, RemoteLogManifestHandle
manifestHandle) {
+ this.tableBucket = tableBucket;
+ this.manifestHandle = manifestHandle;
+ }
+
+ public TableBucket getTableBucket() {
+ return tableBucket;
+ }
+
+ public RemoteLogManifestHandle getManifestHandle() {
+ return manifestHandle;
+ }
+ }
+
+ /**
+ * Lists all completed bucket snapshots for the given {@link TableBucket}.
Reads the {@code
+ * BucketSnapshotsZNode} subtree as a single ZK getChildren followed by
per-snapshot getData.
+ * Snapshot ids are returned ordered ascending.
+ */
+ public List<BucketSnapshotIdAndData> listBucketSnapshots(TableBucket
tableBucket)
+ throws Exception {
+ String path = BucketSnapshotsZNode.path(tableBucket);
+ List<String> snapshotIdStrs = getChildren(path);
+ List<BucketSnapshotIdAndData> result = new
ArrayList<>(snapshotIdStrs.size());
+ for (String snapshotIdStr : snapshotIdStrs) {
+ long snapshotId = Long.parseLong(snapshotIdStr);
+ String childPath = BucketSnapshotIdZNode.path(tableBucket,
snapshotId);
+ Optional<byte[]> data = getOrEmpty(childPath);
+ if (data.isPresent()) {
+ BucketSnapshot bs = BucketSnapshotIdZNode.decode(data.get());
+ result.add(new BucketSnapshotIdAndData(snapshotId, bs));
+ }
+ }
+ result.sort((a, b) -> Long.compare(a.snapshotId, b.snapshotId));
Review Comment:
Can be replaced with 'Comparator.comparingLong'
##########
fluss-flink/fluss-flink-action/pom.xml:
##########
@@ -0,0 +1,82 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-flink</artifactId>
+ <version>1.0-SNAPSHOT</version>
+ </parent>
+
+ <packaging>jar</packaging>
+
+ <artifactId>fluss-flink-action</artifactId>
+ <name>Fluss : Flink : Action</name>
+
+ <properties>
+ <flink.minor.version>1.20.3</flink.minor.version>
Review Comment:
Change to ${flink.version} ?
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/fs/SafeDeleter.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.fs;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.flink.action.orphan.audit.AuditLogger;
+import org.apache.fluss.flink.action.orphan.rule.Decision;
+import org.apache.fluss.flink.action.orphan.rule.RuleId;
+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 java.io.IOException;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+
+/**
+ * Sole entry point for filesystem deletion within the orphan cleanup package.
+ *
+ * <p>Only two operations are exposed:
+ *
+ * <ul>
+ * <li>{@link #deleteFile} - delete a single file (never recursive).
+ * <li>{@link #deleteEmptyDir} - delete a directory only if it is currently
empty.
+ * </ul>
+ *
+ * <p>By design there is no recursive-delete API; any caller that needs
deletion under {@code
+ * fluss-flink-common/.../action/orphan/} should go through this class. The
single-entry-point
+ * invariant is currently enforced only by convention — there is no Checkstyle
rule guarding it.
+ */
+@Internal
+public final class SafeDeleter {
+
+ private final FileSystem fs;
+ private final boolean dryRun;
+ private final AuditLogger audit;
+ private final RateLimiter rateLimiter;
+
+ public SafeDeleter(FileSystem fs, boolean dryRun, AuditLogger audit) {
+ this(fs, dryRun, audit, RateLimiter.create(100.0));
+ }
+
+ public SafeDeleter(FileSystem fs, boolean dryRun, AuditLogger audit,
RateLimiter rateLimiter) {
+ this.fs = fs;
+ this.dryRun = dryRun;
+ this.audit = audit;
+ this.rateLimiter = rateLimiter;
+ }
+
+ /**
+ * Delete a single file.
+ *
+ * @return {@code true} if the file was actually deleted (or recorded as
would-be-deleted under
+ * {@code dryRun}); {@code false} if {@link FileSystem#delete}
returned {@code false}
+ * (deletion silently failed — e.g. permissions, transient
remote-store error). Callers
+ * should track {@code false} returns as delete failures in their run
summary.
+ */
+ public boolean deleteFile(FsPath file, Decision decision, RuleId ruleId)
throws IOException {
Review Comment:
IOException from `fs.delete` propagates all the way up and fails the entire
batch job.
```
fs.delete(...) throws IOException
→ SafeDeleter.deleteFile / deleteEmptyDir (throws IOException, no
try/catch around fs.delete)
→ BucketCleaner.clean / walkAndCleanDir (throws IOException, no
try/catch around safeDeleter.deleteFile)
→ ScanAndCleanFunction.processBucketTask /
processOrphanDirTask (throws IOException)
→ ScanAndCleanFunction.processElement (throws Exception →
Flink runtime)
→ EmptyDirSweeper.sweep / sweepOne (Stage-3 driver, same
pattern)
```
Every layer just re-declares throws IOException — none of them actually
catches it. The OrphanFilesCleanJob runs in RuntimeExecutionMode.BATCH, so a
single uncaught IOException fails the subtask, and BATCH mode has no task-local
restart — the entire job is rescheduled from scratch.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/fs/SafeDeleter.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.fs;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.flink.action.orphan.audit.AuditLogger;
+import org.apache.fluss.flink.action.orphan.rule.Decision;
+import org.apache.fluss.flink.action.orphan.rule.RuleId;
+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 java.io.IOException;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+
+/**
+ * Sole entry point for filesystem deletion within the orphan cleanup package.
+ *
+ * <p>Only two operations are exposed:
+ *
+ * <ul>
+ * <li>{@link #deleteFile} - delete a single file (never recursive).
+ * <li>{@link #deleteEmptyDir} - delete a directory only if it is currently
empty.
+ * </ul>
+ *
+ * <p>By design there is no recursive-delete API; any caller that needs
deletion under {@code
+ * fluss-flink-common/.../action/orphan/} should go through this class. The
single-entry-point
+ * invariant is currently enforced only by convention — there is no Checkstyle
rule guarding it.
+ */
+@Internal
+public final class SafeDeleter {
+
+ private final FileSystem fs;
+ private final boolean dryRun;
+ private final AuditLogger audit;
+ private final RateLimiter rateLimiter;
+
+ public SafeDeleter(FileSystem fs, boolean dryRun, AuditLogger audit) {
+ this(fs, dryRun, audit, RateLimiter.create(100.0));
+ }
+
+ public SafeDeleter(FileSystem fs, boolean dryRun, AuditLogger audit,
RateLimiter rateLimiter) {
+ this.fs = fs;
+ this.dryRun = dryRun;
+ this.audit = audit;
+ this.rateLimiter = rateLimiter;
+ }
+
+ /**
+ * Delete a single file.
+ *
+ * @return {@code true} if the file was actually deleted (or recorded as
would-be-deleted under
+ * {@code dryRun}); {@code false} if {@link FileSystem#delete}
returned {@code false}
+ * (deletion silently failed — e.g. permissions, transient
remote-store error). Callers
+ * should track {@code false} returns as delete failures in their run
summary.
+ */
+ public boolean deleteFile(FsPath file, Decision decision, RuleId ruleId)
throws IOException {
+ checkArgument(
+ decision == Decision.DELETE,
+ "deleteFile must only be called for Decision.DELETE, got %s",
+ decision);
+ if (dryRun) {
+ audit.logWouldDelete(file, ruleId);
+ return true;
+ }
+ rateLimiter.acquire();
+ boolean ok = fs.delete(file, false);
+ audit.logDeleted(file, ruleId, ok);
+ return ok;
+ }
+
+ /**
+ * Delete a directory only if it is currently empty.
+ *
+ * @return {@code true} if the directory was actually deleted (or recorded
as would-be-deleted
+ * under {@code dryRun}); {@code false} if the directory was non-empty
/ unreadable, or if
+ * {@link FileSystem#delete} returned {@code false}. Callers should
not increment a "deleted
+ * directory" counter when this returns {@code false}.
+ */
+ public boolean deleteEmptyDir(FsPath dir) throws IOException {
Review Comment:
ditto. IOException from fs.delete propagates all the way up and fails the
entire batch job.
--
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]
