swuferhong commented on code in PR #3174:
URL: https://github.com/apache/fluss/pull/3174#discussion_r3377570385
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java:
##########
@@ -208,45 +224,44 @@ public void resumeDeletions() {
private void resumeTableDeletions() {
Set<Long> tablesToBeDeleted = new
HashSet<>(coordinatorContext.getTablesToBeDeleted());
- Set<Long> eligibleTableDeletion = new HashSet<>();
-
for (long tableId : tablesToBeDeleted) {
- // if all replicas are marked as deleted successfully, then table
deletion is done
if (coordinatorContext.areAllReplicasInState(
tableId, ReplicaState.ReplicaDeletionSuccessful)) {
completeDeleteTable(tableId);
LOG.info("Deletion of table with id {} successfully
completed.", tableId);
- }
- if (isEligibleForDeletion(tableId)) {
- eligibleTableDeletion.add(tableId);
- }
- }
- if (!eligibleTableDeletion.isEmpty()) {
- for (long tableId : eligibleTableDeletion) {
- onDeleteTable(tableId);
+ } else if (isEligibleForDeletion(tableId)) {
+ if (replicaCleanupManager != null) {
+ replicaCleanupManager.submitTableDropForResume(
+ tableId, () -> onDeleteTable(tableId));
+ } else {
+ onDeleteTable(tableId);
+ }
}
}
}
private void resumePartitionDeletions() {
Set<TablePartition> partitionsToDelete =
new HashSet<>(coordinatorContext.getPartitionsToBeDeleted());
- Set<TablePartition> eligiblePartitionDeletion = new HashSet<>();
-
for (TablePartition partition : partitionsToDelete) {
// if all replicas are marked as deleted successfully, then
partition deletion is done
if (coordinatorContext.areAllReplicasInState(
partition, ReplicaState.ReplicaDeletionSuccessful)) {
completeDeletePartition(partition);
LOG.info("Deletion of partition {} successfully completed.",
partition);
- }
- if (isEligibleForDeletion(partition)) {
- eligiblePartitionDeletion.add(partition);
- }
- }
- if (!eligiblePartitionDeletion.isEmpty()) {
- for (TablePartition partition : eligiblePartitionDeletion) {
- onDeletePartition(partition.getTableId(),
partition.getPartitionId());
+ } else if (isEligibleForDeletion(partition)) {
+ long tableId = partition.getTableId();
+ long partitionId = partition.getPartitionId();
+ if (replicaCleanupManager != null) {
+ String partitionName =
coordinatorContext.getPartitionName(partitionId);
+ replicaCleanupManager.submitPartitionDropForResume(
+ tableId,
+ partitionId,
+ partitionName,
+ () -> onDeletePartition(tableId, partitionId));
Review Comment:
There are two call sites:
- Startup (CoordinatorEventProcessor.startup() →
tableManager.resumeDeletions())
- Runtime (processDeleteReplicaResponseReceived →
tableManager.resumeDeletions())
Why the runtime path never triggers a redundant `onDeletePartition`:Inside
`resumePartitionDeletions()`, the guard `isEligibleForDeletion` requires:
```
isPartitionQueuedForDeletion(partition)
&& !isAnyReplicaInState(partition, ReplicaDeletionStarted)
```
During normal operation, after `processDropPartition` → `onDeletePartition`
fires, all replicas transition to ReplicaDeletionStarted. As deletion responses
come back:
- Some replicas move to DeletionSuccessful, but remaining ones are still in
DeletionStarted
- So isEligibleForDeletion = false → submitPartitionDropForResume is never
called
- Eventually all replicas reach DeletionSuccessful → takes the
completeDeletePartition branch directly
In other words, at runtime resumeDeletions() only ever hits the "all done →
complete" branch, never the "re-trigger" branch. So, the
`isEligibleForDeletion` guard ensures `onDeletePartition` is only re-invoked in
the restart recovery scenario (This case is what we want to throtte). During
normal runtime, it's a no-op branch that's never reached.
--
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]
