Guard the deferred un-assign against a superseding rebalance - #2837
Open
delthas wants to merge 2 commits into
Open
Guard the deferred un-assign against a superseding rebalance#2837delthas wants to merge 2 commits into
delthas wants to merge 2 commits into
Conversation
Contributor
Hello delthas,My role is to assist you with the merge of this Available options
Available commands
Status report is not available. |
Contributor
Request integration branchesWaiting for integration branch creation to be requested by the user. To request integration branches, please comment on this pull request with the following command: Alternatively, the |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files
... and 2 files with indirect coverage changes
@@ Coverage Diff @@
## development/9.5 #2837 +/- ##
===================================================
- Coverage 75.72% 75.51% -0.21%
===================================================
Files 200 200
Lines 13926 13934 +8
===================================================
- Hits 10545 10522 -23
- Misses 3371 3402 +31
Partials 10 10
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
On ERR__REVOKE_PARTITIONS the un-assign is deferred until in-flight work drains. librdkafka however answers a subscribed-topic metadata change with an immediate rejoin even while that revoke is still unanswered, so a fresh assignment can be granted and applied before the drain completes -- and the deferred un-assign then discarded it, leaving a live group member owning partitions at the broker with no local assignment, consuming nothing, with nothing left to trigger a rebalance. Disarm the pending drain triggers and watchdog on every rebalance event, so a superseded un-assign can no longer fire; the watchdog disarm also stops a second revoke from leaving an orphaned timer that would later disconnect a healthy consumer. The async offset publish can still be in flight when a newer rebalance lands, so the un-assign continuation also checks a rebalance counter before releasing the assignment. Issue: BB-835
Manufactures the interleaving deterministically against a real broker: a revoke deferred behind an in-flight task, then a partition count change observed by a group-updating metadata request, which makes librdkafka rejoin and apply a fresh assignment while the revoke is still unanswered. The deferred un-assign must then leave that assignment alone. Two variants differ only in which request observes the change: an application getMetadata() call, and librdkafka's own periodic topic.metadata.refresh.interval.ms refresh, which fires with no application metadata call at all. Both are red without the rebalance guard (the consumer ends with an empty local assignment, ready but consuming nothing) and green with it. Issue: BB-835
delthas
force-pushed
the
improvement/BB-835/superseded-unassign-guard
branch
from
September 2, 2026 09:27
b76f6d0 to
02a7eb7
Compare
delthas
marked this pull request as ready for review
September 2, 2026 09:30
francoisferrand
left a comment
francoisferrand
left a comment
Contributor
There was a problem hiding this comment.
I think we should point clearly that this actually breaks our rebalance handling: we would still be processing previous tasks, yet not be able to commit them...
Would be safer to fit these "unexpected" (and buggy) events in the same automaton, e.g. sending the same events eventually/etc ; and possibly fail like when our own timeout is reached...
| // pending: disarm its drain triggers and watchdog, and stamp | ||
| // this event so a continuation already in flight (the async | ||
| // offset publish below) can tell it has been superseded. | ||
| const rebalanceId = ++this._rebalanceId; |
Contributor
There was a problem hiding this comment.
what events can we get, is this condition not too strong?
- if we receive UNASSIGN while already unassigned, we can just ignore it (and continue). It's weird, but best to finish what was already going on
- if we receive ASSIGN, then we should abort indeed and print a message ; however if wonder if that aborting should not immediately call "doUnassign()" : so we resume partitions, unassign the customer (but not the newly assigned partitions...) and emit signal...
| }); | ||
| return; | ||
| } | ||
| this._resumePausedPartitions(); |
Contributor
There was a problem hiding this comment.
should we not resume paused partitions as well?
this may happen when breakbeat engages - if we don't do it everything is stuck...
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
A consumer that defers answering a partition revoke can lose an assignment it was granted in the meantime, and stall forever.
The bug
When a revoke arrives while tasks are in flight,
BackbeatConsumerdoes not callunassign()right away: it waits for the processing queue and the offset ledger to drain. While we owe librdkafka that answer, the rebalance protocol is effectively on hold — but a subscribed topic that looks changed in metadata is handled outside of it: librdkafka rejoins the group immediately, without waiting for our answer. The new generation can grant the partitions right back, and we apply them… then the drain completes, and the deferredunassign()throws that fresh assignment away.sequenceDiagram participant BB as BackbeatConsumer participant RK as librdkafka participant B as broker RK->>BB: revoke [p0..p4] Note over BB: tasks in flight →<br/>defer unassign() until drained Note over RK: waiting for unassign()… B-->>RK: metadata: subscribed topic looks changed Note over RK: …but a topic change makes librdkafka<br/>rejoin now, revoke still unanswered RK->>B: JoinGroup + SyncGroup B-->>RK: new generation, same partitions RK->>BB: assign [p0..p4] BB->>RK: assign() — consuming again Note over BB: drain completes BB->>RK: unassign() — discards the fresh assignment Note over BB,B: broker: member still owns p0..p4 · local: nothing assigned<br/>membership unchanged → no rebalance will ever fix itWhat makes a topic "look changed": librdkafka compares consecutive snapshots of its local metadata cache, restricted to the subscription, on topic name, partition count and (when
client.rackis set) replica racks. Anything that makes two snapshots differ counts:client.rack(follower fetching);UNKNOWN_TOPIC_OR_PARTfor an existing topic plants a short-lived negative cache entry, so the topic "vanishes" for one observation and "reappears" at the next — two spurious rebalances with no actual change to the topic;TOPIC_AUTHORIZATION_FAILEDflaps (ACL churn), same mechanism with a longer-lived negative entry;Leader changes, ISR changes, and refreshes that return the same data do not trigger it.
The end state is silent and permanent: the consumer is connected, a group member, owns every partition at the broker and passes every liveness check — and consumes nothing. Observed in a CTST run as a GC pod idling 2h35m with 188 of 199 messages stranded.
The fix
Only the newest rebalance may act:
Reproduction
The new functional test manufactures the interleaving deterministically against a real broker: one held task as the in-flight work, a partition-count bump as the topic change. It is red on
development/9.5and green with this fix — in two variants: the change observed by an applicationgetMetadata()call, and by librdkafka's own periodictopic.metadata.refresh.interval.msrefresh, i.e. with no application metadata call involved at all.Versions
On librdkafka 2.3.0 the trigger fired constantly: every full metadata refresh wiped the metadata cache, so even a completely static topic read as "vanished" then "reappeared" on each refresh (fixed upstream in 2.4.0 by confluentinc/librdkafka#4677). The rejoin-while-a-revoke-is-unanswered behaviour itself is unchanged through 2.12.0, where the test reproduces the stall with a genuine topic change.
Replaces #2818.
Issue: BB-835