You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "KAI-JAY TU (Jira)" <ji...@apache.org> on 2026/07/23 11:06:00 UTC
[jira] (KAFKA-20819) StreamThread hangs forever in TaskManager.waitForFuture() after its StateUpdater thread dies, silently dropping out of the consumer group
[ https://issues.apache.org/jira/browse/KAFKA-20819 ]
KAI-JAY TU deleted comment on KAFKA-20819:
------------------------------------
was (Author: JIRAUSER314047):
Hi, I'd like to take a crack at this one.
I spent some time tracing through the code and the production evidence in the ticket, and here's what I found:
*What's going wrong:*
[StateUpdaterThread.run()|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html] only catches {{{}RuntimeException{}}}. When something like {{OutOfMemoryError}} hits (as the reporter confirmed), it blows straight through to the {{finally}} block. The thread logs "State updater thread stopped" and cleans up queues, but nobody ever completes or fails the pending [CompletableFuture|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html]s sitting in those queues.
On the other side, [DefaultStateUpdater.remove()|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html] happily enqueues new work with a fresh [CompletableFuture|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html] — it never checks whether the thread is actually alive. So when [TaskManager.waitForFuture()|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html] calls [future.get(5, MINUTES)|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html] on trunk (or the unbounded [future.get()|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html] on 3.9.x), it's waiting on a future that will never complete.
*How I'd fix it (3 parts):*
# Widen the catch in [StateUpdaterThread.run()|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html] to {{{}Throwable{}}}, so {{{}Error{}}}s get properly logged and [isRunning|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html] gets set to false. On thread exit, fail all pending futures in the queue so nobody blocks on them.
# In [remove()|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html] and [add()|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html], check [isRunning()|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html] before enqueuing. If the thread is already dead, complete the future exceptionally right away instead of queueing work that'll never get processed.
# In [TaskManager.waitForFuture()|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html], don't just warn and return null on timeout — check whether the state updater thread is actually dead, and if so, log at ERROR with a clear message.
Before I go further — are you already working on a fix or planning to? Don't want to duplicate effort. If not, I'm happy to pick this up. I've walked through [DefaultStateUpdater|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html], [TaskManager|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html], and the rebalance listener chain, and I can write the fix with tests that simulate a dead [StateUpdaterThread|vscode-file://vscode-app/c:/Users/markove.tu/AppData/Local/Programs/Microsoft%20VS%20Code/fc3def6774/resources/app/out/vs/code/electron-browser/workbench/workbench.html]. Happy to put up a draft PR early for feedback.
One question: if the state updater dies, should we try to restart it, or is failing fast (letting the StreamThread shut down) the right call? I'd lean toward fail-fast to keep things simple, but want to hear your take.
> StreamThread hangs forever in TaskManager.waitForFuture() after its StateUpdater thread dies, silently dropping out of the consumer group
> -----------------------------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-20819
> URL: https://issues.apache.org/jira/browse/KAFKA-20819
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 3.9.2
> Reporter: Stanislav Savulchik
> Priority: Major
>
> > *Note:* This report was drafted by an AI assistant (GitHub Copilot CLI) based on
> > production evidence (broker logs, thread dump, per-thread application logs, task
> > assignment snapshot) supplied by the reporter. All log excerpts and line-number
> > references below were verified against the actual evidence files and the Kafka 3.9
> > source tree; the reporter has reviewed and confirmed the analysis before filing.
> h2. Summary
> A Kafka Streams {{StateUpdater}} thread can terminate on its own (its {{run()}} loop exits)
> without the owning {{StreamThread}}, {{TaskManager}}, or any supervisor ever finding out.
> {{DefaultStateUpdater.add()}}/{{remove()}} have no check for whether the thread is still
> alive: they unconditionally enqueue the request and, for {{remove()}}, return a
> {{CompletableFuture}} that only the {{StateUpdaterThread}} can complete. If that thread is
> already dead, the future is never completed.
> {{TaskManager.waitForFuture()}} ({{TaskManager.java}} lines ~700-720 on the 3.9 branch) then
> blocks on that future with a plain, *unbounded* {{future.get()}} (no timeout argument):
> {code:java}removedTaskResult = future.get();
> {code}
> This call happens on the {{KafkaConsumer.poll()}} thread, inside the rebalance listener
> callback ({{onPartitionsRevoked}} → {{TaskManager.handleRevocation()}} →
> {{revokeTasksInStateUpdater()}} → {{waitForFuture()}}). Because {{poll()}} never returns while
> blocked there, the consumer stops sending heartbeats/JoinGroup requests, and the broker's
> group coordinator eventually evicts the member on session-timeout — __while the JVM
> thread itself is still alive__, parked in {{WAITING}}, with no exception and no crash. Basic
> liveness checks (thread.isAlive(), process health checks) report the instance as healthy.
> h3. Environment / configuration
> * Kafka Streams 3.9.2
> * {{processing.guarantee}} = {{at_least_once}} (this is the intended/correct mode for this
> application; not exactly-once)
> * Topology deployed across 12 application instances, {{num.stream.threads}} = 2 per
> instance (24 StreamThreads total)
> * {{num.standby.replicas}} = 1
> * Default (KIP-892) {{DefaultStateUpdater}} enabled — one dedicated {{StateUpdater-N}} thread
> per {{StreamThread}}
> * Large RocksDB-backed state stores, standby/warmup replicas configured
> h3. Observed sequence (production evidence)
> # *2026-07-19 ~07:01–07:04* — a transient full network outage makes the broker
> cluster unreachable from one application instance ({{production-2}}). Both of its
> StreamThreads (and their restore consumers) see repeated {{Disconnecting from node …
> due to request/socket connection setup timeout}} across many broker nodes at the same
> time — this affects {{StreamThread-1}}/{{StateUpdater-1}} and {{StreamThread-2}}/
> {{StateUpdater-2}} simultaneously.
> # {{StateUpdater-1}} (belonging to the healthy {{StreamThread-1}}) keeps retrying broker
> connections for several more minutes and eventually recovers; {{StreamThread-1}} is
> healthy today.
> # {{StateUpdater-2}} (belonging to {{StreamThread-2}}) does *not* recover the same way.
> Its restore consumer logs {{Unsubscribed all topics or patterns and assigned
> partitions}} at {{07:04:51.883}}, and one second later, at {{07:04:52.933}},
> {{DefaultStateUpdater}} logs:
> {code} state-updater [sputnik-friend-locator-state-engine-production-2-StateUpdater-2] State updater thread stopped
> {code}
> This is the unconditional {{finally}}-block log line in
> {{DefaultStateUpdater$StateUpdaterThread.run()}}. Critically, *neither* of the two
> code paths that are supposed to precede it appear anywhere in the captured logs for
> this event:
> ** no {{"Shutting down state updater thread"}} (the log line the public
> {{shutdown(Duration)}} method emits before setting {{isRunning=false}}), and
> ** no {{"An unexpected error occurred within the state updater thread: …"}} (the
> {{ERROR}}-level line {{handleRuntimeException()}} emits before setting
> {{isRunning=false}}).
> This suggests the thread's {{run()}} loop exited via an uncaught {{Throwable}} that is
> _not_ a {{RuntimeException}} (e.g. an {{Error}}) — which bypasses the
> {{catch (RuntimeException)}} block entirely (so {{handleRuntimeException}} never runs and
> never logs), while the {{finally}} block still executes ({{clearInputQueue()}},
> {{clearUpdatingAndPausedTasks()}}, {{shutdownGate.countDown()}}, and the "State updater
> thread stopped" log). Such a {{Throwable}} propagating out of {{Thread.run()}} is handled
> by the default {{Thread.UncaughtExceptionHandler}}, which prints to {{System.err}} — a
> stream that was not captured in this evidence bundle (only stdout was captured), so we
> cannot pinpoint the exact {{Throwable}}. This gap is noted as a caveat; regardless of the
> trigger, the architectural problem below holds for any path that ends the
> {{StateUpdaterThread}} without going through {{shutdown(Duration)}}.
> # {{StateUpdater-2}} never restarts. No new thread by that name appears again.
> # Roughly *7 hours later*, at {{14:14:48}}, {{StreamThread-2}} (group-instance-id
> {{…production-2-2}}) performs a fresh {{JoinGroup}} and is briefly re-admitted to the
> consumer group (generation 28757, then a couple more generations through {{14:19:57}}).
> # At {{14:20:57}}, the broker {{GroupCoordinator}} removes that member again:
> {{"Member …production-2-2-e3dc728b… in group … has failed, removing it from the
> group"}} (heartbeat/session-timeout expiration). The group stabilizes one member
> short and *never re-admits it again* — it is still short one member days later.
> # A thread dump taken *~2 days later* (2026-07-21) still shows {{StreamThread-2}} parked
> at exactly this call chain:
> {code} java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2072)
> org.apache.kafka.streams.processor.internals.TaskManager.waitForFuture(TaskManager.java:704)
> org.apache.kafka.streams.processor.internals.TaskManager.lambda$getNonFailedTasks$10(TaskManager.java:667)
> ...
> org.apache.kafka.streams.processor.internals.TaskManager.revokeTasksInStateUpdater(TaskManager.java:1215)
> org.apache.kafka.streams.processor.internals.TaskManager.handleRevocation(TaskManager.java:1114)
> org.apache.kafka.streams.processor.internals.StreamsRebalanceListener.onPartitionsRevoked(StreamsRebalanceListener.java:98)
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:402)
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:504)
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:415)
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:511)
> org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:1280)
> {code}
> i.e. it is stuck in the exact revoke path described above, waiting on a
> {{CompletableFuture}} from a {{StateUpdater-2}} thread that stopped existing 2 days
> earlier. There is no {{StateUpdater-2}} thread anywhere in that thread dump — only
> {{StateUpdater-1}}, which belongs to the healthy {{StreamThread-1}} and is legitimately
> {{RUNNABLE}}, actively restoring tasks.
> # Consequence: the client-side task-assignment view still lists {{StreamThread-2}} with
> only stale STANDBY/WARMUP tasks and *zero active tasks*, while the broker's
> consumer-group listing shows one fewer member than the number of live StreamThreads —
> a discrepancy that is very hard to diagnose operationally, since there is no error, no
> crash, and no metric distinguishing "healthy idle thread" from "livelocked thread that
> fell out of the group because its state updater died."
> h3. Relationship to existing tickets
> This shares the same underlying hazard — a {{StreamThread}} blocking on a
> {{StateUpdater}}-completed future during a rebalance listener callback — as:
> * KAFKA-19853 — *StreamThread blocks on StateUpdater during {{onAssignment()}}*, causing
> transaction timeouts under EOS. Same blocking pattern, different callback
> ({{onAssignment}} vs. {{onPartitionsRevoked}}) and different trigger (slow restore vs. a
> dead thread).
> * KAFKA-20456 (resolved) / KAFKA-20721 (open) — {{waitForFuture()}} timing out (after a
> *5-minute bounded* {{future.get(5, TimeUnit.MINUTES)}}, added for 4.1.2+/4.3.0) and
> producing an {{IllegalStateException}} ("ISE" — {{java.lang.IllegalStateException}}, thrown
> when a removal future resolves to {{null}}).
> * KAFKA-18355 / KAFKA-17789 — other cases of a Streams-internal thread blocking
> indefinitely / getting stuck.
> What appears to be new here, specific to 3.9.x:
> * {{TaskManager.waitForFuture()}} in 3.9.2 has **no timeout at all** (confirmed by reading
> the 3.9 branch source directly), so instead of eventually throwing the ISE described in
> KAFKA-20456/20721, the {{StreamThread}} hangs *forever*.
> * The trigger is not "the state updater is busy/slow" but that **the state updater
> thread is confirmed dead__ (absent from the thread dump, with an explicit "State
> updater thread stopped" log and no subsequent restart), and nothing in {{TaskManager}} or
> {{DefaultStateUpdater}} detects or reports that condition before a later caller blocks on
> a future that thread can no longer fulfill.
> h3. Suggested direction
> * Bound {{TaskManager.waitForFuture()}} (all call sites) with a timeout tied to
> {{max.poll.interval.ms}} / the rebalance timeout, and fail loudly (log + crash/replace the
> {{StreamThread}}) rather than allowing an indefinite park.
> * Have {{DefaultStateUpdater.add()}}/{{remove()}} check whether {{stateUpdaterThread}} is still
> alive (or fail the returned future immediately) instead of silently enqueuing work for a
> thread that has already exited.
> * Ensure {{StateUpdaterThread.run()}} cannot exit without going through a single,
> well-logged path: catch {{Throwable}} (not just {{RuntimeException}}) so unexpected {{Error}}s
> are logged at {{ERROR}} before the thread stops, and/or install a
> {{Thread.UncaughtExceptionHandler}} on the {{StateUpdaterThread}} so any uncaught fatal error
> is guaranteed to reach the structured application logs (not just {{System.err}}).
> * Consider a metric/log line when a {{StreamThread}} misses heartbeats due to a stuck
> rebalance-listener callback, to make this diagnosable without a thread dump.
> h2. Related issues
> KAFKA-19853, KAFKA-20456, KAFKA-20721, KAFKA-18355, KAFKA-17789
> h2. Attachments
> None at filing time. Evidence files (coordinator logs, task-assignment CSV,
> consumer-group state, thread dump, and the per-thread {{StateUpdater-2}} log) are available
> and can be attached later on request — recommend redacting instance/consumer-group names
> first, since this is a public ASF Jira.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)