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 e1c7f27d0c8c CAMEL-24124: Drain pending reply handlers during shutdown
instead of silently discarding
e1c7f27d0c8c is described below
commit e1c7f27d0c8c4df8692cb0a871c16b3c15626bc8
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Jul 16 07:13:48 2026 +0200
CAMEL-24124: Drain pending reply handlers during shutdown instead of
silently discarding
Fix DefaultTimeoutMap to emit Evict events for remaining entries during
stop instead of silently calling map.clear(). This ensures reply-manager
listeners are notified. The drain-and-evict logic is exposed as a
protected method so only reply-manager subclasses (camel-jms, camel-sjms,
camel-rocketmq) opt in — preserving the aggregator's restart semantics.
Relax the isRunAllowed() guard in processReply() to allow reply
processing during the STOPPING phase. Replies arriving during shutdown
are processed normally; pending entries that never received a reply are
drained with RejectedExecutionException and their callbacks completed.
This fixes a 45-second graceful-shutdown stall caused by InOut exchanges
whose AsyncCallback was never completed when the reply manager stopped.
Closes #24742
Co-Authored-By: Claude Opus 4.6 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../component/jms/reply/CorrelationTimeoutMap.java | 6 ++
.../component/jms/reply/ReplyManagerSupport.java | 10 ++-
.../component/rocketmq/reply/ReplyTimeoutMap.java | 6 ++
.../reply/RocketMQReplyManagerSupport.java | 26 ++++--
.../sjms/reply/CorrelationTimeoutMap.java | 6 ++
.../component/sjms/reply/ReplyManagerSupport.java | 94 +++++++++++-----------
.../apache/camel/support/DefaultTimeoutMap.java | 22 +++++
7 files changed, 114 insertions(+), 56 deletions(-)
diff --git
a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
index 2908918c009c..efd5316f6ba2 100644
---
a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
+++
b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
@@ -57,6 +57,12 @@ class CorrelationTimeoutMap extends
DefaultTimeoutMap<String, ReplyHandler> {
}
}
+ @Override
+ protected void doStop() throws Exception {
+ drainAndEvictAll();
+ super.doStop();
+ }
+
@Override
public ReplyHandler put(String key, ReplyHandler value, long
timeoutMillis) {
return super.put(key, value, encode(timeoutMillis));
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/ReplyTimeoutMap.java
b/components/camel-rocketmq/src/main/java/org/apache/camel/component/rocketmq/reply/ReplyTimeoutMap.java
index b60d43038588..c567ef363b1b 100644
---
a/components/camel-rocketmq/src/main/java/org/apache/camel/component/rocketmq/reply/ReplyTimeoutMap.java
+++
b/components/camel-rocketmq/src/main/java/org/apache/camel/component/rocketmq/reply/ReplyTimeoutMap.java
@@ -53,6 +53,12 @@ public class ReplyTimeoutMap extends
DefaultTimeoutMap<String, ReplyHandler> {
}
}
+ @Override
+ protected void doStop() throws Exception {
+ drainAndEvictAll();
+ super.doStop();
+ }
+
@Override
public ReplyHandler put(String key, ReplyHandler value, long
timeoutMillis) {
return super.put(key, value, encode(timeoutMillis));
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/CorrelationTimeoutMap.java
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/CorrelationTimeoutMap.java
index 584c2c3bcc65..632a0413b67c 100644
---
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/CorrelationTimeoutMap.java
+++
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/CorrelationTimeoutMap.java
@@ -57,6 +57,12 @@ class CorrelationTimeoutMap extends
DefaultTimeoutMap<String, ReplyHandler> {
}
}
+ @Override
+ protected void doStop() throws Exception {
+ drainAndEvictAll();
+ super.doStop();
+ }
+
@Override
public ReplyHandler put(String key, ReplyHandler value, long
timeoutMillis) {
return super.put(key, value, encode(timeoutMillis));
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..e64f980078df 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
@@ -279,6 +279,28 @@ public class DefaultTimeoutMap<K, V> extends
ServiceSupport implements TimeoutMa
schedulePoll();
}
+ /**
+ * Drain all remaining entries from the map, emitting an {@link
Listener.Type#Evict} event for each one. Subclasses
+ * that need to notify listeners about leftover entries on shutdown (e.g.
reply-manager timeout maps) should call
+ * this from their {@link #doStop()} override before delegating to {@code
super.doStop()}.
+ */
+ protected void drainAndEvictAll() {
+ if (map.isEmpty()) {
+ return;
+ }
+ List<TimeoutMapEntry<K, V>> remaining;
+ 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());
+ }
+ }
+
@Override
protected void doStop() throws Exception {
if (future != null) {
