Ignore a deferred un-assign superseded by a later rebalance

[delthas]

On ERR__REVOKE_PARTITIONS the un-assign is deferred until in-flight work has drained, so offsets can be committed before the partitions are released. If the next rebalance grants those partitions back before the drain finishes, the deferred callback still fires and un-assigns them.

sequenceDiagram
    participant K as Kafka
    participant C as BackbeatConsumer
    K->>C: revoke [0-4]
    Note right of C: work in flight,<br/>un-assign deferred
    K->>C: assign [0-4]
    C->>C: assign() applied
    Note right of C: drain completes
    C->>K: unassign() — discards the new assignment
    Note right of C: owns partitions at the broker,<br/>no local assignment, no rebalance to recover

Loading

Membership has not changed, so nothing triggers another rebalance and the consumer is stuck until restarted, while looking healthy on every liveness signal. Observed in a CTST census run: idle for 2 h 35 min with 188 of 199 messages stranded.

Changes

Every rebalance takes the next _rebalanceId. A deferred un-assign carries the id it was created with and gives up as soon as that no longer matches — the partitions belong to a later rebalance by then, so there is nothing left for it to do.

sequenceDiagram
    participant K as Kafka
    participant C as BackbeatConsumer
    participant Z as ZooKeeper
    K->>C: revoke [0-4]
    Note right of C: id 7 — work in flight,<br/>un-assign deferred
    Note over C: drain completes,<br/>deferred un-assign wakes
    C->>C: guard 1 — still id 7, proceed
    C->>Z: publish offsets (asynchronous)
    K->>C: assign [0-4]
    Note right of C: id 8 supersedes 7
    Z-->>C: published
    C->>C: guard 2 — 7 is not 8, give up
    Note right of C: assignment from id 8 kept,<br/>counter records SUPERSEDED

Loading

Two guards rather than one because the offset publish between them is asynchronous: checking only on wake-up leaves exactly the window above. The first guard also protects the drain signals and the watchdog, which by then belong to the later rebalance.

• The drain watchdog is now cleared on every rebalance, not only on assignment. A revoke following another revoke used to orphan the first timer, which would later disconnect a consumer that was not stuck.
• An arbitrary rebalance error now synchronises the assignment. The id bump supersedes whatever revoke was pending and drops its watchdog, so nothing else would answer that callback — and librdkafka requires one.
• New SUPERSEDED value on the existing rebalance counter, so the race is visible in metrics. Nothing downstream enumerates these values.

Tests

Eight unit tests in tests/unit/backbeatConsumer.js, no broker needed (existing BackbeatConsumerMock). Each guard was verified by reverting it and confirming exactly one test fails.

Note

A superseded cycle does not emit 'unassign', so a shutdown racing an assignment still leaves close() waiting. That is pre-existing and owned by BB-833: emitting the event without having un-assigned would push close() into disconnect() while still holding partitions, which hangs.

Issue: BB-835

[bert-e]

Hello delthas,

My role is to assist you with the merge of this
pull request. Please type @bert-e help to get information
on this process, or consult the user documentation.

Available options

name	description	privileged	authored
/after_pull_request	Wait for the given pull request id to be merged before continuing with the current one.
/bypass_author_approval	Bypass the pull request author's approval	⭐
/bypass_build_status	Bypass the build and test status	⭐
/bypass_commit_size	Bypass the check on the size of the changeset TBA	⭐
/bypass_incompatible_branch	Bypass the check on the source branch prefix	⭐
/bypass_jira_check	Bypass the Jira issue check	⭐
/bypass_peer_approval	Bypass the pull request peers' approval	⭐
/bypass_leader_approval	Bypass the pull request leaders' approval	⭐
/approve	Instruct Bert-E that the author has approved the pull request.		✍️
/create_pull_requests	Allow the creation of integration pull requests.
/create_integration_branches	Allow the creation of integration branches.
/no_octopus	Prevent Wall-E from doing any octopus merge and use multiple consecutive merge instead
/unanimity	Change review acceptance criteria from one reviewer at least to all reviewers
/wait	Instruct Bert-E not to run until further notice.

Available commands

name	description	privileged
/help	Print Bert-E's manual in the pull request.
/status	Print Bert-E's current status in the pull request.
/clear	Remove all comments from Bert-E from the history TBA
/retry	Re-start a fresh build TBA
/build	Re-start a fresh build TBA
/force_reset	Delete integration branches & pull requests, and restart merge process from the beginning.
/reset	Try to remove integration branches unless there are commits on them which do not appear on the source branch.

Status report is not available.

[codecov]

Codecov Report

❌ Patch coverage is 90.00000% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 75.51%. Comparing base (56a3c42) to head (6c6acda).

Files with missing lines	Patch %	Lines
lib/BackbeatConsumer.js	90.00%	2 Missing ⚠️

Additional details and impacted files

Files with missing lines	Coverage Δ
lib/constants.js	100.00% <ø> (ø)
lib/BackbeatConsumer.js	93.80% <90.00%> (+0.05%)	⬆️

... and 2 files with indirect coverage changes

Components	Coverage Δ
Bucket Notification	80.27% <ø> (ø)
Core Library	81.30% <90.00%> (-0.54%)	⬇️
Ingestion	70.09% <ø> (ø)
Lifecycle	80.46% <ø> (ø)
Oplog Populator	85.83% <ø> (ø)
Replication	62.01% <ø> (ø)
Bucket Scanner	85.76% <ø> (ø)

@@                 Coverage Diff                 @@
##           development/9.5    #2818      +/-   ##
===================================================
- Coverage            75.72%   75.51%   -0.21%     
===================================================
  Files                  200      200              
  Lines                13926    13946      +20     
===================================================
- Hits                 10545    10531      -14     
- Misses                3371     3405      +34     
  Partials                10       10              

Flag	Coverage Δ
api:retry	9.09% <0.00%> (-0.02%)	⬇️
api:routes	8.85% <0.00%> (-0.02%)	⬇️
bucket-scanner	85.76% <ø> (ø)
ft_test:queuepopulator	9.18% <5.00%> (-1.85%)	⬇️
ingestion	12.24% <0.00%> (-0.02%)	⬇️
lib	8.85% <45.00%> (+0.05%)	⬆️
lifecycle	19.32% <45.00%> (+0.01%)	⬆️
notification	1.01% <0.00%> (-0.01%)	⬇️
oplogPopulator	0.13% <0.00%> (-0.01%)	⬇️
replication	18.87% <45.00%> (+0.03%)	⬆️
unit	55.24% <90.00%> (+0.30%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[bert-e]

Waiting for approval

The following approvals are needed before I can proceed with the merge:

• the author

• 2 peers

[bert-e]

Request integration branches

Waiting for integration branch creation to be requested by the user.

To request integration branches, please comment on this pull request with the following command:

/create_integration_branches

Alternatively, the /approve and /create_pull_requests commands will automatically
create the integration branches.

[francoisferrand]

AFAIK this cannot happen:

• while draining, we did not "reply" to kafka yet
• if we don't reply within the expected period (5 min I think), this consumer is kicked out of the group, and will not take part in the next rebalance: i.e. it will not be assigned any partition
• the consumer may rejoin the group later, when it calls consume(); but this cannot happen while draining.

In addition, I think we have an extra safety: if we fail to drain quickly enough, the process will sabotage itself to fail health check, so it is restarted.

[delthas]

Hm, there is really a run where we found exactly that apparently:

01:42:18.914  rdkafka.revoke   parts=[0,1,2,3,4]  queueLen=0 running=5 ledger=5
01:42:19.046  rdkafka.assign   parts=[0,1,2,3,4]
01:42:19.059  processing queue drained, un-assigning   queueLen=0 running=0 ledger=0

But I will try to do a large census again to see if I can reasonably reproduce it.

[delthas]

Running a census to see if I can reproduce this

[delthas]

Replaced by #2837