shashank created CAMEL-24948:
--------------------------------
Summary: camel-seda - SedaProducer swallows InterruptedException
and reports success: message not queued, or the request returned as the reply
Key: CAMEL-24948
URL: https://issues.apache.org/jira/browse/CAMEL-24948
Project: Camel
Issue Type: Bug
Components: camel-seda
Reporter: shashank
{{SedaProducer}} catches {{InterruptedException}}, re-sets the interrupt flag
and carries on as if the send had worked:
* {{addToQueue}} with {{blockWhenFull=true}} ({{queue.put}},
SedaProducer.java:237-243).
* {{addToQueue}} with {{blockWhenFull=true&offerTimeout=...}} ({{queue.offer}},
:244-256, commented {{// ignore}}).
* Request/reply with {{timeout<=0}} ({{latch.await()}}, :141-145).
{{process()}} then calls {{callback.done(true)}} without setting an exception,
so the caller is told the send succeeded:
* *InOnly:* the message is not in the queue and is lost.
* *InOut with timeout<=0:* the caller gets back its own request as the "reply".
The real reply is later copied into that exchange when the consumer finishes.
Reproduced against 4.23.0-SNAPSHOT (the caller thread is interrupted while it
waits):
{noformat}
InOnly to seda:full?size=1&blockWhenFull=true (queue full) ->
exception=null, queue=[m1] (m2 lost)
InOnly to seda:full2?size=1&blockWhenFull=true&offerTimeout=5000 ->
exception=null, queue=[m1] (m2 lost)
InOut to seda:slow?timeout=0 ->
body=request, exception=null
{noformat}
Camel interrupts such threads itself when a route stop hits its timeout and the
consumer's thread pool is shut down with {{shutdownNow}}. Route {{seda:up ->
to(seda:down?size=1&blockWhenFull=true)}}, with the {{down}} queue full,
stopped with a 1 s timeout: the step after {{to(seda:down...)}} ran with
{{exception=null}}, and the message was never put in {{down}}. An upstream
transactional or acknowledging consumer would commit/ack such a message.
The non-blocking paths already fail the exchange ("Queue full"
{{IllegalStateException}}, and the {{offerTimeout}} "Fails to insert element
into queue" exception), so only the interrupted paths report a false success.
Proposed fix: keep re-setting the interrupt flag, but fail the exchange:
* {{addToQueue}}, when interrupted in {{put}} ({{blockWhenFull}}), in
{{offer(offerTimeout)}} and in the {{discardWhenFull}} {{offer(0)}} (which
throws when the interrupt flag is already set): throw a
{{RejectedExecutionException("Interrupted while adding the exchange to the
queue")}} with the {{InterruptedException}} as cause. It surfaces to the caller
like the existing "Queue full" / "Fails to insert element into queue"
exceptions.
* Reply wait with {{timeout<=0}}, when interrupted: set the
{{InterruptedException}} on the exchange (as {{DirectProducer}} does), remove
the copy from the queue if it is still there, and make a later reply be ignored
(as on timeout).
* The interrupted *timed* reply wait already fails with
{{ExchangeTimedOutException}}; unchanged.
A PR with the fix follows, with regression test
{{SedaProducerInterruptedTest}}, plus an upgrade-guide note. Regarding
CAMEL-22390 ("do not interrupt Camel threads"): Camel interrupts these threads
itself when a route stop is forced after the shutdown timeout (case D above).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
