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.
