Guard the consumer subscription read against ERR__STATE - #2804
Conversation
Hello delthas,My role is to assist you with the merge of this Available options
Available commands
Status report is not available. |
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 #2804 +/- ##
===================================================
- Coverage 75.76% 75.47% -0.29%
===================================================
Files 200 200
Lines 13922 13926 +4
===================================================
- Hits 10548 10511 -37
- Misses 3364 3405 +41
Partials 10 10
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
Waiting for approvalThe following approvals are needed before I can proceed with the merge:
|
3950fba to
f6fee9c
Compare
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
3d0b402 to
4413584
Compare
node-rdkafka's subscription() throws `Local: Erroneous state` when the native layer reports itself not connected, which it does as soon as `m_is_closing` is set -- while the JS isConnected() flag stays true until the `disconnected` event lands. So a caller cannot rely on isConnected() alone. isPaused() read the subscription unguarded, and onEntryCommittable() evaluates isPaused() inside its `if` condition, outside the try/catch that was added to keep offsetsStore() from crashing the process. The condition meant to prevent ERR__STATE could therefore raise it. The throw escaped through _onEntryProcessingDone, which runs from the processing-queue completion callback, so finishProcessingTask() was skipped: the task slot leaked and _processingQueue.idle() could never become true. With no uncaughtException handler anywhere in the service, the exception then exited the process without leaving the consumer group, delaying the rebalance for surviving members. Read the subscription through a _getSubscription() helper that reports "no subscription" when the call fails, and route all four readers through it. isPaused() treats the consumer as paused and getServiceStatus() reports inactive, so callers skip storing offsets and the entry is re-delivered to whichever consumer holds the partition next. close() relied on a `|| []` fallback that could never run, and a throw there would have skipped the waterfall callback and hung shutdown; it now proceeds straight to disconnect. resume() subscribes when the subscription is unreadable rather than throwing, since leaving the consumer paused with nothing to retry it is worse. Reporting "not subscribed" is the conservative answer for every one of these readers. subscribe() is deliberately left unguarded: swallowing a failure there would leave a consumer that never subscribes, which isReady() does not detect (it checks only connectivity), turning a loud crash into a silent stall. Same log-don't-crash contract as BB-758 and BB-823. Issue: BB-845
4413584 to
a33d1f1
Compare
francoisferrand
left a comment
francoisferrand
left a comment
There was a problem hiding this comment.
- not sure if we should handle in 9.5 or just 9.6
- if we remove the librdkafka pin to 2.3 (and bump to 2.12), do we still need? It seems we may actually be able to to it after all (i.e. revert the revert)
- nothing wrong with this PR, but I wonder if we are not missing something: we are already guarding lot of code with
isConnectedcalls, so I wonder if the exception here is a real corner case or just a missing isConencted call (i.e. is it expected that the API throws, or are we just not respecting the contract)
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 |
|
Thanks — on the two technical points: Is this a missing It isn't:
So the API is expected to throw here; we were respecting the contract as documented by Does bumping librdkafka to 2.12 remove the need? No — the window is in node-rdkafka's JS wrapper state model, not in librdkafka, so it is version-independent. It does get rarer in practice. From the CI A/B (25 runs per arm, full lib suite), the |
|
Waiting untilethe 9.5 vs 9.6 discussion is settled. |
|
/wait |
4 participants
isPaused()readthis._consumer.subscription()unguarded, andonEntryCommittable()evaluatesisPaused()inside itsifcondition — outside thetry/catchthat protects the call it guards. node-rdkafka throwsLocal: Erroneous statefrom that read while the consumer is disconnecting, so the check meant to preventERR__STATEcan raise it itself.Both callers invoke
onEntryCommittable()from a producer delivery callback with notry/catch, so the throw is uncaught and the process exits. It also skipsfinishProcessingTask(), leaking the task slot so_processingQueue.idle()never becomes true.resume(),close()andgetServiceStatus()read the subscription the same way; inclose()the throw skipsnext(), soclose(cb)never fires and the pod is SIGKILLed after the grace period.Shutdown path only, no data loss — the in-flight entry is re-delivered. The unguarded read dates to 2018; BB-758 added the
isConnected()check and thetry/catchbut stopped one line short of it.Fix: the four readers go through a
_getSubscription()helper that reports no subscription when the read fails, which is the conservative answer for each caller.subscribe()is deliberately not guarded — swallowing there would leave a consumer that never subscribes whileisReady()keeps passing, trading a loud crash for a silent stall.This is not a CI-flakiness fix. That is caused by the librdkafka 2.3.0 pin and is handled separately under BB-849.
Tests: six unit tests, three of which fail with
Local: Erroneous stateagainst the unfixed code.Issue: BB-845