This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24073 in repository https://gitbox.apache.org/repos/asf/camel.git
commit a6e0dcff9e273fba92a9de679cee041095589597 Author: Claus Ibsen <[email protected]> AuthorDate: Wed Jul 15 15:58:12 2026 +0200 CAMEL-24073: camel-jms - Cancel reply correlation on send failure to prevent double callback Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/component/jms/JmsProducer.java | 11 +- .../camel/component/jms/reply/ReplyManager.java | 10 ++ .../component/jms/reply/ReplyManagerSupport.java | 10 ++ .../jms/JmsInOutSendFailureCallbackTest.java | 150 +++++++++++++++++++++ 4 files changed, 180 insertions(+), 1 deletion(-) diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsProducer.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsProducer.java index 0db06aba2d33..f3b2ddf36cc2 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsProducer.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsProducer.java @@ -221,6 +221,8 @@ public class JmsProducer extends DefaultAsyncProducer { in.setHeader(correlationPropertyToUse, GENERATED_CORRELATION_ID_PREFIX + getUuidGenerator().generateUuid()); } + final String[] registeredCorrelationId = new String[1]; + MessageCreator messageCreator = new MessageCreator() { public Message createMessage(Session session) throws JMSException { Message answer = endpoint.getBinding().makeJmsMessage(exchange, in, session, null); @@ -241,6 +243,7 @@ public class JmsProducer extends DefaultAsyncProducer { String correlationId = determineCorrelationId(answer, provisionalCorrelationId); replyManager.registerReply(replyManager, exchange, callback, originalCorrelationId, correlationId, timeout); + registeredCorrelationId[0] = correlationId; if (correlationProperty != null) { replyManager.setCorrelationProperty(correlationProperty); @@ -256,7 +259,13 @@ public class JmsProducer extends DefaultAsyncProducer { } }; - doSend(exchange, true, destinationName, destination, messageCreator, messageSentCallback); + try { + doSend(exchange, true, destinationName, destination, messageCreator, messageSentCallback); + } catch (Exception e) { + // send failed after reply was registered, cancel to prevent double callback on timeout + replyManager.cancelCorrelationId(registeredCorrelationId[0]); + throw e; + } // continue routing asynchronously (reply will be processed async when its received) return false; diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManager.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManager.java index a18e20b69ff8..c95de55f8ede 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManager.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManager.java @@ -101,6 +101,16 @@ public interface ReplyManager extends SessionAwareMessageListener { */ void updateCorrelationId(String correlationId, String newCorrelationId, long requestTimeout); + /** + * Cancels a pending reply correlation, removing it from the correlation map without invoking the callback. + * <p/> + * This is used when the JMS send fails after the reply has been registered, to prevent the timeout handler from + * firing a second callback on an already-completed exchange. + * + * @param correlationId the correlation id to cancel + */ + void cancelCorrelationId(String correlationId); + /** * Process the reply * diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java index 54e73ddb2601..6f873e8dd3e4 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java @@ -135,6 +135,16 @@ public abstract class ReplyManagerSupport extends ServiceSupport implements Repl return correlationId; } + @Override + public void cancelCorrelationId(String correlationId) { + if (correlationId != null && correlation != null) { + ReplyHandler handler = correlation.remove(correlationId); + if (handler != null) { + log.debug("Cancelled reply correlation [{}]", correlationId); + } + } + } + protected abstract ReplyHandler createReplyHandler( ReplyManager replyManager, Exchange exchange, AsyncCallback callback, String originalCorrelationId, String correlationId, long requestTimeout); diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsInOutSendFailureCallbackTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsInOutSendFailureCallbackTest.java new file mode 100644 index 000000000000..e6ac4083d6f3 --- /dev/null +++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/JmsInOutSendFailureCallbackTest.java @@ -0,0 +1,150 @@ +/* + * 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; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Proxy; +import java.util.concurrent.TimeUnit; + +import jakarta.jms.Connection; +import jakarta.jms.ConnectionFactory; +import jakarta.jms.JMSException; +import jakarta.jms.MessageProducer; +import jakarta.jms.Session; + +import org.apache.camel.CamelContext; +import org.apache.camel.Exchange; +import org.apache.camel.ExchangePattern; +import org.apache.camel.ExchangeTimedOutException; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.test.infra.core.CamelContextExtension; +import org.apache.camel.test.infra.core.TransientCamelContextExtension; +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.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; + +/** + * Tests that when a JMS send fails after the reply correlation has been registered, the AsyncCallback is invoked + * exactly once (not a second time by the timeout handler). + * + * @see <a href="https://issues.apache.org/jira/browse/CAMEL-24073";>CAMEL-24073</a> + */ +public class JmsInOutSendFailureCallbackTest extends AbstractJMSTest { + + @Order(2) + @RegisterExtension + public static CamelContextExtension camelContextExtension = new TransientCamelContextExtension(); + + protected CamelContext context; + protected ProducerTemplate template; + + @Test + public void testCallbackInvokedOnceOnSendFailure() throws Exception { + Exchange result = template.send("direct:JmsInOutSendFailureCallbackTest", ExchangePattern.InOut, + p -> p.getIn().setBody("Hello")); + + assertNotNull(result.getException()); + assertFalse(result.getException() instanceof ExchangeTimedOutException, + "Should fail with JMS send exception, not ExchangeTimedOutException"); + + // wait past the requestTimeout (2s) and verify the timeout handler does not + // overwrite the exception with ExchangeTimedOutException via a second callback + Exception originalException = result.getException(); + await().during(3, TimeUnit.SECONDS) + .atMost(4, TimeUnit.SECONDS) + .untilAsserted(() -> assertSame(originalException, result.getException(), + "Exception changed after send failure - timeout handler fired a second callback")); + } + + @Override + public String getComponentName() { + return "activemq"; + } + + @Override + protected JmsComponent setupComponent( + CamelContext camelContext, ConnectionFactory connectionFactory, String componentName) { + ConnectionFactory failingCf = createFailingSendConnectionFactory(connectionFactory); + return jmsComponentAutoAcknowledge(failingCf); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:JmsInOutSendFailureCallbackTest") + .to(ExchangePattern.InOut, + "activemq:queue:JmsInOutSendFailureCallbackTest?requestTimeout=2000"); + } + }; + } + + @Override + public CamelContextExtension getCamelContextExtension() { + return camelContextExtension; + } + + @BeforeEach + void setUpRequirements() { + context = camelContextExtension.getContext(); + template = camelContextExtension.getProducerTemplate(); + } + + private static ConnectionFactory createFailingSendConnectionFactory(ConnectionFactory delegate) { + return proxyOf(ConnectionFactory.class, delegate, (proxy, method, args) -> { + Object result = method.invoke(delegate, args); + return result instanceof Connection conn ? wrapConnection(conn) : result; + }); + } + + private static Connection wrapConnection(Connection delegate) { + return proxyOf(Connection.class, delegate, (proxy, method, args) -> { + Object result = method.invoke(delegate, args); + return result instanceof Session session ? wrapSession(session) : result; + }); + } + + private static Session wrapSession(Session delegate) { + return proxyOf(Session.class, delegate, (proxy, method, args) -> { + Object result = method.invoke(delegate, args); + return result instanceof MessageProducer producer ? wrapProducer(producer) : result; + }); + } + + private static MessageProducer wrapProducer(MessageProducer delegate) { + return proxyOf(MessageProducer.class, delegate, (proxy, method, args) -> { + if ("send".equals(method.getName())) { + throw new JMSException("Simulated send failure: broker rejected message"); + } + return method.invoke(delegate, args); + }); + } + + @SuppressWarnings("unchecked") + private static <T> T proxyOf(Class<T> iface, T delegate, InvocationHandler handler) { + return (T) Proxy.newProxyInstance(iface.getClassLoader(), new Class<?>[] { iface }, handler); + } +}
