Matthias J. Sax created KAFKA-20860:
---------------------------------------
Summary: Broken error handling in the async consumer background
thread
Key: KAFKA-20860
URL: https://issues.apache.org/jira/browse/KAFKA-20860
Project: Kafka
Issue Type: Bug
Components: clients, consumer
Affects Versions: 3.7.0
Reporter: Matthias J. Sax
Report generate by Claude:
The async consumer's background thread has three error-containment gaps in
machinery shared by consumer, share and streams groups. They compound: (2)
hides (1), and (1)'s failure mode is a silent, permanent, group-wide rebalance
stall.
h3. 1. reconciliationInProgress is a latch that a synchronous throw never
releases
markReconciliationInProgress() runs before synchronous work;
markReconciliationCompleted() is only ever called from the asynchronous
completion callbacks. Any exception escaping the synchronous portion leaves the
flag true forever, and every later maybeReconcile() early-returns at the gate.
In AbstractMembershipManager:
* \{{:885}} the gate -- \{{if (reconciliationInProgress) { ... return; }}}
* \{{:931}} markReconciliationInProgress()
* \{{:949}} markPendingRevocationToPauseFetching(...) ->
subscriptions.markPendingRevocation ->
assignedState(tp), which throws IllegalStateException when the partition is
absent
(SubscriptionState.java:426-431)
* \{{:953}} signalReconciliationStarted() -- synchronous call returning a future
* \{{:956}} commitResult.whenComplete(...) -- the only route to the three
markReconciliationCompleted() sites (\{{:1022}}, \{{:1031}}, \{{:1050}}), all
asynchronous
StreamsMembershipManager has the same shape with a wider synchronous window:
markReconciliationInProgress() at \{{:1117}}, then an invariant
IllegalStateException at \{{:1154}} and an unchecked subtopologies().get(...)
at \{{:1292}}/\{{:1307}} (NPE), all before the first future is built at
\{{:1164}}.
There is a second window in both: revokeAndAssign(...) (consumer \{{:968}}) and
revokeActiveTasks(...) are invoked inside a whenComplete lambda but before the
inner future chain exists, so a throw is captured by that lambda's dependent
stage instead of reaching the error branch that clears the flag. This window is
wide in wall-clock terms -- it opens after the auto-commit-before-rebalance
round trip, i.e. many runOnce iterations later.
Failure mode -- the member neither dies nor gets fenced:
* state stays RECONCILING and reconciliation never runs again;
* heartbeats continue at the current epoch (separate RequestManager), so no
session timeout, no FENCED_MEMBER_EPOCH, no rejoin, ever;
* the member keeps serving its old assignment and never re-echoes its owned
partitions/tasks;
* the broker therefore never receives the ack it is waiting for,
partitions/tasks pending revocation are never released, and OTHER members
stall in UNRELEASED_PARTITIONS/UNRELEASED_TASKS.
So one client's swallowed exception freezes the entire group's rebalance
indefinitely.
Long-standing: the same latch shape exists in MembershipManagerImpl at 3.7.0
(markReconciliationInProgress \{{:808}}, sole markReconciliationCompleted
\{{:911}}) and 3.8.0 (\{{:943}} vs \{{:1054}}/\{{:1064}}/\{{:1083}}), and in
AbstractMembershipManager at 4.1.0 (\{{:854}} vs \{{:964}}/\{{:973}}/\{{:992}}).
Reachability, stated honestly: I found no currently reachable synchronous throw
for consumer or share groups. Within a single runOnce() the background thread
is the only mutator of SubscriptionState (the app thread only reads
pausedPartitions(), AsyncKafkaConsumer.java:1363), and an unsubscribe() during
the commit window is caught by maybeAbortReconciliation() because it changes
state. It IS reachable for streams groups (KIP-1071) -- see the linked issue.
So this is a live bug for one group type and a defensive fix for the others, in
shared code, where the cost
asymmetry is stark: a few lines of try/catch against a silent permanent
group-wide stall.
Fix: wrap the synchronous body from markReconciliationInProgress() to the point
the future chain is established in try/catch; on throw, log at ERROR and
markReconciliationCompleted() so the member retries on the next poll rather
than freezing. Same for the StreamsMembershipManager override and for the
revokeAndAssign / revokeActiveTasks prologues.
h3. 2. Every response-handling callback silently swallows exceptions
{code:java}
// NetworkClientDelegate.java:385-388
UnsentRequest whenComplete(BiConsumer<ClientResponse, Throwable> callback) {
handler.future().whenComplete(callback); // dependent stage discarded
return this;
}
{code}
FutureCompletionHandler.onComplete completes the source future with
future.complete(response) (\{{:451}}). Per the CompletableFuture contract, an
exception thrown by a whenComplete action is captured into the DEPENDENT stage
and does not propagate to complete(); the source future and the
completing thread are unaffected. That dependent stage is discarded here, and
nothing anywhere observes it -- this is the only call site in the module and
there is no isCompletedExceptionally() check. The exception is lost with no
log, no metric, no counter.
Blast radius (count of .whenComplete( handlers riding this seam per file):
OffsetsRequestManager 12, AbstractMembershipManager 9, StreamsMembershipManager
8, CommitRequestManager 8, StreamsGroupHeartbeatRequestManager 2,
ShareConsumeRequestManager 2, FetchRequestManager 2,
AbstractHeartbeatRequestManager 2, CoordinatorRequestManager 1,
TopicMetadataRequestManager 1, StreamsGroupTopologyDescriptionRequestManager 1.
Present in 3.7.0 through trunk (verbatim at :385-388 in 4.2 and 4.3); the
current expression was introduced by c81a725219 (KAFKA-15534, 2023-10-20).
This is why item 1 is undiagnosable, and it silently breaks forward
compatibility on its own. A concrete example already in the tree:
StreamsGroupHeartbeatResponse.Status.fromCode (\{{:114-120}}) throws
IllegalArgumentException on an unknown status code, and
StreamsMembershipManager.isGroupReady (\{{:802}}) decodes every status on every
successful heartbeat. The member epoch has already been updated (\{{:763}})
when it throws, so the client keeps heartbeating healthily while silently
ignoring every new assignment -- with zero log output attributable to the
cause. Any future broker release that adds a heartbeat status code does this to
every existing client.
Fix: make the seam non-silent. In UnsentRequest.whenComplete, observe the
dependent stage and log callback failures (with API key / request header for
attribution). One place, covers all eleven managers; highest value-per-line in
this ticket.
h3. 3. A throwing RequestManager.poll() skips the network pump for that
iteration
ConsumerNetworkThread.runOnce() (\{{:210-240}}) runs the manager loop
(\{{:222-226}}) before networkClientDelegate.poll(...) (\{{:228}}), the
maximumTimeToWait loop (\{{:232-235}}) and reapExpiredApplicationEvents
(\{{:239}}). An exception from one manager's poll() aborts all remaining
managers, the network pump and event reaping for that iteration; the catch-all
in run() (\{{:163-166}}) logs "Unexpected error caught in consumer network
thread" and loops. A repeating
per-iteration throw in any single manager therefore stops all client I/O -- no
heartbeats sent, no responses processed, no fetches -- while the thread spins,
even though the other managers are healthy.
With item 1 this is a one-shot, because the latch suppresses the repeat (which
is why the streams symptom is a zombie rather than a hot spin). A non-latching
repeating throw gives the spin.
Fix: isolate per-manager failures in the runOnce loops (log and continue), so
one manager's defect degrades that manager rather than the whole client.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
