You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "swuferhong (via GitHub)" <gi...@apache.org> on 2026/09/29 03:55:52 UTC
[PR] [server] Resume pending replica deletion after server restart [fluss]
swuferhong opened a new pull request, #4505:
URL: https://github.com/apache/fluss/pull/4505
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink], [rust], [python], [c++], [elixir]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: [Tool Name and Version] following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #4504
<!-- What is the purpose of the change -->
During a rolling TabletServer upgrade, a table or partition may be dropped while one of its replicas is offline. The corresponding `StopReplica(delete=true)` request cannot complete, and the deletion may remain unfinished after the TabletServer restarts.
This can leave stale assignments in ZooKeeper and prevent cluster health from converging.
This PR introduces replica deletion recovery so unfinished deletion is regenerated from the Coordinator state when the TabletServer comes back. It does not rely on a request queued for the previous TabletServer process.
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Resume pending replica deletion after server restart [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4505:
URL: https://github.com/apache/fluss/pull/4505#discussion_r4257086844
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java:
##########
@@ -209,22 +210,87 @@ public void onDeletePartition(long tableId, long partitionId) {
*
* <p>It does the following:
*
- * <p>1. Move all the replicas to offline state. This will send stop replica request to the
- * replicas.
+ * <p>1. Move all replicas to offline state. For live or shutting-down tablet servers, this
+ * sends stop replica requests so they stop fetching from their leaders.
*
- * <p>2. Move all the replicas to deletion started state. This will send stop replica request
- * with delete=true which will delete all persistent data from all the replicas of the all the
- * respective buckets.
+ * <p>2. Move replicas on live or shutting-down tablet servers to deletion started state and
+ * send stop replica requests with delete=true. Replicas on dead tablet servers are moved to
+ * deletion ineligible state and retried when their tablet servers come back.
*/
private void onDeleteTableBucket(
Set<TableBucket> tableBuckets, Set<TableBucketReplica> allReplicas) {
coordinatorContext.removeKvBuckets(tableBuckets);
updateObservedKvLeaderReplicaCount();
- // to offline, send stop replica to all followers that are not in the OfflineReplica state
- // so they stop sending fetch requests to the leader
+ Map<Boolean, Set<TableBucketReplica>> replicasByTabletServerLiveness =
+ allReplicas.stream()
+ .collect(
+ Collectors.partitioningBy(
+ replica ->
+ coordinatorContext
+ .liveOrShuttingDownTabletServers()
+ .contains(replica.getReplica()),
+ Collectors.toSet()));
+
+ Set<TableBucketReplica> replicasOnLiveServers = replicasByTabletServerLiveness.get(true);
+ Set<TableBucketReplica> replicasOnDeadServers = replicasByTabletServerLiveness.get(false);
+ // Replica state alone does not indicate whether a delete request can be sent: an
+ // OfflineReplica may still be hosted by a live tablet server. Only replicas on dead tablet
+ // servers have no active channel and must wait for the server to come back.
+ // A dead server may still have OnlineReplica entries, for example when it goes down
+ // while a recovered deletion waits in the lifecycle throttler. Move all replicas through
+ // OfflineReplica before starting deletion or marking it ineligible.
replicaStateMachine.handleStateChanges(allReplicas, ReplicaState.OfflineReplica);
- // to deletion started
- replicaStateMachine.handleStateChanges(allReplicas, ReplicaState.ReplicaDeletionStarted);
+ replicaStateMachine.handleStateChanges(
+ replicasOnDeadServers, ReplicaState.ReplicaDeletionIneligible);
+ replicaStateMachine.handleStateChanges(
+ replicasOnLiveServers, ReplicaState.ReplicaDeletionStarted);
+ }
+
+ /** Marks pending replica deletions on an offline tablet server as ineligible. */
+ public void failReplicaDeletion(int tabletServerId) {
+ Set<TableBucketReplica> replicas =
+ coordinatorContext.replicasOnTabletServer(tabletServerId).stream()
+ .filter(
+ replica ->
+ coordinatorContext.isToBeDeleted(replica.getTableBucket()))
+ .filter(
+ replica -> {
+ ReplicaState state =
+ coordinatorContext.getReplicaState(replica);
+ return state == ReplicaState.OfflineReplica
+ || state == ReplicaState.ReplicaDeletionStarted;
+ })
+ .collect(Collectors.toSet());
+ if (!replicas.isEmpty()) {
+ LOG.info(
+ "Marking deletion of replicas {} ineligible because tablet server {} is offline.",
+ replicas,
+ tabletServerId);
+ replicaStateMachine.handleStateChanges(
+ replicas, ReplicaState.ReplicaDeletionIneligible);
+ }
+ }
+
+ /** Retries ineligible replica deletions after a tablet server comes back. */
+ public void resumeReplicaDeletion(int tabletServerId) {
+ Set<TableBucketReplica> replicas =
+ coordinatorContext.replicasOnTabletServer(tabletServerId).stream()
+ .filter(
+ replica ->
+ coordinatorContext.isToBeDeleted(replica.getTableBucket()))
+ .filter(
+ replica ->
+ coordinatorContext.getReplicaState(replica)
+ == ReplicaState.ReplicaDeletionIneligible)
+ .collect(Collectors.toSet());
Review Comment:
```suggestion
Set<TableBucketReplica> ineligibleReplicas = new HashSet<>();
Set<TableBucketReplica> replicasToDelete = new HashSet<>();
// Candidates must belong to deletion tasks that have already been admitted.
for (TableBucketReplica replica : deletionCandidates) {
if (!coordinatorContext.isToBeDeleted(replica.getTableBucket())) {
continue;
}
ReplicaState state = coordinatorContext.getReplicaState(replica);
if (state == ReplicaState.ReplicaDeletionIneligible) {
ineligibleReplicas.add(replica);
replicasToDelete.add(replica);
} else if (state == ReplicaState.OfflineReplica) {
replicasToDelete.add(replica);
}
}
// Only ineligible replicas need the intermediate Offline transition.
if (!ineligibleReplicas.isEmpty()) {
replicaStateMachine.handleStateChanges(
ineligibleReplicas, ReplicaState.OfflineReplica);
}
// Both groups can now enter ReplicaDeletionStarted.
if (!replicasToDelete.isEmpty()) {
replicaStateMachine.handleStateChanges(
replicasToDelete, ReplicaState.ReplicaDeletionStarted);
}
```
[P2] Handle pending deletions moved back to OfflineReplica by late responses
A late failure response from an earlier NotifyLeaderAndIsr request can move a pending-deletion replica from ReplicaDeletionIneligible back to OfflineReplica. When its tablet server re-registers, resumeReplicaDeletion() skips it, while the normal startup path also excludes it because its table/partition is pending deletion. Consequently, no new delete request is sent.
Could we prevent these late responses from overwriting deletion state, or recover already-admitted deletions whose replicas are now OfflineReplica?
For the latter approach, we can separate candidate selection from state transitions. Once the candidates are restricted to deletion tasks that have already been admitted, the transition logic could look like this:
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Resume pending replica deletion after server restart [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on PR #4505:
URL: https://github.com/apache/fluss/pull/4505#issuecomment-6158891487
@loserwang1024 comments addressed. Could you take another look? Thanks.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Resume pending replica deletion after server restart [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on PR #4505:
URL: https://github.com/apache/fluss/pull/4505#issuecomment-6123782400
@platinumhamburg comments addressed. Could you take another look? Thanks.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Resume pending replica deletion after server restart [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #4505:
URL: https://github.com/apache/fluss/pull/4505#discussion_r4245933414
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java:
##########
@@ -209,22 +210,84 @@ public void onDeletePartition(long tableId, long partitionId) {
*
* <p>It does the following:
*
- * <p>1. Move all the replicas to offline state. This will send stop replica request to the
- * replicas.
+ * <p>1. Move replicas on live tablet servers to offline state. This sends stop replica requests
+ * so they stop fetching from their leaders.
*
- * <p>2. Move all the replicas to deletion started state. This will send stop replica request
- * with delete=true which will delete all persistent data from all the replicas of the all the
- * respective buckets.
+ * <p>2. Move those replicas to deletion started state and send stop replica requests with
+ * delete=true. Replicas on dead tablet servers are moved to deletion ineligible state and
+ * retried when their tablet servers come back.
*/
private void onDeleteTableBucket(
Set<TableBucket> tableBuckets, Set<TableBucketReplica> allReplicas) {
coordinatorContext.removeKvBuckets(tableBuckets);
updateObservedKvLeaderReplicaCount();
- // to offline, send stop replica to all followers that are not in the OfflineReplica state
- // so they stop sending fetch requests to the leader
- replicaStateMachine.handleStateChanges(allReplicas, ReplicaState.OfflineReplica);
- // to deletion started
- replicaStateMachine.handleStateChanges(allReplicas, ReplicaState.ReplicaDeletionStarted);
+ Map<Boolean, Set<TableBucketReplica>> replicasByTabletServerLiveness =
+ allReplicas.stream()
+ .collect(
+ Collectors.partitioningBy(
+ replica ->
+ coordinatorContext
+ .liveTabletServerSet()
+ .contains(replica.getReplica()),
+ Collectors.toSet()));
+
+ Set<TableBucketReplica> replicasOnLiveServers = replicasByTabletServerLiveness.get(true);
+ Set<TableBucketReplica> replicasOnDeadServers = replicasByTabletServerLiveness.get(false);
+ // Replica state alone does not indicate whether a delete request can be sent: an
+ // OfflineReplica may still be hosted by a live tablet server. Only replicas on dead tablet
+ // servers have no active channel and must wait for the server to come back.
+ replicaStateMachine.handleStateChanges(replicasOnLiveServers, ReplicaState.OfflineReplica);
+ replicaStateMachine.handleStateChanges(
Review Comment:
Could we retain the transition through `OfflineReplica` for replicas on unavailable servers before moving them to `ReplicaDeletionIneligible`? Being on an unavailable server does not guarantee that the replica state is already `OfflineReplica`.
For example, `liveTabletServerSet()` excludes servers undergoing controlled shutdown, but a former leader replica can still be `OnlineReplica` after leadership moves. Dropping its table at that point attempts `OnlineReplica -> ReplicaDeletionIneligible`, which is rejected. The subsequent server-down handler does not recover this replica either: `failReplicaDeletion()` only handles Offline/DeletionStarted, while normal offline handling skips replicas pending deletion. When the server returns, `resumeReplicaDeletion()` only selects Ineligible replicas, so no delete request is rebuilt and the assignment can remain.
The same invalid transition is possible after Coordinator recovery: a pending deletion waits in the lifecycle throttler with an Online replica, and its server goes down before the deletion is admitted. Using `liveOrShuttingDownTabletServers()` would fix the shutdown classification, but would not cover this second case.
Please preserve the legal transition through Offline for these replicas and use the server set that includes shutting-down servers when deciding whether to send deletion requests. Regression coverage should include both interleavings and verify that the assignment is eventually removed after the server returns.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Resume pending replica deletion after server restart [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong merged PR #4505:
URL: https://github.com/apache/fluss/pull/4505
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org