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 f1bc42bd89b5 CAMEL-24074: camel-jms - Synchronize
TemporaryQueueReplyManager destination resolution (#24731)
f1bc42bd89b5 is described below
commit f1bc42bd89b5da7549ea968967371f730825fb6d
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 15 17:59:59 2026 +0200
CAMEL-24074: camel-jms - Synchronize TemporaryQueueReplyManager destination
resolution (#24731)
Fix race condition where concurrent listener threads could each create
separate temporary queues in TemporaryReplyQueueDestinationResolver,
causing replies to be black-holed. Applied ReentrantLock + volatile
double-checked locking pattern consistent with QueueReplyManager.
Uses a dedicated lock to preserve the CAMEL-20769 shutdown-deadlock fix.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../jms/reply/TemporaryQueueReplyManager.java | 77 ++++++++++++++--------
1 file changed, 49 insertions(+), 28 deletions(-)
diff --git
a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java
b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java
index 6f223ba98da2..6f5b4ef9275f 100644
---
a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java
+++
b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java
@@ -17,6 +17,8 @@
package org.apache.camel.component.jms.reply;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import jakarta.jms.Destination;
import jakarta.jms.ExceptionListener;
@@ -272,9 +274,15 @@ public class TemporaryQueueReplyManager extends
ReplyManagerSupport {
}
- private final class TemporaryReplyQueueDestinationResolver extends
ServiceSupport
+ final class TemporaryReplyQueueDestinationResolver extends ServiceSupport
implements DestinationResolver, NonManagedService {
- private TemporaryQueue queue;
+ // Use a dedicated lock instead of BaseService.lock to avoid deadlock
+ // during shutdown: BaseService.stop() holds its lock while calling
+ // doStop() -> listenerContainer.destroy() -> doShutdown() which waits
+ // for listener threads to finish. If a listener thread needs to
resolve
+ // the destination, it would deadlock trying to acquire
BaseService.lock.
+ private final Lock destinationLock = new ReentrantLock();
+ private volatile TemporaryQueue queue;
private final AtomicBoolean refreshWanted = new AtomicBoolean();
private final TemporaryQueueResolver custom;
@@ -285,26 +293,34 @@ public class TemporaryQueueReplyManager extends
ReplyManagerSupport {
@Override
public Destination resolveDestinationName(Session session, String
destinationName, boolean pubSubDomain)
throws JMSException {
- // use a temporary queue to gather the reply message
- if (queue == null || refreshWanted.get()) {
- refreshWanted.set(false);
- if (custom != null) {
- if (queue != null) {
- // delete previous queue
- try {
- custom.delete(queue);
- } catch (Exception e) {
- // ignore
+ // fast path: queue already resolved and no refresh needed
+ TemporaryQueue answer = queue;
+ if (answer != null && !refreshWanted.get()) {
+ return answer;
+ }
+ destinationLock.lock();
+ try {
+ if (queue == null || refreshWanted.get()) {
+ refreshWanted.set(false);
+ if (custom != null) {
+ if (queue != null) {
+ try {
+ custom.delete(queue);
+ } catch (Exception e) {
+ // ignore
+ }
}
+ queue = custom.createTemporaryQueue(session);
+ } else {
+ queue = session.createTemporaryQueue();
+ }
+ setReplyTo(queue);
+ if (log.isDebugEnabled()) {
+ log.debug("Refreshed Temporary ReplyTo Queue. New
queue: {}", queue.getQueueName());
}
- queue = custom.createTemporaryQueue(session);
- } else {
- queue = session.createTemporaryQueue();
- }
- setReplyTo(queue);
- if (log.isDebugEnabled()) {
- log.debug("Refreshed Temporary ReplyTo Queue. New queue:
{}", queue.getQueueName());
}
+ } finally {
+ destinationLock.unlock();
}
return queue;
}
@@ -316,17 +332,22 @@ public class TemporaryQueueReplyManager extends
ReplyManagerSupport {
@Override
protected void doStop() throws Exception {
- if (queue != null) {
- try {
- if (custom != null) {
- custom.delete(queue);
- } else {
- queue.delete();
+ destinationLock.lock();
+ try {
+ if (queue != null) {
+ try {
+ if (custom != null) {
+ custom.delete(queue);
+ } else {
+ queue.delete();
+ }
+ } catch (Exception e) {
+ // ignore
}
- } catch (Exception e) {
- // ignore
+ queue = null;
}
- queue = null;
+ } finally {
+ destinationLock.unlock();
}
}
}
