gyang94 commented on code in PR #3391:
URL: https://github.com/apache/fluss/pull/3391#discussion_r3349680919
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java:
##########
@@ -919,43 +959,93 @@ private void processDropPartition(DropPartitionEvent
dropPartitionEvent) {
dropTableInfo.getTablePath(), tableId,
tablePartition.getPartitionId());
}
+ /**
+ * Handles StopReplica deletion responses from the sender thread.
Partitions per-bucket results
+ * into succeeded/failed sets, then moves replicas to the appropriate
state.
+ *
+ * <p>Mirrors Kafka's {@code
processTopicDeletionStopReplicaResponseReceived}
+ * (KafkaController.scala:1451-1471). Fluss's StopReplicaResponse has no
top-level error code
+ * (FlussApi.proto:373-375), so the Kafka "requestError != NONE → all in
error" branch does not
+ * apply. Each bucket's error code is evaluated individually.
+ *
+ * <p>No {@code isActive} guard is needed (unlike Kafka :1454): {@code
+ * coordinatorEventManager.close()} in {@link #shutdown()} joins the event
thread before {@code
+ * coordinatorChannelManager.shutdown()}, so no event runs after
resignation.
+ */
private void processDeleteReplicaResponseReceived(
DeleteReplicaResponseReceivedEvent
deleteReplicaResponseReceivedEvent) {
List<DeleteReplicaResultForBucket> deleteReplicaResultForBuckets =
deleteReplicaResponseReceivedEvent.getDeleteReplicaResults();
- Set<TableBucketReplica> failDeletedReplicas = new HashSet<>();
- Set<TableBucketReplica> successDeletedReplicas = new HashSet<>();
- for (DeleteReplicaResultForBucket deleteReplicaResultForBucket :
- deleteReplicaResultForBuckets) {
- TableBucketReplica tableBucketReplica =
- deleteReplicaResultForBucket.getTableBucketReplica();
- if (deleteReplicaResultForBucket.succeeded()) {
- successDeletedReplicas.add(tableBucketReplica);
+ Set<TableBucketReplica> succeeded = new HashSet<>();
+ Set<TableBucketReplica> failed = new HashSet<>();
+ for (DeleteReplicaResultForBucket result :
deleteReplicaResultForBuckets) {
+ if (result.succeeded()) {
+ succeeded.add(result.getTableBucketReplica());
} else {
- failDeletedReplicas.add(tableBucketReplica);
+ failed.add(result.getTableBucketReplica());
}
}
- // clear the fail deleted number for the success deleted replicas
- coordinatorContext.clearFailDeleteNumbers(successDeletedReplicas);
- // pick up the replicas to retry delete and replicas that considered
as success delete
- Tuple2<Set<TableBucketReplica>, Set<TableBucketReplica>>
- retryDeleteAndSuccessDeleteReplicas =
-
coordinatorContext.retryDeleteAndSuccessDeleteReplicas(failDeletedReplicas);
+ if (!failed.isEmpty()) {
+ failReplicaDeletion(failed);
Review Comment:
if "TS stays alive, never reconnects", but always response with errors, the
retry-with-backoff may cost more than "ineligible mark" (comparing with
repeated StopReplica requests to a TS, and TS processed it and keeps returning
errors), and for permanent errors like disk failure, that retry never
terminates.
--
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]
