shashank created CAMEL-25038:
--------------------------------
Summary: camel-disruptor - request/reply: a late reply is written
into the caller's exchange after ExchangeTimedOutException, and an interrupted
wait reports success (same bugs as CAMEL-24947 / CAMEL-24948 for seda)
Key: CAMEL-25038
URL: https://issues.apache.org/jira/browse/CAMEL-25038
Project: Camel
Issue Type: Bug
Components: camel-disruptor
Reporter: shashank
{{DisruptorProducer.process()}}
({{components/camel-disruptor/.../DisruptorProducer.java:73-149}}) waits for a
reply the same way {{SedaProducer}} does, and has the same two bugs that
CAMEL-24947 and CAMEL-24948 fix for camel-seda. The seda PRs #26793 (open) and
#26794 (merged) only change {{SedaProducer}}.
# Timeout race. The consumer side ({{newOnCompletion}}, {{:152-172}}) does {{if
(latch.getCount() == 0) ignore; else ExchangeHelper.copyResults(exchange,
response); latch.countDown()}}. The timeout branch ({{:108-123}}) does
{{exchange.setProperty(DISRUPTOR_IGNORE_EXCHANGE)}},
{{exchange.setException(new ExchangeTimedOutException(...))}},
{{latch.countDown()}} and then {{callback.done(true)}}. The check and the copy
are not atomic with the timeout branch. When the timeout fires while the
consumer is inside {{copyResults}}, the caller gets the reply body together
with {{ExchangeTimedOutException}}. The consumer then keeps writing into the
caller's exchange after the producer has returned: {{copyResults}} ends with
{{setException(source.getException())}}, which clears the timeout exception
while the caller's error handler or doCatch works on it. (Setting
{{DISRUPTOR_IGNORE_EXCHANGE}} on the caller's exchange has no effect on the
copy that is already in the ring.)
# Interrupt. With {{timeout <= 0}} ({{:124-135}}), an interrupt of
{{latch.await()}} is logged at INFO and otherwise swallowed. The producer then
calls {{callback.done(true)}} with no exception, so the caller sees a
successful exchange whose "reply" is its own request. The real reply is written
into the exchange later. With {{timeout > 0}} ({{:101-106}}), an interrupt
leaves {{done}} false and is reported as a timeout, which is acceptable, but
that path still has race 1.
*Reproduction* (4.23.0-SNAPSHOT). As in the seda ticket, the pause is a
{{SafeCopyProperty}} set by the consumer route; {{copyResults}} calls it after
the message copy and before {{setException}}:
{noformat}
paused: template.send(disruptor:b?timeout=300, InOut), consumer reply copy
paused inside onDone
returned to caller after 337 ms: body=reply,
exception=ExchangeTimedOutException ...
same exchange 400 ms later: body=reply, exception=null
natural: 400 requests, timeout=20 ms, consumer takes 18-22 ms, no pause:
timeouts=238, returned with the reply body AND
ExchangeTimedOutException=25
interrupt: InOut timeout=0, caller thread interrupted while waiting:
returned body=request, exception=null, failed=false
600 ms later: body=reply, exception=null
{noformat}
TLA+: the seda producer/consumer model, adapted to {{DisruptorProducer}}:
{{pw_timeout_race}} violates {{NoWriteAfterReturn}}, {{pw_timeout_view}}
violates {{CallerViewStable}}, and {{pw_notimeout_interrupt}} violates
{{ConsistentOutcome}}. The fixed variant {{pw_fixed}} satisfies all of them
plus {{ProducerReturns}}.
*Proposed fix:* the same as for seda (CAMEL-24947 / CAMEL-24948):
* Let exactly one side own the result through an {{AtomicBoolean}} shared by
the producer and the onDone synchronization. onDone claims it before copying
and ignores the reply if the claim fails. On timeout, the producer claims it.
If the producer wins, it sets {{ExchangeTimedOutException}}. If it loses, the
consumer is copying the reply, so it waits for the latch without a timeout and
returns the reply.
* On {{InterruptedException}}, restore the interrupt flag, claim the result,
and fail the exchange (for example with a {{RejectedExecutionException}} or
{{CamelExchangeException}} "interrupted while waiting for reply") instead of
reporting success.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
