[jira] [Created] (CAMEL-25053) camel-core - Stream Resequence EIP stops delivering after its route is stopped and started while a message waits for its timeout, and callers waiting for capacity are not released on stop

["shashank (Jira)" <ji...@apache.org>]

shashank created CAMEL-25053:
--------------------------------

             Summary: camel-core - Stream Resequence EIP stops delivering after its route is stopped and started while a message waits for its timeout, and callers waiting for capacity are not released on stop
                 Key: CAMEL-25053
                 URL: https://issues.apache.org/jira/browse/CAMEL-25053
             Project: Camel
          Issue Type: Bug
          Components: camel-core
            Reporter: shashank


The stream resequencer holds back a message that has a gap in front of it until the missing message arrives or the timeout expires. The very first message the resequencer receives (no message delivered yet) is held the same way. The timeout is a {{TimerTask}} on the engine's {{java.util.Timer}}:
* {{ResequencerEngine.insert}} ({{core/camel-core-processor/.../resequencer/ResequencerEngine.java:241-281}}) calls {{element.schedule(new Timeout(timer, timeout))}} at {{:273}}.
* {{deliverNext}} ({{:312-345}}) does not deliver the first element while it still has a {{Timeout}} ({{:322}}), and nothing behind it is delivered either.

When the route stops, {{StreamResequencer.doStop}} ({{StreamResequencer.java:234-238}}) calls {{engine.stop()}}, which calls {{timer.cancel()}} ({{ResequencerEngine.java:114-116}}). That discards the pending {{TimerTask}}s. The elements stay in the engine, because the processor, its engine and its sequence are kept across a route restart, and each one still carries its {{Timeout}} ({{Element.scheduled()}} stays true). On the next start, {{engine.start()}} ({{:106}}) creates a new timer but does not schedule those timeouts again. So an element that was waiting at the stop is never delivered, unless the one message it waits for still arrives (then {{insert}} cancels its timeout). Everything inserted behind it waits as well. In other words, the stall is permanent exactly when the gap is never filled, which is the case the timeout exists for. Only removing and re-adding the route recovers it.

Once {{capacity}} elements are held, every caller blocks in {{engine.waitUntil}} ({{StreamResequencer.java:254}}, {{ResequencerEngine.java:139-152}}). That wait has no timeout, checks no state, and is not released by {{stop()}}. The consumer threads stay blocked, and so does a graceful stop of the route: it waits for the blocked exchanges until the shutdown timeout, forces the stop, and the threads are still blocked afterwards. This capacity wait was noted as a follow-up in the review of CAMEL-24995 (PR #26851): "a caller blocked on a full stream resequencer could hang during shutdown". The restart stall above was not mentioned there.

What restarts the resequencer processor: {{ClusteredRoutePolicy}} (the route is stopped when leadership is lost and started when it is regained), the quartz {{ScheduledRoutePolicy}} ({{stopRoute}}/{{startRoute}} on a schedule), and {{stopRoute}}/{{startRoute}} from the Java API, JMX or the controlbus. Policies that only stop the consumer (the throttling route policies, {{HazelcastRoutePolicy}}, the ZooKeeper {{MasterRoutePolicy}}) do not stop the processor and do not trigger this. CAMEL-20435 (the batch resequencer could not be restarted) came from a user restarting routes for active/passive failover, which is the same kind of use.

h3. Reproduction

Route:
{noformat}
from("direct:in").routeId("r").resequence(header("seq")).stream().timeout(500).deliveryAttemptInterval(100).process(collect)
{noformat}
{noformat}
no restart:     send 1, 3, then 4, 5, 6                                    -> delivered [1, 3, 4, 5, 6]
restart:        send 1, 3; stopRoute(r) + startRoute(r) before the 500 ms timeout; send 4, 5, 6
                                                                            -> 3 s later: []   8 s later: []
capacity=3, same restart, then 4 more senders on 4 threads:
                6 s later: delivered [], producers returned 1/4
                stopRoute(r) returned after 3016 ms (forced, inflight=3); afterwards still 1/4 returned
capacity=2, 2 and 3 held for a missing 1, a third sender waits for capacity; stopRoute, no restart:
                stopRoute returned after 3015 ms (forced); 2 s later the third sender has still not returned
{noformat}

With the fix below, the restart case delivers {{[1, 3, 4, 5, 6]}}, all producers of the capacity case return and everything is delivered, and the waiting sender returns when the route stops. The unit tests in the PR cover the same cases.

A TLA+ model (insert, the timer, the delivery thread, the capacity wait, stop and start) finds the same traces: {{NoDeadTimeoutWhileRunning}} is violated in 4 states (Arrive(1), where 1 waits for its timeout, Stop, Start), {{EventuallyDelivered}} is violated after a restart, and {{WaiterReleased}} is violated after a stop. Without a stop, all properties hold. The fixed model holds {{EventuallyDelivered}}, {{WaiterReleased}}, {{Ordered}} and {{NoDup}}.

h3. Proposed fix

In {{ResequencerEngine}}:
* {{start()}}: after creating the new timer, schedule a timeout again for every element that still has one, under the engine lock. The element then waits a full {{timeout}} from the restart, not the time that was left at the stop.
* {{stop()}}: mark the engine stopped and release all callers waiting in {{waitUntil}}.
* {{waitUntil}}: fail with {{RejectedExecutionException}} when the engine is stopped, before and after the wait.

In {{StreamResequencer}}: set that {{RejectedExecutionException}} on the exchange (as the Delay and Throttle EIPs do when they are stopped). This is visible to callers: a caller that was blocked on capacity when the route stopped now gets this exception, where before it stayed blocked. Also end the {{Delivery}} thread on stop with a flag (not an interrupt), so that quick restarts do not leave old delivery threads running next to the new one.

Messages held when the route stops stay in memory and are delivered after the restart. Delivering or waiting for them on stop is a separate question (not part of this ticket).

Affected: long-standing. {{engine.stop()}} has cancelled the timer and {{engine.start()}} has created a new one without rescheduling since at least camel-3.21.0. The capacity wait was a {{Thread.sleep(timeout)}} loop without a stop check up to 4.6 and has been a latch since "Fix busy-wait loops" (12e7338a5c, 4.7).

Duplicate check (2026-09-27): JIRA text "resequencer" returns 40 issues and "StreamResequencer" returns 10. CAMEL-20435 is the batch resequencer that could not restart at all. CAMEL-1034 and CAMEL-1037 (2008) are messages stuck between JMS queues, a different cause and long fixed. GitHub PRs for "resequencer" and "resequence": only #26851, which names the capacity wait as a follow-up. No open PR touches {{ResequencerEngine}} or {{StreamResequencer}}.

_Filed with Claude Code on behalf of allthingssecurity._




--
This message was sent by Atlassian Jira
(v8.20.10#820010)