shashank created CAMEL-25297:
--------------------------------
Summary: camel-direct, camel-kamelet - while one exchange waits
for the consumer of a suspended or stopped route, the other exchanges of the
same producer are sent to that route
Key: CAMEL-25297
URL: https://issues.apache.org/jira/browse/CAMEL-25297
Project: Camel
Issue Type: Bug
Components: camel-kamelet, camel-core
Reporter: shashank
{{DirectProducer}} caches the consumer of its endpoint in two fields that all
threads sending with the producer share (a route's {{to("direct:x")}} has one
producer):
{code:java}
if (consumer == null || stateCounter != component.getStateCounter()) {
stateCounter = component.getStateCounter();
consumer = component.getConsumer(key, block, timeout);
}
// send to consumer
{code}
{{DirectComponent}} increments {{stateCounter}} whenever a consumer is added or
removed, and a {{DirectConsumer}} removes itself when its route is suspended
({{doSuspend}}) or stopped ({{doStop}}). With {{block=true}} (the default)
{{getConsumer}} then waits up to {{timeout}} (30 s by default) for the consumer
to come back.
The first exchange after the route was suspended sets {{stateCounter}} to the
new value *before* it looks up the consumer, and then waits in {{getConsumer}}.
Every other exchange sent with the same producer in the meantime sees
{{stateCounter == component.getStateCounter()}} and the old, non-null
{{consumer}}, skips the lookup, and is processed by the consumer of the
suspended route. So suspending a direct route (route controller, JMX, a route
policy such as {{ThrottlingInflightRoutePolicy}}, which suspends the consumer
of a direct route so that its callers wait while too many exchanges are
inflight) does not stop concurrent senders: all but one keep going into the
suspended route, and only the first one waits as documented. The same happens
when the route is stopped: the exchanges go to the stopped route, whose error
handler fails them with a {{RejectedExecutionException}}, instead of waiting
for the consumer (and being processed once the route is started again) or
failing with {{DirectConsumerNotAvailableException}} after the timeout. With
{{block=false}} the window is short (the lookup does not wait), but it is the
same race; there two threads that both look up the consumer can also write
their results in the wrong order (a lookup that found the consumer before it
was removed writes it after a later lookup wrote {{null}}), and the removed
consumer then stays cached until the next add or remove.
h3. Reproduction
Test {{DirectProducerSuspendedConsumerTest}} (camel-core):
{{from("direct:start").to("direct:b?timeout=20000")}} and
{{from("direct:b").routeId("b").to("mock:b")}}. One exchange warms up the
producer, then route {{b}} is suspended. A first exchange is sent
asynchronously; a {{DirectComponent}} subclass counts down a latch in
{{getConsumer}}, so the test knows the first exchange is waiting for the
consumer. A second exchange is sent asynchronously. On main it is not waiting:
it reached the suspended route.
{noformat}
No exchange should reach the suspended route ==> expected: <1> but was: <2>
{noformat}
With the fix the second exchange also waits in {{getConsumer}}, and after
{{resumeRoute("b")}} both exchanges are delivered. The same test with
{{stopRoute("b")}} / {{startRoute("b")}}: on main the second exchange does not
wait, it fails at once with a {{RejectedExecutionException}} from the stopped
route's error handler:
{noformat}
The second exchange should wait for the consumer of route b ==> expected:
<false> but was: <true>
{noformat}
With the fix it waits and is delivered after {{startRoute("b")}}. No sleeps:
latches and Awaitility.
The defect was found with a TLA+ model of the cache (two sending threads,
consumer removed/added, each Java statement one step). The property "an
exchange whose check starts after the consumer was removed is not sent to that
consumer" is violated: thread 1 sets the counter and starts the lookup, the
consumer is removed, thread 2's check passes with the old consumer. It holds
with one sending thread, and with the fix for 2 and 3 threads.
h3. Proposed fix
Keep the consumer and the counter in one immutable holder ({{record
CachedConsumer(DirectConsumer consumer, int stateCounter)}} in a {{volatile}}
field). The counter is read *before* the lookup and stored together with its
result, so another exchange either sees the old holder (counter differs, so it
looks up the consumer too, and waits) or the holder with the result of the
lookup. The cost on the fast path is the same as today (one volatile read of
the holder, one of the counter); a holder is allocated only when the consumer
changed.
h3. camel-kamelet
{{KameletProducer}} has the identical code ({{consumer}}/{{stateCounter}},
{{KameletComponent.getConsumer}} with block/timeout,
{{KameletConsumer.doSuspend}}/{{doStop}} remove the consumer) and the same
defect: {{KameletProducerSuspendedConsumerTest}} (a route template
{{kamelet:source -> mock:kamelet}} used with
{{kamelet:echo/echo?timeout=20000}}, the kamelet route {{echo}} suspended or
stopped) fails on main the same way. The same holder fixes it.
With the fix the new tests pass, and {{Direct*}}, {{*Suspend*}}, {{*Resume*}}
in camel-core (60 tests, 0 failures) and the {{Kamelet*Test}} tests of
camel-kamelet (84 tests, 3 skipped) pass.
Affected: 4.14.x, 4.18.x and main (same code, checked with the GitHub contents
API).
Duplicate check (2026-10-03): JIRA text "DirectProducer" with "stateCounter"
(none), "direct" with "suspend" since 2018 (15 issues, none about the producer
cache: CAMEL-14944 rest routes and controlbus, CAMEL-23494 preparingShutdown,
...), "DirectConsumerNotAvailableException" since 2020 (CAMEL-20867,
CAMEL-15165, configuration issues). GitHub pull requests "DirectProducer",
"direct suspended": only #24698 (flaky DirectProducerBlockingTest).
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
