shashank created CAMEL-24947:
--------------------------------
Summary: camel-seda - request/reply timeout race: a late reply is
copied into the caller's exchange after the producer already returned
ExchangeTimedOutException
Key: CAMEL-24947
URL: https://issues.apache.org/jira/browse/CAMEL-24947
Project: Camel
Issue Type: Bug
Components: camel-seda
Reporter: shashank
With {{waitForTaskToComplete=Always}}/{{IfReplyExpected}} and {{timeout > 0}},
a reply that arrives at about the time of the timeout can be copied into the
caller's exchange after {{SedaProducer}} has already returned to the caller
with {{ExchangeTimedOutException}}. The caller can also see the reply body
together with the timeout exception.
{{SedaProducer.process()}}:
* The consumer side ({{onDone}}, SedaProducer.java:73-93) runs {{if
(latch.getCount() == 0) return;}}, then {{ExchangeHelper.copyResults(exchange,
response)}}, then {{latch.countDown()}}.
* The producer side (:125-134) runs {{if (!latch.await(timeout)) {
exchange.setException(timeout); queue.remove(copy); latch.countDown(); }}},
then {{callback.done(true)}}.
The latch check in {{onDone}} and the copy are not atomic with the timeout
branch. If the consumer passes the check and the timeout fires before it counts
down, both threads write the caller's exchange. The consumer can keep writing
after {{callback.done}}. {{copyResults}} replaces the message, properties,
variables and flags (routeStop, rollbackOnly, redelivery counters,
errorHandlerHandled), and finally calls {{setException(null)}}, while the
caller's error handler / doCatch is already working on the same exchange.
Reproduced against 4.23.0-SNAPSHOT. The pause below is a {{SafeCopyProperty}}
set by the consumer route; {{copyResults}} calls it after the message copy and
before {{setException}}:
{noformat}
template.send(seda:a?timeout=300), consumer reply copy paused inside onDone
returned to caller after 315 ms: body=reply,
exception=ExchangeTimedOutException
same exchange 300 ms later: body=reply, exception=null
route doTry{to(seda:b?timeout=300)} doCatch(ExchangeTimedOutException)
inside doCatch: body=reply, exception=null, caught=ExchangeTimedOutException
{noformat}
Without any pause, 400 requests with {{timeout=20}} and consumers taking 18-22
ms gave 24 exchanges returned with the reply body plus
{{ExchangeTimedOutException}}, and 2 exchanges that were modified after the
producer had returned.
The comment in {{onDone}} ("check for timeout, which then already would have
invoked the latch") shows the intent: a late response should be ignored.
Proposed fix: let exactly one side own the result, with an {{AtomicBoolean
completed}} shared by the producer and the {{onDone}} synchronization:
* {{onDone}} does {{completed.compareAndSet(false, true)}} before copying, and
ignores the response if that fails.
* On timeout, the producer does {{completed.compareAndSet(false, true)}}; if it
wins it sets {{ExchangeTimedOutException}} and removes the copy from the queue
as today. If it loses (the consumer is copying the reply), it waits for the
latch without timeout (the copy is short) and returns the reply instead of the
timeout.
* The {{latch.countDown()}} in the timeout branch is no longer needed to make
{{onDone}} ignore a late reply.
A PR with the fix follows, with regression test {{SedaTimeoutLateReplyTest}}
(pauses the consumer inside the reply copy via a {{SafeCopyProperty}}). If the
fix for the related interrupt issue (SedaProducer swallows
InterruptedException) is merged too, the interrupted reply wait must claim the
exchange the same way.
(Found with a TLA+ model of SedaProducer/SedaConsumer; the fixed variant
passes.)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
