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.
