This is an automated email from the ASF dual-hosted git repository. apupier pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit e5ad48340d3e259450cca78e148562ad6e984dcb Author: smjain <[email protected]> AuthorDate: Wed Sep 23 18:27:40 2026 +0530 CAMEL-24948: camel-seda - fail the exchange when the producer is interrupted while waiting Cause: SedaProducer caught InterruptedException, re-set the interrupt flag and carried on as if the send had worked, when blocked on a full queue (blockWhenFull put, blockWhenFull with offerTimeout, discardWhenFull offer) and when waiting for the reply without timeout (timeout=0). process() then completed the exchange without an exception. Effect: an InOnly message that was never added to the queue is reported as sent (and an upstream transactional or acknowledging consumer commits it), and an InOut caller gets its own request back as the reply. Camel itself interrupts a route thread blocked in put when a route stop hits its timeout, so this also happens without user code interrupting threads. Fix: when interrupted while adding to the queue, keep the interrupt flag and fail with a RejectedExecutionException caused by the InterruptedException, like the other "not added" paths. When interrupted while waiting for the reply, keep the interrupt flag, fail the exchange with the InterruptedException (as DirectProducer does), remove the copy from the queue if still there, and ignore a later reply, as on timeout. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../apache/camel/component/seda/SedaProducer.java | 17 ++- .../seda/SedaProducerInterruptedTest.java | 145 +++++++++++++++++++++ 2 files changed, 161 insertions(+), 1 deletion(-) diff --git a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaProducer.java b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaProducer.java index 77bd10b92c9d..96090c55a377 100644 --- a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaProducer.java +++ b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaProducer.java @@ -19,6 +19,7 @@ package org.apache.camel.component.seda; import java.io.IOException; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import org.apache.camel.AsyncCallback; @@ -141,7 +142,14 @@ public class SedaProducer extends DefaultAsyncProducer { try { latch.await(); } catch (InterruptedException e) { + LOG.debug("Interrupted while waiting for task to complete at [{}]", endpoint.getEndpointUri()); Thread.currentThread().interrupt(); + // the task has not completed so fail the exchange (do not return the request as the reply) + exchange.setException(e); + // remove the Exchange from queue (if not yet processed) + endpoint.getQueue().remove(copy); + // count down to indicate the reply must be ignored + latch.countDown(); } } } else { @@ -233,6 +241,7 @@ public class SedaProducer extends DefaultAsyncProducer { } catch (InterruptedException e) { LOG.debug("Offer interrupted, are we stopping? {}", isStopping() || isStopped()); Thread.currentThread().interrupt(); + throw interruptedWhileAddingToQueue(e); } } else if (blockWhenFull && offerTimeout == 0) { try { @@ -240,6 +249,7 @@ public class SedaProducer extends DefaultAsyncProducer { } catch (InterruptedException e) { LOG.debug("Put interrupted, are we stopping? {}", isStopping() || isStopped()); Thread.currentThread().interrupt(); + throw interruptedWhileAddingToQueue(e); } } else if (blockWhenFull && offerTimeout > 0) { try { @@ -250,13 +260,18 @@ public class SedaProducer extends DefaultAsyncProducer { + "after timeout of " + offerTimeout + " milliseconds"); } } catch (InterruptedException e) { - // ignore LOG.debug("Offer interrupted, are we stopping? {}", isStopping() || isStopped()); Thread.currentThread().interrupt(); + throw interruptedWhileAddingToQueue(e); } } else { queue.add(target); } } + private static RejectedExecutionException interruptedWhileAddingToQueue(InterruptedException cause) { + // the exchange was not added to the queue, so the exchange must fail + return new RejectedExecutionException("Interrupted while adding the exchange to the queue", cause); + } + } diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaProducerInterruptedTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaProducerInterruptedTest.java new file mode 100644 index 000000000000..67d6f421c7ae --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaProducerInterruptedTest.java @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.seda; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.ExchangePattern; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.support.SynchronizationAdapter; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * A seda producer that is interrupted while it waits must fail the exchange, and not report the send as successful. + */ +public class SedaProducerInterruptedTest extends ContextTestSupport { + + private final ExecutorService executor = Executors.newSingleThreadExecutor(); + private final CountDownLatch releaseConsumer = new CountDownLatch(1); + private final CountDownLatch consumerDone = new CountDownLatch(1); + + @Override + @AfterEach + public void tearDown() throws Exception { + releaseConsumer.countDown(); + executor.shutdownNow(); + super.tearDown(); + } + + @Test + public void testInterruptedWhileBlockedWhenFull() throws Exception { + assertInterruptedWhileAddingToQueue("seda:full?size=1&blockWhenFull=true"); + } + + @Test + public void testInterruptedWhileBlockedWhenFullWithOfferTimeout() throws Exception { + assertInterruptedWhileAddingToQueue("seda:full?size=1&blockWhenFull=true&offerTimeout=20000"); + } + + private void assertInterruptedWhileAddingToQueue(String uri) throws Exception { + // the queue is full (and has no consumer) + template.sendBody(uri, "A"); + + Exchange exchange = context.getEndpoint(uri).createExchange(ExchangePattern.InOnly); + exchange.getMessage().setBody("B"); + Exchange out = sendAndInterrupt(uri, exchange); + + RejectedExecutionException e = assertInstanceOf(RejectedExecutionException.class, out.getException()); + assertInstanceOf(InterruptedException.class, e.getCause()); + // B was not added to the queue + List<Object> bodies = new ArrayList<>(); + for (Exchange queued : context.getEndpoint(uri, SedaEndpoint.class).getQueue()) { + bodies.add(queued.getMessage().getBody()); + } + assertEquals(List.of("A"), bodies); + } + + @Test + public void testInterruptedWhileWaitingForReply() throws Exception { + Exchange exchange = context.getEndpoint("seda:slow").createExchange(ExchangePattern.InOut); + exchange.getMessage().setBody("request"); + Exchange out = sendAndInterrupt("seda:slow?timeout=0", exchange); + + // the request must not be returned as the reply + assertInstanceOf(InterruptedException.class, out.getException()); + + // and the reply from the consumer is ignored when it completes later + releaseConsumer.countDown(); + assertTrue(consumerDone.await(10, TimeUnit.SECONDS)); + assertInstanceOf(InterruptedException.class, out.getException()); + assertEquals("request", out.getMessage().getBody()); + } + + private Exchange sendAndInterrupt(String uri, Exchange exchange) throws Exception { + AtomicReference<Thread> sender = new AtomicReference<>(); + Future<Exchange> future = executor.submit(() -> { + sender.set(Thread.currentThread()); + return template.send(uri, exchange); + }); + // wait until the sender is blocked in the seda producer + await().atMost(10, TimeUnit.SECONDS).until(() -> isWaitingInSedaProducer(sender.get())); + sender.get().interrupt(); + return future.get(10, TimeUnit.SECONDS); + } + + private static boolean isWaitingInSedaProducer(Thread thread) { + if (thread == null + || thread.getState() != Thread.State.WAITING && thread.getState() != Thread.State.TIMED_WAITING) { + return false; + } + for (StackTraceElement element : thread.getStackTrace()) { + if (SedaProducer.class.getName().equals(element.getClassName())) { + return true; + } + } + return false; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("seda:slow").routeId("slow") + .process(e -> e.getExchangeExtension().addOnCompletion(new SynchronizationAdapter() { + @Override + public void onDone(Exchange exchange) { + consumerDone.countDown(); + } + })) + .process(e -> releaseConsumer.await(20, TimeUnit.SECONDS)) + .setBody(constant("reply")); + } + }; + } +}
