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 <[email protected]> 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 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../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()); + } } }
