Hi Richard, Thank you - this is exactly the level of detail we were hoping for, and it confirms our reading of the internals (and clears up the bits we were unsure about). Special thanks for the STORM-2359 / STORM-3314 pointers; knowing the acker-redesign path was explored and abandoned tells us not to go looking for a reclaim-on-loss hook that was never going to exist.
For the record, here is the fix we are currently running. To be precise about it: it is a no-stall fix with bounded, recorded loss - not a no-loss fix - and I want to state that plainly rather than overclaim. Context: we do not use the stock KafkaSpout. We have a custom consumer whose offset tracking is a faithful port of storm-kafka-client's OffsetManager (contiguous-acked-prefix commit), feeding a Storm spout. We require strict per-partition (effectively per-device) ordering at the sink, and we already run a Hazelcast dedup store. We are on 2.0.0. What we did was move correctness out of Storm's ack/timeout machinery and into the consumer's offset layer: 1. topology.enable.message.timeouts = false and maxSpoutPending = 0. We understand this is STORM-3514 / debugging-flag territory. We went there to avoid two failure modes at once: the orphan-stall with a finite cap, and the false-expiry of legitimately queued tuples under backpressure (queue-wait is wall-clock since emit, as you note, and resetTimeout cannot cover the receive queue). The cost we accept is that Storm's pending map then only accumulates orphan entries as memory - bounded per worker-death, cleared on restart - which we monitor. 2. fail-as-ack: fail() calls markProcessed rather than re-emitting, to preserve per-partition order, and the dedup store suppresses any duplicate. The OffsetManager commits only the contiguous acked prefix. 3. A liveness guard (detectStuckCommits, ~800s) force-advances past a blocker that has stalled the commit far longer than any legitimate processing time, recorded as a structured failure event. We also found and fixed a bridge-skip bug (committing past a re-polled-but-not-yet-emitted offset) by tracking the offset at stage time rather than emit time. On worker death, our reading is that this design splits into two cases, and I would like to cross-check it with you: - Case A - the dead worker hosted the spout/consumer for that partition. The partition is reassigned, the new consumer seeks to the committed offset and re-polls, and the orphaned suffix is reprocessed in order. No loss. (This is the clean case you described.) - Case B - the spout/consumer survived, but the bolt/acker that was processing the record was on the dead worker (common for us, since our grouping routes a record to a bolt on another worker). That partition is not reassigned, so there is no re-poll; the orphan never acks and, with timeouts off, never fails, so it blocks the commit until the ~800s force-advance commits past it without re-emitting - i.e. that record is dropped. fail-as-ack is the same shape: a record that fails processing is committed past, not retried. So our question: is that the correct reading - that under fail-as-ack + force-advance, Case B is genuinely a drop, and the only no-loss case is Case A (spout-own-worker death recovered by re-poll)? It matches your point (4), but we want to be sure we are not missing a recovery path. Assuming that is right, your point (4) is the one that lands hardest, and we agree the drop-vs-recovery-latency dilemma comes from fail-as-ack, not the timeout. The sink-side resequencer is the piece we had been missing - we had ruled out replay because it reorders in transit, without seeing that order can be re-imposed at the point of application. So we are now evaluating a move to real redelivery with unbounded retries, a per-partition resequencer at the apply point (on top of the dedup state we already keep), and an explicit DLQ-after-N as the only deliberate drop - so that the one residual drop is explicit and recorded rather than a timeout side effect. The compacted-topic gap case you flagged (next-expected from emission order, not last+1) is on the list. On the upgrade: point taken, and it is sharper for us than for most, since we have hand-ported the 2.0.0-era OffsetManager and may have faithfully reproduced bugs that the post-2.0.0 at-least-once fixes addressed. We are going to plan the move off 2.0.0 and, in doing so, seriously evaluate adopting storm-kafka-client's AT_LEAST_ONCE path directly rather than maintaining the port, keeping our custom code to the sink-side resequencer + DLQ. Thanks again - this reframing saved us from optimising the wrong knob. On Thu, Jun 11, 2026 at 12:47 PM Richard Zowalla <[email protected]> wrote: > Hi, > > Nice analysis - your reading of the internals is correct (at least if I > followed correctly. Rui might want to add or disagree ;-) ) > > Answers inline, plus a suggestion at the end that I think dissolves the > short-vs-long timeout dilemma. > > 1) Is the spout timeout really the only orphan-reclamation path? > > Yes. An entry leaves the spout's pending map (SpoutExecutor.pending, a > RotatingMap) in exactly three ways: an ack or fail delivered by the acker, > or expiry on rotation. failSpoutMsg() is invoked from exactly two places — > the rotation-expiry callback ("TIMEOUT") and the acker fail stream. There > is no worker-death -> fail path. And with topology.enable.message.timeouts > set to false, Executor.setupTicks() never schedules the tick that rotates > the spout's (and the acker's) maps, so orphans accumulate forever and, with > maxSpoutPending, stall the spout. This is the STORM-3514 behavior you > reproduced. > That issue was closed Won't Fix; the flag is meant for debugging sessions, > not production, afaik. > > The reason reclaim-on-loss doesn't exist is structural. > The acker keeps ~20 bytes per tree (XOR value, start time, spout task id, > see Acker.AckObject). It deliberately does not know which tuples, tasks or > workers hold pieces of a tree; that is the trick that makes tracking > millions of in-flight trees cheap. So when a worker dies, no component can > enumerate "the trees that had state on that worker": the spout can't, the > acker can't (for acker loss the tracking state itself is what died), and as > you observed the messaging layer only sees serialized TaskMessages with no > anchor information. Failing orphans on the loss event would require > materializing tree->worker membership somewhere, i.e. a redesign of the > acking subsystem. That redesign was sketched twice - STORM-2359 "Revising > Message Timeouts" and STORM-3314 "Acker redesign" and both were > abandoned. So: the timeout is the supported mechanism, by design. > > One asymmetry worth knowing: death of the spout's *own* worker is the clean > case: the spout restarts and (with storm-kafka-client AT_LEAST_ONCE) > rewinds to the committed offsets, which are always a contiguous acked > prefix, replaying the suffix in order. If reclamation latency after the > death of *other* workers really matters, a blunt operational workaround is > a watchdog that forces a spout-worker restart when it sees a worker > reassignment, converting orphan limbo into an immediate, order-preserving > offset rewind. Only sane if replayed duplicates are safe at your sink, > see (4). > > 2) Can the timeout exclude queue-wait, i.e. act on > time-since-last-progress? > > No. The clock is wall-clock since emit, implemented as bucketed rotating > maps; the spout map has 2 buckets rotated every message.timeout.secs, so > expiry actually fires between 1x and 2x the configured value. No per-tuple > "last progress" timestamp exists anywhere, that means the progress events > are the > acks themselves, and they are XOR-folded the moment they arrive. What you > are describing is precisely STORM-2359; it was never implemented. If you > wanted to revive that work, dev@ or the issue tracker would welcome the > discussion but there is no existing code point I can point you at. > > 3) Is resetTimeout the intended tool / can it cover queued tuples? > > It is the only tool, and you have found its real limits. It cannot cover > queue wait by design: a receive queue is drained solely by its executor > thread, so while that thread is busy nothing can see, let alone reset, the > tuples waiting in it. It is also costly: each call sends one message per > anchor root to the acker and the acker forwards one to the spout, all > through the same transfer/receive queues. AFAIK, that means that > under backpressure the resets queue behind the very backlog they are > trying to excuse. You can cut the traffic (rate-limit per root id via > tuple.getMessageId().getAnchors(), at > most one reset per timeout/3), but no variant fixes the queued-tuple case. > I would not build on it here. > > The practical way to make the timeout a pure liveness bound is to bound > legitimate sojourn time instead of trying to measure progress: with > maxSpoutPending set, in-flight work per spout task is capped, so worst-case > queue-wait-plus-processing is roughly maxSpoutPending x slowest per-tuple > service time / parallelism. Set topology.message.timeout.secs comfortably > above that bound and a live tuple cannot expire under backpressure; only > genuinely stuck or orphaned ones can. The cost is that orphan reclamation > after a worker death takes up to ~2x the timeout, during which a spout > stalled at the cap stays stalled. Recovery latency == timeout is the > inherent trade in core Storm, because per (1) the timeout *is* the orphan > detector. > > 4) Is core Storm's model the right fit? > > The bigger point: your dilemma (drop live data vs slow recovery) is not > created by the timeout; it is created by fail-as-ack. Storm's at-least- > once contract is "fail/timeout => redeliver"; once fail means drop, every > premature timeout is data loss and the timeout becomes a correctness knob. > If you restore redelivery and enforce ordering at the point of application, > the timeout goes back to being a latency-vs-duplicate-churn and the > tension disappears: > > - Use real retries (storm-kafka-client AT_LEAST_ONCE: on fail the spout > seeks back to the earliest retriable offset per partition and re-emits > forward; it only ever commits the contiguous acked prefix). Make sure > maxRetries on the retry service is unbounded, otherwise the spout will > eventually commit past a failing offset and your ordering chain gets a > permanent hole. > > - In the sink, next to the dedup state you already keep, track last-applied > offset per partition and apply in offset order: next expected offset -> > apply and ack; at-or-below last applied -> duplicate, ack without > applying; ahead of a gap (a retry is still in flight behind it) -> fail > or park it so it is redelivered once the gap fills. (If the topic is > compacted or transactional, offsets have legitimate gaps, so "next > expected" must be derived from emission order rather than last+1.) > > That gives strict per-partition order, no drops, and worker-loss survival: > orphans are reclaimed by timeout and redelivered; out-of-order redeliveries > are resequenced by the offset check. You are effectively adding a small > per-partition resequencer at the apply point, on top of the dedup store you > already run. > > If you would rather not build that: Trident automates exactly this pattern > (ordered, txid-gated batches; opaque/transactional Kafka spouts exist in > storm-kafka-client), at micro-batch latency; however, I will be honest that > Trident is in maintenance mode. Of the two I would most likely pick core > Storm plus the > sink-side resequencer. And for completeness: "per-partition order + > effectively-once + replay on failure" is the native model of checkpoint/ > replay engines (Flink, Kafka Streams), so if you are at a decision point > anyway it is fair to evaluate them. But you do not need to leave Storm to > get correctness here. > > Finally, please upgrade from 2.0.0 - current is 2.8.8, and the at-least- > once paths in storm-kafka-client in particular received a long series of > correctness fixes after 2.0.0. > > Refs: > - STORM-3514 (closed Won't Fix): > https://github.com/apache/storm/issues/7296 > - STORM-2359 Revising Message Timeouts (abandoned): > https://github.com/apache/storm/issues/6141 > - STORM-3314 Acker redesign (abandoned): > https://github.com/apache/storm/issues/7096 > > Gruß > Richard > > > > Am 10.06.2026 um 10:33 schrieb Karthick <[email protected]>: > > > > Hi all, > > We run Apache Storm 2.0.0 with a Kafka-fed topology that needs strict > per-partition ordering. To preserve order we use at-least-once with fail() > treated as ack (no replay) plus an external dedup store. > > > > We've hit a tension between two Storm behaviors and would appreciate > guidance on the intended approach. > > > > Background: when a worker dies, every in-flight tuple tree that had a > bolt or acker on it is orphaned — its ack/fail never returns. With > topology.max.spout.pending set, those orphans fill the spout's pending map > and nextTuple stops being called (we've confirmed this against STORM-3514: > with topology.enable.message.timeouts=false + maxSpoutPending, an un-acked > tuple stalls the spout permanently;) > > > > Case 1 — reclaiming orphans seems to require the timeout, and only the > timeout. > > A pending entry leaves the spout only via ack, fail, or > timeout-expiry. Orphans get none unless timeouts are on. We considered > failing the tuple at the point Netty drops a message to the dead worker > ("Dropping N messages"), but that only addresses bolt-loss orphans — and > even then the messaging layer doesn't know the originating tuple tree. It > does nothing for acker-loss orphans, where the acker's tracking state died > with the worker and there's no drop signal to act on. So the spout's own > timeout appears to be the only mechanism that reclaims both classes. Is > that correct, or is there a supported way to fail/reclaim orphaned trees on > worker loss without waiting for the timeout? > > > > Case 2 — but the message timeout counts queue/backpressure wait, not > just processing. > > topology.message.timeout.secs is wall-clock from emit and covers the > whole journey — queue wait + processing + ack. Under backpressure, a tuple > sitting idle in a downstream bolt's receive queue (behind slower tuples) > has its clock running while it waits, and can be failed before it is ever > processed. With our fail-as-ack semantics that drops a live, valid message > purely because the pipeline was momentarily backed up. So a short timeout > risks dropping good data under load, while a long timeout slows orphan > reclamation — and we can't turn it off (Case 1). > > > > We'd like the timeout to behave as a liveness / no-progress timer — > expire a tuple only if it has made no progress for N seconds (genuinely > stuck/orphaned), not if it has merely been waiting in a queue. > > > > What we've tried: collector.resetTimeout(tuple) at the start of every > bolt's execute(). It correctly resets the clock for tuples that are being > processed, but it can't cover the wait before a bolt dequeues a tuple > (nothing resets a tuple while it's idle in the receive queue, since the > executor thread is busy), and at our throughput the per-hop resetTimeout > traffic to the ackers is significant. > > > > Questions: > > 1. For surviving worker loss without dropping live-but-slow tuples, is > the spout timeout really the only orphan-reclamation path, or is there a > supported way to fail orphaned trees on the loss event itself? > > 2. Is there any way to make the timeout exclude time spent waiting in > receive queues (i.e. expire on "time since last progress" rather than "time > since emit")? > > 3. Is resetTimeout the intended tool here? Is there a recommended > pattern to reset a tuple that is queued but not yet in execute() without > flooding the ackers? > > 4. More broadly: for ordered, effectively-once processing on Kafka > that must survive worker failures, is core Storm's per-tuple ack/timeout > model the right fit, or is the guidance to use Trident / a different > > framework for this case? > > > > We've read STORM-3514 and the Guaranteeing-Message-Processing docs; > this is the gap that leaves us choosing between dropping live data (short > timeout) and slow recovery (long timeout). > > > > Thanks for any pointers. > >
