[jira] [Updated] (CAMEL-24947) camel-seda - request/reply timeout race: a late reply is copied into the caller's exchange after the producer already returned ExchangeTimedOutException

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

     [ https://issues.apache.org/jira/browse/CAMEL-24947?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen updated CAMEL-24947:
--------------------------------
    Fix Version/s: 4.23.0

> 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)