This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24070 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 3309ab82143abc890b438416ab688f236b14641f Author: Claus Ibsen <[email protected]> AuthorDate: Wed Jul 15 15:10:50 2026 +0200 CAMEL-24070: camel-spring-rabbitmq - Fix thread-unsafe lazy template creation in SpringRabbitMQProducer Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../springrabbit/SpringRabbitMQProducer.java | 48 +++++++++++----- .../SpringRabbitMQProducerThreadSafetyTest.java | 66 ++++++++++++++++++++++ 2 files changed, 99 insertions(+), 15 deletions(-) diff --git a/components/camel-spring-parent/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducer.java b/components/camel-spring-parent/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducer.java index e1252b4dcdc4..c4b21f688f28 100644 --- a/components/camel-spring-parent/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducer.java +++ b/components/camel-spring-parent/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducer.java @@ -19,6 +19,8 @@ package org.apache.camel.component.springrabbit; import java.util.Map; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.apache.camel.AsyncCallback; import org.apache.camel.Endpoint; @@ -41,6 +43,7 @@ public class SpringRabbitMQProducer extends DefaultAsyncProducer { private static final Logger LOG = LoggerFactory.getLogger(SpringRabbitMQProducer.class); + private final Lock lock = new ReentrantLock(); private RabbitTemplate inOnlyTemplate; private AsyncRabbitTemplate inOutTemplate; @@ -54,10 +57,15 @@ public class SpringRabbitMQProducer extends DefaultAsyncProducer { } public RabbitTemplate getInOnlyTemplate() { - if (inOnlyTemplate == null) { - inOnlyTemplate = getEndpoint().createInOnlyTemplate(); + lock.lock(); + try { + if (inOnlyTemplate == null) { + inOnlyTemplate = getEndpoint().createInOnlyTemplate(); + } + return inOnlyTemplate; + } finally { + lock.unlock(); } - return inOnlyTemplate; } public void setInOnlyTemplate(RabbitTemplate inOnlyTemplate) { @@ -65,11 +73,16 @@ public class SpringRabbitMQProducer extends DefaultAsyncProducer { } public AsyncRabbitTemplate getInOutTemplate() { - if (inOutTemplate == null) { - inOutTemplate = getEndpoint().createInOutTemplate(); + lock.lock(); + try { + if (inOutTemplate == null) { + inOutTemplate = getEndpoint().createInOutTemplate(); + inOutTemplate.start(); + } + return inOutTemplate; + } finally { + lock.unlock(); } - inOutTemplate.start(); - return inOutTemplate; } public void setInOutTemplate(AsyncRabbitTemplate inOutTemplate) { @@ -96,13 +109,18 @@ public class SpringRabbitMQProducer extends DefaultAsyncProducer { @Override protected void doStop() throws Exception { - if (inOnlyTemplate != null) { - inOnlyTemplate.stop(); - inOnlyTemplate = null; - } - if (inOutTemplate != null) { - inOutTemplate.stop(); - inOutTemplate = null; + lock.lock(); + try { + if (inOnlyTemplate != null) { + inOnlyTemplate.stop(); + inOnlyTemplate = null; + } + if (inOutTemplate != null) { + inOutTemplate.stop(); + inOutTemplate = null; + } + } finally { + lock.unlock(); } super.doStop(); } @@ -223,7 +241,7 @@ public class SpringRabbitMQProducer extends DefaultAsyncProducer { sent = true; } - if (Boolean.FALSE == sent) { + if (!sent) { exchange.setException(new TimeoutException("Message not sent within " + timeout + " millis")); } } catch (Exception e) { diff --git a/components/camel-spring-parent/camel-spring-rabbitmq/src/test/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducerThreadSafetyTest.java b/components/camel-spring-parent/camel-spring-rabbitmq/src/test/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducerThreadSafetyTest.java new file mode 100644 index 000000000000..59550a710b88 --- /dev/null +++ b/components/camel-spring-parent/camel-spring-rabbitmq/src/test/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducerThreadSafetyTest.java @@ -0,0 +1,66 @@ +/* + * 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.springrabbit; + +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class SpringRabbitMQProducerThreadSafetyTest { + + @Test + void concurrentGetInOnlyTemplateMustReturnSameInstance() throws Exception { + CachingConnectionFactory cf = new CachingConnectionFactory("localhost"); + SpringRabbitMQComponent component = new SpringRabbitMQComponent(); + SpringRabbitMQEndpoint endpoint = new SpringRabbitMQEndpoint( + "spring-rabbitmq:test", component, "test"); + endpoint.setConnectionFactory(cf); + SpringRabbitMQProducer producer = new SpringRabbitMQProducer(endpoint); + + int threads = 8; + CountDownLatch barrier = new CountDownLatch(threads); + Set<RabbitTemplate> observed = ConcurrentHashMap.newKeySet(); + ExecutorService pool = Executors.newFixedThreadPool(threads); + + for (int i = 0; i < threads; i++) { + pool.submit(() -> { + barrier.countDown(); + try { + barrier.await(5, TimeUnit.SECONDS); + } catch (Exception e) { + throw new RuntimeException(e); + } + observed.add(producer.getInOnlyTemplate()); + }); + } + + pool.shutdown(); + pool.awaitTermination(10, TimeUnit.SECONDS); + + assertEquals(1, observed.size(), + "concurrent getInOnlyTemplate() must always return the same template"); + } +}
