CAMEL-24401: Fix permanent temporary replyTo strand in camel-jms

components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultJmsMessageListenerContainer.java

@@ -161,4 +161,15 @@ protected void stopSharedConnection() {
         }
         super.stopSharedConnection();
     }
+
+    /**
+     * Forces the listener container to refresh its connection and recreate consumers, which will re-resolve the
+     * temporary reply destination. Used when a pending reply-destination refresh must be consumed but cached consumers
+     * would otherwise never call the destination resolver again.
+     */
+    public void recoverReplyDestinationAfterRefresh() {
+        if (isRunning() && !isRecovering()) {
+            recoverAfterListenerSetupFailure();
+        }
+    }
 }

components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java

@@ -16,7 +16,10 @@
  */
 package org.apache.camel.component.jms.reply;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -40,6 +43,7 @@
 import org.apache.camel.support.service.ServiceSupport;
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
 import org.springframework.jms.listener.DefaultMessageListenerContainer;
+import org.springframework.jms.listener.SimpleMessageListenerContainer;
 import org.springframework.jms.support.destination.DestinationResolver;
 
 /**
@@ -48,6 +52,8 @@
 public class TemporaryQueueReplyManager extends ReplyManagerSupport {
 
     final TemporaryReplyQueueDestinationResolver destinationResolver;
+    private ExecutorService refreshRecoveryExecutor;
+    private final AtomicBoolean refreshRecoveryScheduled = new AtomicBoolean();
 
     public TemporaryQueueReplyManager(CamelContext camelContext, TemporaryQueueResolver resolver) {
         super(camelContext);
@@ -57,9 +63,95 @@ public TemporaryQueueReplyManager(CamelContext camelContext, TemporaryQueueResol
     @Override
     protected void doStop() throws Exception {
         super.doStop();
+        if (refreshRecoveryExecutor != null) {
+            camelContext.getExecutorServiceManager().shutdownNow(refreshRecoveryExecutor);
+            refreshRecoveryExecutor = null;
+        }
         ServiceHelper.stopService(destinationResolver);
     }
 
+    private void triggerReplyDestinationRecovery() {
+        if (listenerContainer == null || isStopping() || isStopped()) {
+            return;
+        }
+        if (!destinationResolver.isRefreshPending()) {
+            return;
+        }
+        if (!refreshRecoveryScheduled.compareAndSet(false, true)) {
+            return;
+        }
+        try {
+            getRefreshRecoveryExecutor().execute(this::runReplyDestinationRecovery);
+        } catch (RejectedExecutionException e) {
+            refreshRecoveryScheduled.set(false);
+        }
+    }
+
+    private void runReplyDestinationRecovery() {
+        try {
+            long delay = endpoint.getRecoveryInterval() >= 0 ? endpoint.getRecoveryInterval() : 5000L;
+            if (!sleepQuietly(delay)) {
+                return;
+            }
+            int attempts = 0;
+            while (destinationResolver.isRefreshPending() && !isStopping() && !isStopped()
+                    && listenerContainer != null && listenerContainer.isRunning() && attempts < 20) {
+                try {
+                    if (listenerContainer instanceof DefaultJmsMessageListenerContainer dmlc) {
+                        if (dmlc.isRecovering()) {
+                            if (!sleepQuietly(delay)) {
+                                return;
+                            }
+                            attempts++;
+                            continue;
+                        }
+                        dmlc.recoverReplyDestinationAfterRefresh();
+                    } else if (listenerContainer instanceof SimpleMessageListenerContainer smlc) {
+                        smlc.stop();
+                        smlc.start();
+                    } else {
+                        listenerContainer.stop();
+                        listenerContainer.start();
+                    }
+                } catch (Exception e) {
+                    log.warn("Failed to trigger recovery of temporary reply destination on endpoint: {}",
+                            endpoint.getEndpointUri(), e);
+                }
+                if (!destinationResolver.isRefreshPending()) {
+                    break;
+                }
+                if (!sleepQuietly(delay)) {
+                    return;
+                }
+                attempts++;
+            }
+        } finally {
+            refreshRecoveryScheduled.set(false);
+            if (destinationResolver.isRefreshPending() && !isStopping() && !isStopped()
+                    && listenerContainer != null && listenerContainer.isRunning()) {
+                triggerReplyDestinationRecovery();
+            }
+        }
+    }
+
+    private boolean sleepQuietly(long millis) {
+        try {
+            Thread.sleep(millis);
+            return !isStopping() && !isStopped();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return false;
+        }
+    }
+
+    private ExecutorService getRefreshRecoveryExecutor() {
+        if (refreshRecoveryExecutor == null) {
+            String name = "JmsTemporaryReplyToRefresh[" + endpoint.getDestinationName() + "]";
+            refreshRecoveryExecutor = camelContext.getExecutorServiceManager().newSingleThreadExecutor(this, name);
+        }
+        return refreshRecoveryExecutor;
+    }
+
     @Override
     protected ReplyHandler createReplyHandler(
             ReplyManager replyManager, Exchange exchange, AsyncCallback callback,
@@ -283,51 +375,76 @@ final class TemporaryReplyQueueDestinationResolver extends ServiceSupport
         // 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 AtomicLong refreshGeneration = new AtomicLong();
+        private volatile long publishedGeneration;
         private final TemporaryQueueResolver custom;
 
         public TemporaryReplyQueueDestinationResolver(TemporaryQueueResolver custom) {
             this.custom = custom;
         }
 
+        boolean isRefreshPending() {
+            return refreshGeneration.get() != publishedGeneration;
+        }
+
         @Override
         public Destination resolveDestinationName(Session session, String destinationName, boolean pubSubDomain)
                 throws JMSException {
-            // 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
-                            }
+                TemporaryQueue answer = queue;
+                if (answer != null && !isRefreshPending()) {
+                    return answer;
+                }
+                long generationToHandle = refreshGeneration.get();
+                TemporaryQueue previousQueue = queue;
+                if (previousQueue != null) {
+                    try {
+                        if (custom != null) {
+                            custom.delete(previousQueue);
+                        } else {
+                            previousQueue.delete();
                         }
-                        queue = custom.createTemporaryQueue(session);
-                    } else {
-                        queue = session.createTemporaryQueue();
+                    } catch (Exception e) {
+                        // ignore
                     }
-                    setReplyTo(queue);
+                    queue = null;
+                }
+                TemporaryQueue refreshedQueue;
+                if (custom != null) {
+                    refreshedQueue = custom.createTemporaryQueue(session);
+                } else {
+                    refreshedQueue = session.createTemporaryQueue();
+                }
+                if (refreshGeneration.get() == generationToHandle) {
+                    queue = refreshedQueue;
+                    setReplyTo(refreshedQueue);
+                    publishedGeneration = generationToHandle;
                     if (log.isDebugEnabled()) {
-                        log.debug("Refreshed Temporary ReplyTo Queue. New queue: {}", queue.getQueueName());
+                        log.debug("Refreshed Temporary ReplyTo Queue. New queue: {}", refreshedQueue.getQueueName());
+                    }
+                    return refreshedQueue;
+                }
+                // a newer refresh was requested while creating the queue; discard this attempt
+                try {
+                    if (custom != null) {
+                        custom.delete(refreshedQueue);
+                    } else {
+                        refreshedQueue.delete();
                     }
+                } catch (Exception e) {
+                    // ignore
                 }
+                return null;
             } finally {
                 destinationLock.unlock();
             }
-            return queue;
         }
 
         public void scheduleRefresh() {
-            refreshWanted.set(true);
+            refreshGeneration.incrementAndGet();
             replyTo = null;
+            triggerReplyDestinationRecovery();
         }
 
         @Override
@@ -346,6 +463,7 @@ protected void doStop() throws Exception {
                     }
                     queue = null;
                 }
+                publishedGeneration = refreshGeneration.get();
             } finally {
                 destinationLock.unlock();
             }

components/camel-jms/src/test/java/org/apache/camel/component/jms/reply/JmsTemporaryReplyToRequestReplyIT.java

@@ -0,0 +1,68 @@
+/*
+ * 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.reply;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.jms.JmsComponent;
+import org.apache.camel.test.infra.artemis.common.ConnectionFactoryHelper;
+import org.apache.camel.test.infra.artemis.services.ArtemisService;
+import org.apache.camel.test.infra.artemis.services.ArtemisServiceFactory;
+import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
+import org.junit.jupiter.api.BeforeEach;
+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.assertj.core.api.Assertions.assertThat;
+
+class JmsTemporaryReplyToRequestReplyIT {
+
+    private static final String REQUEST_QUEUE = "JmsTemporaryReplyToRequestReplyIT.request";
+
+    @RegisterExtension
+    static ArtemisService service = ArtemisServiceFactory.createVMService();
+
+    @RegisterExtension
+    static DefaultCamelContextExtension contextExtension = new DefaultCamelContextExtension();
+
+    private ProducerTemplate template;
+
+    @BeforeEach
+    void setUp() throws Exception {
+        CamelContext context = contextExtension.getContext();
+        JmsComponent component = jmsComponentAutoAcknowledge(ConnectionFactoryHelper.createConnectionFactory(service));
+        context.addComponent("jms", component);
+        context.addRoutes(new RouteBuilder() {
+            @Override
+            public void configure() {
+                from("jms:queue:" + REQUEST_QUEUE).routeId("server")
+                        .transform(simple("echo:${body}"));
+            }
+        });
+        template = contextExtension.getProducerTemplate();
+    }
+
+    @Test
+    void shouldSupportConsecutiveTemporaryReplyRequests() {
+        assertThat(template.requestBody("jms:queue:" + REQUEST_QUEUE, "first", String.class))
+                .isEqualTo("echo:first");
+        assertThat(template.requestBody("jms:queue:" + REQUEST_QUEUE, "second", String.class))
+                .isEqualTo("echo:second");
+    }
+}