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 47620fdc809e CAMEL-25298: camel-rocketmq - complete an InOut exchange
once when the reply races with the timeout or a duplicate (#27339)
47620fdc809e is described below
commit 47620fdc809ed24cdf39d186df138fbab54ad1b4
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 5 14:10:24 2026 +0530
CAMEL-25298: camel-rocketmq - complete an InOut exchange once when the
reply races with the timeout or a duplicate (#27339)
RocketMQReplyManagerSupport.handleReplyMessage looked up the reply handler
with
get() and removed it afterwards. The request timeout (the timeout map
evicts the
entry and calls onTimeout) or a second copy of the reply (RocketMQ delivers
at
least once, and the reply consumer has several threads) could find the same
handler in between, so both completed the exchange: the exchange got the
reply
after the ExchangeTimedOutException, and the rest of the route ran twice.
The reply consumer now removes the handler in one step and only completes
the
exchange when it removed it, as camel-jms and camel-sjms do.
cancelMessageKey also removes in one step.
Found with a TLA+ model of the request-reply correlation (invariant: the
exchange is completed at most once).
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../reply/RocketMQReplyManagerSupport.java | 11 +-
.../reply/RocketMQReplyManagerSupportTest.java | 157 +++++++++++++++++++++
2 files changed, 162 insertions(+), 6 deletions(-)
diff --git
a/components/camel-rocketmq/src/main/java/org/apache/camel/component/rocketmq/reply/RocketMQReplyManagerSupport.java
b/components/camel-rocketmq/src/main/java/org/apache/camel/component/rocketmq/reply/RocketMQReplyManagerSupport.java
index ed217441c71e..2aaf29906e2a 100644
---
a/components/camel-rocketmq/src/main/java/org/apache/camel/component/rocketmq/reply/RocketMQReplyManagerSupport.java
+++
b/components/camel-rocketmq/src/main/java/org/apache/camel/component/rocketmq/reply/RocketMQReplyManagerSupport.java
@@ -189,17 +189,16 @@ public class RocketMQReplyManagerSupport extends
ServiceSupport implements Reply
@Override
public void cancelMessageKey(String messageKey) {
- if (null == timeoutMap.get(messageKey)) {
- return;
+ if (timeoutMap.remove(messageKey) != null) {
+ log.warn("Cancelled messageKey: {}", messageKey);
}
- log.warn("Cancelling messageKey: {}", messageKey);
- timeoutMap.remove(messageKey);
}
protected void handleReplyMessage(String messageKey, MessageExt
messageExt) {
- ReplyHandler handler = timeoutMap.get(messageKey);
+ // remove the handler in one step, so only one of the reply, a
duplicate of the reply, and the timeout
+ // completes the exchange
+ ReplyHandler handler = timeoutMap.remove(messageKey);
if (handler != null) {
- timeoutMap.remove(messageKey);
handler.onReply(messageKey, messageExt);
} else {
log.warn("Reply received for unknown messageKey [{}]. The message
will be ignored: {}", messageKey, messageExt);
diff --git
a/components/camel-rocketmq/src/test/java/org/apache/camel/component/rocketmq/reply/RocketMQReplyManagerSupportTest.java
b/components/camel-rocketmq/src/test/java/org/apache/camel/component/rocketmq/reply/RocketMQReplyManagerSupportTest.java
new file mode 100644
index 000000000000..14b7d83d8332
--- /dev/null
+++
b/components/camel-rocketmq/src/test/java/org/apache/camel/component/rocketmq/reply/RocketMQReplyManagerSupportTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.rocketmq.reply;
+
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangeTimedOutException;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * The reply consumer, the request timeout and a duplicate of the reply may
each try to complete an InOut exchange: only
+ * one of them may complete it.
+ */
+public class RocketMQReplyManagerSupportTest {
+
+ private static final String KEY = "camel-rocketmq-test-key";
+
+ private final AtomicInteger completions = new AtomicInteger();
+ private CamelContext context;
+ private ScheduledExecutorService executorService;
+ private TestReplyManager replyManager;
+ private Exchange exchange;
+
+ @BeforeEach
+ public void setUp() {
+ context = new DefaultCamelContext();
+ context.start();
+ executorService = Executors.newSingleThreadScheduledExecutor();
+ replyManager = new TestReplyManager(context, executorService);
+ ServiceHelper.startService(replyManager);
+ exchange = new DefaultExchange(context);
+ replyManager.registerReply(replyManager, exchange, doneSync ->
completions.incrementAndGet(), KEY, 10000);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ ServiceHelper.stopService(replyManager);
+ executorService.shutdownNow();
+ context.stop();
+ }
+
+ @Test
+ public void testReply() {
+ replyManager.handleReplyMessage(KEY, reply("Bye World"));
+
+ assertEquals(1, completions.get());
+ assertNull(exchange.getException());
+ assertEquals("Bye World", exchange.getMessage().getBody(String.class));
+ }
+
+ @Test
+ public void testTimeoutWhileReplyIsHandled() {
+ // the request times out after the reply consumer found the handler of
the reply
+ replyManager.testTimeoutMap.beforeRemove = () ->
replyManager.testTimeoutMap.timeout();
+
+ replyManager.handleReplyMessage(KEY, reply("Bye World"));
+
+ assertEquals(1, completions.get(), "The exchange should be completed
once");
+ assertInstanceOf(ExchangeTimedOutException.class,
exchange.getException());
+ }
+
+ @Test
+ public void testDuplicateReply() {
+ // a duplicate of the reply is handled by another consumer thread at
the same time
+ replyManager.testTimeoutMap.beforeRemove = () ->
replyManager.handleReplyMessage(KEY, reply("Bye World"));
+
+ replyManager.handleReplyMessage(KEY, reply("Bye World"));
+
+ assertEquals(1, completions.get(), "The exchange should be completed
once");
+ assertEquals("Bye World", exchange.getMessage().getBody(String.class));
+ }
+
+ private static MessageExt reply(String body) {
+ MessageExt messageExt = new MessageExt();
+ messageExt.setBody(body.getBytes(StandardCharsets.UTF_8));
+ return messageExt;
+ }
+
+ private static final class TestReplyManager extends
RocketMQReplyManagerSupport {
+
+ private TestTimeoutMap testTimeoutMap;
+
+ TestReplyManager(CamelContext camelContext, ScheduledExecutorService
executorService) {
+ super(camelContext);
+ setScheduledExecutorService(executorService);
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ // no consumer of the reply topic (it needs a broker): the test
hands the replies over itself
+ testTimeoutMap = new TestTimeoutMap(executorService);
+ timeoutMap = testTimeoutMap;
+ ServiceHelper.startService(timeoutMap);
+ }
+ }
+
+ private static final class TestTimeoutMap extends ReplyTimeoutMap {
+
+ private final AtomicLong now = new AtomicLong(1000);
+ private Runnable beforeRemove;
+
+ TestTimeoutMap(ScheduledExecutorService executorService) {
+ // the purge task does not run during the test (the test lets the
request time out itself)
+ super(executorService, 3600000);
+ }
+
+ @Override
+ protected long currentTime() {
+ return now.get();
+ }
+
+ @Override
+ public ReplyHandler remove(String key) {
+ Runnable task = beforeRemove;
+ beforeRemove = null;
+ if (task != null) {
+ task.run();
+ }
+ return super.remove(key);
+ }
+
+ void timeout() {
+ now.addAndGet(60000);
+ purge();
+ }
+ }
+}
