[
https://issues.apache.org/jira/browse/CAMEL-24947?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-24947.
---------------------------------
Resolution: Fixed
Fixed by https://github.com/apache/camel/pull/26793 (merged as 42f9284a04c2).
Thanks allthingssecurity!
_Claude Code on behalf of davsclaus_
> 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
> Priority: Minor
> Fix For: 4.23.0
>
>
> 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)
