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 b7fee3af195a CAMEL-24070: camel-spring-rabbitmq - Fix thread-unsafe
lazy template creation
b7fee3af195a is described below
commit b7fee3af195a8b898e1963f861a17f161eb41465
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 15 17:10:01 2026 +0200
CAMEL-24070: camel-spring-rabbitmq - Fix thread-unsafe lazy template
creation
SpringRabbitMQProducer lazily creates its RabbitTemplate and
AsyncRabbitTemplate without synchronization. Concurrent first messages
can each create their own template instance; all but the last are leaked.
Additionally, inOutTemplate.start() was called on every invocation
instead of only at creation time.
Adds a ReentrantLock to guard getInOnlyTemplate(), getInOutTemplate(),
and doStop(), matching the existing pattern in EndpointMessageListener.
Also fixes Boolean.FALSE == sent identity comparison to !sent.
Closes #24727
Co-Authored-By: Claude Opus 4.6 <[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");
+ }
+}
