This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new e41ec4c77d60 CAMEL-24073: camel-jms - Cancel reply correlation on send
failure to prevent double callback (#24730)
e41ec4c77d60 is described below
commit e41ec4c77d600f2c6566135a76db9a3e9a307460
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 15 17:58:42 2026 +0200
CAMEL-24073: camel-jms - Cancel reply correlation on send failure to
prevent double callback (#24730)
When a JMS send fails after reply correlation is registered, the
correlation entry stays in the timeout map and fires a second
callback.done() at requestTimeout. Added cancelCorrelationId() to
ReplyManager to remove pending handlers on send failure, preventing
the double callback and the spurious ExchangeTimedOutException.
Co-Authored-By: Claude Opus 4.6 <[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);
+ }
+}
