[
https://issues.apache.org/jira/browse/KAFKA-20860?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18101081#comment-18101081
]
Yunseop Eom commented on KAFKA-20860:
-------------------------------------
PR opened: https://github.com/apache/kafka/pull/23023
Fixed consumer reconciliation recovery after a synchronous setup failure by
resetting the in-progress latch and allowing the next background loop to retry.
Added regression coverage for failures during revocation and assignment startup
while preserving existing auto-commit failure behavior.
Validation: targeted membership-manager tests and full :clients:check passed.
Status: PR open and awaiting review.
> 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
> Priority: Critical
>
> 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 loss of a group member. Line numbers are against trunk
> 0d0ec5af1e.
> h2. 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, and every later maybeReconcile() early-returns
> at the gate.
> In AbstractMembershipManager:
> * \{{:885}} the gate -- if (reconciliationInProgress) then 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}} reached
> from \{{:1160}} (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 there 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.
> h3. Failure mode
> The member does not die:
> * state stays RECONCILING; every later maybeReconcile() early-returns at the
> gate;
> * heartbeats continue (the heartbeat manager is a separate RequestManager),
> so no session timeout;
> * *the latch survives fencing and rejoin.* transitionToJoining() does NOT
> clear
> reconciliationInProgress; it sets rejoinedWhileReconciliationInProgress = true
> (AbstractMembershipManager:581-582, StreamsMembershipManager:457-459). That
> flag is only ever
> consumed by maybeAbortReconciliation(), i.e. on the async completion path
> that -- by the premise
> of this bug -- never runs. So the member is wedged for the lifetime of the
> membership-manager
> instance: it can be fenced, rejoin, be assigned, and still never reconcile.
> * the member keeps serving its old assignment and never re-echoes its owned
> partitions/tasks, so
> the broker never receives the ack it is waiting for.
> Whether the broker intervenes depends on the shape of the new assignment,
> because the rebalance
> timeout is armed only while the member is broker-side UNREVOKED_TASKS and
> cancelled otherwise
> (GroupMetadataManager:4019-4028, timeout action at :5287-5305):
> * *new assignment revokes something* -> timeout armed -> member fenced after
> rebalanceTimeoutMs
> (for Streams that is max.poll.interval.ms, default 5 min), releasing its
> tasks so the other
> members unblock. But the latch survives the rejoin, so this becomes a
> permanent
> fence -> rejoin -> never-reconcile -> fence cycle, once per
> rebalanceTimeoutMs, indefinitely.
> * *new assignment only adds* -> no ack is required, so no timeout is armed
> and the broker considers
> the member reconciled -> no fencing at all, indefinitely, while the client
> never starts those
> partitions/tasks. Silent under-processing: the broker believes the member
> owns partitions it is
> not consuming, and lag grows without bound.
> Either way, one swallowed exception permanently removes that client from
> useful work, and in the
> first case repeatedly stalls the group's migration for a rebalance-timeout
> window at a time.
> h3. Long-standing
> The same latch shape exists in MembershipManagerImpl at 3.7.0
> (markReconciliationInProgress
> {\{:808}}, sole markReconciliationCompleted call \{{:911}}) and 3.8.0
> (\{{:943}} vs
> {\{:1054}}/\{{:1064}}/\{{:1083}}), and in AbstractMembershipManager at 4.1.0
> (\{{:854}} vs
> {\{:964}}/\{{:973}}/\{{:992}}). It therefore dates to the async consumer's
> first release.
> h3. 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:1363), and an unsubscribe() during the
> commit window is
> caught by maybeAbortReconciliation() because it changes state. It IS
> reachable for streams groups
> -- 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
> permanent, silent loss of a group member.
> h3. Fix
> Wrap the synchronous body from markReconciliationInProgress() to the point
> the future chain is
> established in try/catch (RuntimeException); on throw, log at ERROR and
> markReconciliationCompleted() so the member retries on the next poll instead
> of freezing. Same for
> the StreamsMembershipManager override and for the revokeAndAssign /
> revokeActiveTasks prologues.
> Additionally, clear the latch in transitionToJoining() so that a rejoin is a
> genuine reset.
> Note on scope: this restores containment, it does not make an unexecutable
> assignment executable.
> For a transient cause the next poll recovers, which is the value. For a
> deterministic cause the
> throw recurs every background-thread iteration (state still RECONCILING,
> target still not equal to
> current, gate now open), giving a loud retry loop rather than a silent
> freeze. Deciding what to do
> about input a client can never execute is a separate policy question, handled
> in the linked issue.
> h2. 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 and no counter.
> Blast radius, counting .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 the API key / request header for attribution. One
> place, covers all eleven
> managers; highest value-per-line in this ticket.
> h2. 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.
> In combination with item 1 this is a one-shot, because the latch suppresses
> the repeat (which is
> why the streams symptom is a wedged member rather than a hot spin). A throw
> that does not latch
> 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)
