This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24075 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 5f1ca0f1fa1039e3f03873d6b5df4c74f0e49681 Author: Claus Ibsen <[email protected]> AuthorDate: Wed Jul 15 16:09:28 2026 +0200 CAMEL-24075: camel-jms/camel-sjms - transferException should not swallow failures when no JMSReplyTo Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../component/jms/EndpointMessageListener.java | 3 +- .../JmsTransferExceptionNoReplyToIT.java | 123 +++++++++++++++++++++ .../sjms/consumer/EndpointMessageListener.java | 3 +- 3 files changed, 127 insertions(+), 2 deletions(-) diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java index f6c9db538afc..29be0c3b4838 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java @@ -213,7 +213,8 @@ public class EndpointMessageListener implements SessionAwareMessageListener { if (exchange.isRollbackOnly()) { return wrapRuntimeCamelException(new RollbackExchangeException(exchange)); } - if (exchange.isFailed() && exchange.getException() != null && !endpoint.isTransferException()) { + if (exchange.isFailed() && exchange.getException() != null + && !(endpoint.isTransferException() && sendReply)) { return wrapRuntimeCamelException(exchange.getException()); } return null; diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/integration/JmsTransferExceptionNoReplyToIT.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/integration/JmsTransferExceptionNoReplyToIT.java new file mode 100644 index 000000000000..f76fb0ce75ca --- /dev/null +++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/integration/JmsTransferExceptionNoReplyToIT.java @@ -0,0 +1,123 @@ +/* + * 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.jms.integration; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import jakarta.jms.ConnectionFactory; + +import org.apache.camel.CamelContext; +import org.apache.camel.ConsumerTemplate; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.jms.AbstractJMSTest; +import org.apache.camel.component.jms.JmsComponent; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.infra.core.CamelContextExtension; +import org.apache.camel.test.infra.core.DefaultCamelContextExtension; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Order; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.awaitility.Awaitility.await; + +/** + * Tests that transferException=true does not swallow exceptions when the incoming message has no JMSReplyTo + * (fire-and-forget / InOnly). Before the fix, a failed exchange would silently commit the transaction and lose the + * message because transferException suppressed the exception unconditionally — even when no reply destination existed + * to transfer it to. + */ +public class JmsTransferExceptionNoReplyToIT extends AbstractJMSTest { + + @Order(2) + @RegisterExtension + public static CamelContextExtension camelContextExtension = new DefaultCamelContextExtension(); + + private static final String QUEUE_NAME + = "JmsTransferExceptionNoReplyToIT"; + + private static final AtomicInteger counter = new AtomicInteger(); + + protected CamelContext context; + protected ProducerTemplate template; + protected ConsumerTemplate consumer; + + @BeforeEach + public void setUp() { + counter.set(0); + } + + @Test + public void testTransferExceptionNoReplyToShouldRedeliver() throws Exception { + MockEndpoint mock = context.getEndpoint("mock:dead", MockEndpoint.class); + mock.expectedMinimumMessageCount(1); + + // Send fire-and-forget (InOnly, no JMSReplyTo) to a transacted queue with transferException=true. + // The route always throws, so without the fix the message is silently lost. + template.sendBody("activemq:queue:" + QUEUE_NAME, "Kaboom"); + + MockEndpoint.assertIsSatisfied(context, 20, TimeUnit.SECONDS); + + // The processor should have been invoked more than once due to redelivery + await().atMost(5, TimeUnit.SECONDS).until(() -> counter.get() > 1); + } + + @Override + protected String getComponentName() { + return "activemq"; + } + + @Override + protected JmsComponent setupComponent( + CamelContext camelContext, ConnectionFactory connectionFactory, String componentName) { + JmsComponent component = super.setupComponent(camelContext, connectionFactory, componentName); + component.setObjectMessageEnabled(true); + return component; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + errorHandler(deadLetterChannel("mock:dead") + .maximumRedeliveries(2) + .redeliveryDelay(0)); + + from("activemq:queue:" + QUEUE_NAME + "?transferException=true&transacted=true") + .process(exchange -> { + counter.incrementAndGet(); + throw new IllegalArgumentException("Boom"); + }); + } + }; + } + + @Override + public CamelContextExtension getCamelContextExtension() { + return camelContextExtension; + } + + @BeforeEach + void setUpRequirements() { + context = camelContextExtension.getContext(); + template = camelContextExtension.getProducerTemplate(); + consumer = camelContextExtension.getConsumerTemplate(); + } +} diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java index f8b90213a1bc..3e16b22f85fb 100644 --- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java +++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java @@ -419,12 +419,13 @@ public class EndpointMessageListener implements SessionMessageListener { rce = wrapRuntimeCamelException(new RollbackExchangeException(exchange)); } else if (exchange.getException() != null) { // an exception occurred while processing - if (endpoint.isTransferException()) { + if (endpoint.isTransferException() && sendReply) { // send the exception as reply, so null body and set the exception as the cause body = null; cause = exchange.getException(); } else { // only throw exception if endpoint is not configured to transfer exceptions back to caller + // or if there is no reply destination to send the exception to // do not send a reply but wrap and rethrow the exception rce = wrapRuntimeCamelException(exchange.getException()); }
