Skip to content

Commit b7fee3a

Browse files
davsclausclaude
andauthored
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 <noreply@anthropic.com>
1 parent b163e65 commit b7fee3a

2 files changed

Lines changed: 99 additions & 15 deletions

File tree

components/camel-spring-parent/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducer.java

Lines changed: 33 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@
1919
import java.util.Map;
2020
import java.util.concurrent.RejectedExecutionException;
2121
import java.util.concurrent.TimeoutException;
22+
import java.util.concurrent.locks.Lock;
23+
import java.util.concurrent.locks.ReentrantLock;
2224

2325
import org.apache.camel.AsyncCallback;
2426
import org.apache.camel.Endpoint;
@@ -41,6 +43,7 @@ public class SpringRabbitMQProducer extends DefaultAsyncProducer {
4143

4244
private static final Logger LOG = LoggerFactory.getLogger(SpringRabbitMQProducer.class);
4345

46+
private final Lock lock = new ReentrantLock();
4447
private RabbitTemplate inOnlyTemplate;
4548
private AsyncRabbitTemplate inOutTemplate;
4649

@@ -54,22 +57,32 @@ public SpringRabbitMQEndpoint getEndpoint() {
5457
}
5558

5659
public RabbitTemplate getInOnlyTemplate() {
57-
if (inOnlyTemplate == null) {
58-
inOnlyTemplate = getEndpoint().createInOnlyTemplate();
60+
lock.lock();
61+
try {
62+
if (inOnlyTemplate == null) {
63+
inOnlyTemplate = getEndpoint().createInOnlyTemplate();
64+
}
65+
return inOnlyTemplate;
66+
} finally {
67+
lock.unlock();
5968
}
60-
return inOnlyTemplate;
6169
}
6270

6371
public void setInOnlyTemplate(RabbitTemplate inOnlyTemplate) {
6472
this.inOnlyTemplate = inOnlyTemplate;
6573
}
6674

6775
public AsyncRabbitTemplate getInOutTemplate() {
68-
if (inOutTemplate == null) {
69-
inOutTemplate = getEndpoint().createInOutTemplate();
76+
lock.lock();
77+
try {
78+
if (inOutTemplate == null) {
79+
inOutTemplate = getEndpoint().createInOutTemplate();
80+
inOutTemplate.start();
81+
}
82+
return inOutTemplate;
83+
} finally {
84+
lock.unlock();
7085
}
71-
inOutTemplate.start();
72-
return inOutTemplate;
7386
}
7487

7588
public void setInOutTemplate(AsyncRabbitTemplate inOutTemplate) {
@@ -96,13 +109,18 @@ protected void doStart() throws Exception {
96109

97110
@Override
98111
protected void doStop() throws Exception {
99-
if (inOnlyTemplate != null) {
100-
inOnlyTemplate.stop();
101-
inOnlyTemplate = null;
102-
}
103-
if (inOutTemplate != null) {
104-
inOutTemplate.stop();
105-
inOutTemplate = null;
112+
lock.lock();
113+
try {
114+
if (inOnlyTemplate != null) {
115+
inOnlyTemplate.stop();
116+
inOnlyTemplate = null;
117+
}
118+
if (inOutTemplate != null) {
119+
inOutTemplate.stop();
120+
inOutTemplate = null;
121+
}
122+
} finally {
123+
lock.unlock();
106124
}
107125
super.doStop();
108126
}
@@ -223,7 +241,7 @@ protected boolean processInOnly(Exchange exchange, AsyncCallback callback) {
223241
sent = true;
224242
}
225243

226-
if (Boolean.FALSE == sent) {
244+
if (!sent) {
227245
exchange.setException(new TimeoutException("Message not sent within " + timeout + " millis"));
228246
}
229247
} catch (Exception e) {
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.camel.component.springrabbit;
18+
19+
import java.util.Set;
20+
import java.util.concurrent.ConcurrentHashMap;
21+
import java.util.concurrent.CountDownLatch;
22+
import java.util.concurrent.ExecutorService;
23+
import java.util.concurrent.Executors;
24+
import java.util.concurrent.TimeUnit;
25+
26+
import org.junit.jupiter.api.Test;
27+
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
28+
import org.springframework.amqp.rabbit.core.RabbitTemplate;
29+
30+
import static org.junit.jupiter.api.Assertions.assertEquals;
31+
32+
public class SpringRabbitMQProducerThreadSafetyTest {
33+
34+
@Test
35+
void concurrentGetInOnlyTemplateMustReturnSameInstance() throws Exception {
36+
CachingConnectionFactory cf = new CachingConnectionFactory("localhost");
37+
SpringRabbitMQComponent component = new SpringRabbitMQComponent();
38+
SpringRabbitMQEndpoint endpoint = new SpringRabbitMQEndpoint(
39+
"spring-rabbitmq:test", component, "test");
40+
endpoint.setConnectionFactory(cf);
41+
SpringRabbitMQProducer producer = new SpringRabbitMQProducer(endpoint);
42+
43+
int threads = 8;
44+
CountDownLatch barrier = new CountDownLatch(threads);
45+
Set<RabbitTemplate> observed = ConcurrentHashMap.newKeySet();
46+
ExecutorService pool = Executors.newFixedThreadPool(threads);
47+
48+
for (int i = 0; i < threads; i++) {
49+
pool.submit(() -> {
50+
barrier.countDown();
51+
try {
52+
barrier.await(5, TimeUnit.SECONDS);
53+
} catch (Exception e) {
54+
throw new RuntimeException(e);
55+
}
56+
observed.add(producer.getInOnlyTemplate());
57+
});
58+
}
59+
60+
pool.shutdown();
61+
pool.awaitTermination(10, TimeUnit.SECONDS);
62+
63+
assertEquals(1, observed.size(),
64+
"concurrent getInOnlyTemplate() must always return the same template");
65+
}
66+
}

0 commit comments

Comments
 (0)