You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/07/15 16:39:16 UTC
(camel) 01/01: CAMEL-24124: Drain pending reply handlers during shutdown instead of silently discarding
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch fix/CAMEL-24124
in repository https://gitbox.apache.org/repos/asf/camel.git
commit e05a3dd7b055887f818f3fb773485c220f3723b8
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Jul 15 18:38:59 2026 +0200
CAMEL-24124: Drain pending reply handlers during shutdown instead of silently discarding
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
Signed-off-by: Claus Ibsen <cl...@gmail.com>
---
.../component/jms/reply/ReplyManagerSupport.java | 10 ++-
.../reply/RocketMQReplyManagerSupport.java | 26 ++++--
.../component/sjms/reply/ReplyManagerSupport.java | 94 +++++++++++-----------
.../apache/camel/support/DefaultTimeoutMap.java | 16 +++-
4 files changed, 88 insertions(+), 58 deletions(-)
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 6f873e8dd3e4..870ad77d7c90 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
@@ -19,6 +19,7 @@ package org.apache.camel.component.jms.reply;
import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import jakarta.jms.Destination;
@@ -176,13 +177,18 @@ public abstract class ReplyManagerSupport extends ServiceSupport implements Repl
@Override
public void processReply(ReplyHolder holder) {
- if (holder == null || !isRunAllowed()) {
+ if (holder == null) {
return;
}
try {
Exchange exchange = holder.getExchange();
if (holder.isTimeout()) {
- handleTimeout(holder, exchange);
+ if (isStopping() || isStopped()) {
+ exchange.setException(new RejectedExecutionException(
+ "Cannot process JMS reply as the reply manager is shutting down"));
+ } else {
+ handleTimeout(holder, exchange);
+ }
} else {
handleSuccessfulReply(holder, exchange);
}
diff --git a/components/camel-rocketmq/src/main/java/org/apache/camel/component/rocketmq/reply/RocketMQReplyManagerSupport.java b/components/camel-rocketmq/src/main/java/org/apache/camel/component/rocketmq/reply/RocketMQReplyManagerSupport.java
index ec2b025d5d82..ed217441c71e 100644
--- a/components/camel-rocketmq/src/main/java/org/apache/camel/component/rocketmq/reply/RocketMQReplyManagerSupport.java
+++ b/components/camel-rocketmq/src/main/java/org/apache/camel/component/rocketmq/reply/RocketMQReplyManagerSupport.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.rocketmq.reply;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import org.apache.camel.AsyncCallback;
@@ -148,20 +149,18 @@ public class RocketMQReplyManagerSupport extends ServiceSupport implements Reply
@Override
public void processReply(ReplyHolder holder) {
- if (!isRunAllowed()) {
+ if (holder == null) {
return;
}
try {
Exchange exchange = holder.getExchange();
if (holder.isTimeout()) {
- if (log.isWarnEnabled()) {
- log.warn("Timeout occurred after {} millis waiting for reply message with messageKey [{}] on topic {}." +
- " Setting ExchangeTimedOutException on {} and continue routing.",
- holder.getTimeout(), holder.getMessageKey(), replyToTopic, ExchangeHelper.logIds(exchange));
+ if (isStopping() || isStopped()) {
+ exchange.setException(new RejectedExecutionException(
+ "Cannot process reply as the reply manager is shutting down"));
+ } else {
+ handleTimeout(holder, exchange);
}
- String msg = "reply message with messageKey: " + holder.getMessageKey() + " not received on topic: "
- + replyToTopic;
- exchange.setException(new ExchangeTimedOutException(exchange, holder.getTimeout(), msg));
} else {
processReceivedReply(holder);
}
@@ -170,6 +169,17 @@ public class RocketMQReplyManagerSupport extends ServiceSupport implements Reply
}
}
+ private void handleTimeout(ReplyHolder holder, Exchange exchange) {
+ if (log.isWarnEnabled()) {
+ log.warn("Timeout occurred after {} millis waiting for reply message with messageKey [{}] on topic {}." +
+ " Setting ExchangeTimedOutException on {} and continue routing.",
+ holder.getTimeout(), holder.getMessageKey(), replyToTopic, ExchangeHelper.logIds(exchange));
+ }
+ String msg = "reply message with messageKey: " + holder.getMessageKey() + " not received on topic: "
+ + replyToTopic;
+ exchange.setException(new ExchangeTimedOutException(exchange, holder.getTimeout(), msg));
+ }
+
private static void processReceivedReply(ReplyHolder holder) {
Message message = holder.getExchange().getOut();
MessageExt messageExt = holder.getMessageExt();
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/ReplyManagerSupport.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/ReplyManagerSupport.java
index 3c2603cfb435..e2800ee1fcde 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/ReplyManagerSupport.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/ReplyManagerSupport.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.sjms.reply;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -143,55 +144,56 @@ public abstract class ReplyManagerSupport extends ServiceSupport implements Repl
@Override
public void processReply(ReplyHolder holder) {
- if (holder != null && isRunAllowed()) {
- try {
- Exchange exchange = holder.getExchange();
-
- boolean timeout = holder.isTimeout();
- if (timeout) {
- // timeout occurred do a WARN log so its easier to spot in the logs
- if (log.isWarnEnabled()) {
- log.warn(
- "Timeout occurred after {} millis waiting for reply message with correlationID [{}] on destination {}."
- + " Setting ExchangeTimedOutException on {} and continue routing.",
- holder.getRequestTimeout(), holder.getCorrelationId(), replyTo,
- ExchangeHelper.logIds(exchange));
- }
-
- // no response, so lets set a timed out exception
- String msg = "reply message with correlationID: " + holder.getCorrelationId()
- + " not received on destination: " + replyTo;
- exchange.setException(new ExchangeTimedOutException(exchange, holder.getRequestTimeout(), msg));
+ if (holder == null) {
+ return;
+ }
+ try {
+ Exchange exchange = holder.getExchange();
+ if (holder.isTimeout()) {
+ if (isStopping() || isStopped()) {
+ exchange.setException(new RejectedExecutionException(
+ "Cannot process JMS reply as the reply manager is shutting down"));
} else {
- Message message = holder.getMessage();
- Session session = holder.getSession();
- SjmsMessage response = new SjmsMessage(exchange, message, session, endpoint.getBinding());
- // the JmsBinding is designed to be "pull-based": it will populate the Camel message on demand
- // therefore, we link Exchange and OUT message before continuing, so that the JmsBinding has full access
- // to everything it may need, and can populate headers, properties, etc. accordingly (solves CAMEL-6218).
- exchange.setOut(response);
- Object body = response.getBody();
-
- if (endpoint.isTransferException() && body instanceof Exception exception) {
- log.debug("Reply was an Exception. Setting the Exception on the Exchange: {}", body);
- // we got an exception back and endpoint was configured to transfer exception
- // therefore set response as exception
- exchange.setException(exception);
- } else {
- log.debug("Reply received. OUT message body set to reply payload: {}", body);
- }
-
- // restore correlation id in case the remote server messed with it
- if (holder.getOriginalCorrelationId() != null) {
- JmsMessageHelper.setCorrelationId(message, holder.getOriginalCorrelationId());
- exchange.getOut().setHeader(SjmsConstants.JMS_CORRELATION_ID, holder.getOriginalCorrelationId());
- }
+ handleTimeout(holder, exchange);
}
- } finally {
- // notify callback
- AsyncCallback callback = holder.getCallback();
- callback.done(false);
+ } else {
+ handleSuccessfulReply(holder, exchange);
}
+ } finally {
+ holder.getCallback().done(false);
+ }
+ }
+
+ private void handleTimeout(ReplyHolder holder, Exchange exchange) {
+ if (log.isWarnEnabled()) {
+ log.warn(
+ "Timeout occurred after {} millis waiting for reply message with correlationID [{}] on destination {}."
+ + " Setting ExchangeTimedOutException on {} and continue routing.",
+ holder.getRequestTimeout(), holder.getCorrelationId(), replyTo,
+ ExchangeHelper.logIds(exchange));
+ }
+ String msg = "reply message with correlationID: " + holder.getCorrelationId()
+ + " not received on destination: " + replyTo;
+ exchange.setException(new ExchangeTimedOutException(exchange, holder.getRequestTimeout(), msg));
+ }
+
+ private void handleSuccessfulReply(ReplyHolder holder, Exchange exchange) {
+ Message message = holder.getMessage();
+ Session session = holder.getSession();
+ SjmsMessage response = new SjmsMessage(exchange, message, session, endpoint.getBinding());
+ exchange.setOut(response);
+ Object body = response.getBody();
+
+ if (endpoint.isTransferException() && body instanceof Exception exception) {
+ log.debug("Reply was an Exception. Setting the Exception on the Exchange: {}", body);
+ exchange.setException(exception);
+ } else {
+ log.debug("Reply received. OUT message body set to reply payload: {}", body);
+ }
+
+ if (holder.getOriginalCorrelationId() != null) {
+ JmsMessageHelper.setCorrelationId(message, holder.getOriginalCorrelationId());
+ exchange.getOut().setHeader(SjmsConstants.JMS_CORRELATION_ID, holder.getOriginalCorrelationId());
}
}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java b/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
index b6cda4f3eccd..50c4aa27c5c3 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
@@ -285,8 +285,20 @@ public class DefaultTimeoutMap<K, V> extends ServiceSupport implements TimeoutMa
future.cancel(false);
future = null;
}
- // clear map if we stop
- map.clear();
+ // drain remaining entries, emitting Evict events so listeners can handle cleanup
+ List<TimeoutMapEntry<K, V>> remaining = List.of();
+ if (!map.isEmpty()) {
+ lock.lock();
+ try {
+ remaining = new ArrayList<>(map.values());
+ map.clear();
+ } finally {
+ lock.unlock();
+ }
+ }
+ for (TimeoutMapEntry<K, V> entry : remaining) {
+ emitEvent(Evict, entry.getKey(), entry.getValue());
+ }
}
}