diff --git a/rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/core/RocketMQClientTemplate.java b/rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/core/RocketMQClientTemplate.java index 4213cb4b..929252ac 100644 --- a/rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/core/RocketMQClientTemplate.java +++ b/rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/core/RocketMQClientTemplate.java @@ -372,9 +372,11 @@ public List receive(int maxMessageNum, Duration invisibleDuration) public CompletableFuture> receiveAsync(int maxMessageNum, Duration invisibleDuration) throws ClientException, IOException { SimpleConsumer simpleConsumer = this.getSimpleConsumer(); - CompletableFuture> listCompletableFuture = simpleConsumer.receiveAsync(maxMessageNum, invisibleDuration); - simpleConsumer.close(); - return listCompletableFuture; + // Do not close the shared SimpleConsumer here: it is a reusable singleton whose lifecycle + // is managed by destroy(). Closing it right after starting the async receive aborts the + // in-flight future and leaves the (non-null) consumer closed, breaking every subsequent + // receive/ack/receiveAsync call on this template. + return simpleConsumer.receiveAsync(maxMessageNum, invisibleDuration); } diff --git a/rocketmq-v5-client-spring-boot/src/test/java/org/apache/rocketmq/client/core/RocketMQClientTemplateReceiveAsyncTest.java b/rocketmq-v5-client-spring-boot/src/test/java/org/apache/rocketmq/client/core/RocketMQClientTemplateReceiveAsyncTest.java new file mode 100644 index 00000000..e4086ce3 --- /dev/null +++ b/rocketmq-v5-client-spring-boot/src/test/java/org/apache/rocketmq/client/core/RocketMQClientTemplateReceiveAsyncTest.java @@ -0,0 +1,140 @@ +/* + * 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.rocketmq.client.core; + +import org.apache.rocketmq.client.apis.ClientException; +import org.apache.rocketmq.client.apis.consumer.FilterExpression; +import org.apache.rocketmq.client.apis.consumer.SimpleConsumer; +import org.apache.rocketmq.client.apis.message.MessageView; +import org.junit.Test; + +import java.io.IOException; +import java.time.Duration; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; + +public class RocketMQClientTemplateReceiveAsyncTest { + + /** + * Minimal fake SimpleConsumer that records how many times close()/receiveAsync() were invoked, + * and rejects operations once it has been closed (mirroring a real closed consumer). + */ + static class RecordingSimpleConsumer implements SimpleConsumer { + final AtomicInteger closeCount = new AtomicInteger(); + final AtomicInteger receiveAsyncCount = new AtomicInteger(); + volatile boolean closed = false; + + @Override + public String getConsumerGroup() { + return "test-group"; + } + + @Override + public SimpleConsumer subscribe(String topic, FilterExpression filterExpression) { + return this; + } + + @Override + public SimpleConsumer unsubscribe(String topic) { + return this; + } + + @Override + public Map getSubscriptionExpressions() { + return Collections.emptyMap(); + } + + @Override + public List receive(int maxMessageNum, Duration invisibleDuration) throws ClientException { + if (closed) { + throw new IllegalStateException("consumer already closed"); + } + return Collections.emptyList(); + } + + @Override + public CompletableFuture> receiveAsync(int maxMessageNum, Duration invisibleDuration) { + receiveAsyncCount.incrementAndGet(); + CompletableFuture> future = new CompletableFuture<>(); + if (closed) { + future.completeExceptionally(new IllegalStateException("consumer already closed")); + } else { + future.complete(Collections.emptyList()); + } + return future; + } + + @Override + public void ack(MessageView messageView) { + } + + @Override + public CompletableFuture ackAsync(MessageView messageView) { + return CompletableFuture.completedFuture(null); + } + + @Override + public void changeInvisibleDuration(MessageView messageView, Duration invisibleDuration) { + } + + @Override + public CompletableFuture changeInvisibleDurationAsync(MessageView messageView, Duration invisibleDuration) { + return CompletableFuture.completedFuture(null); + } + + @Override + public void close() throws IOException { + closeCount.incrementAndGet(); + closed = true; + } + } + + /** + * receiveAsync() must not close the shared, reusable SimpleConsumer: its lifecycle is owned by + * destroy(). Before the fix, receiveAsync() called simpleConsumer.close() right after starting + * the async receive, which aborted the in-flight future and left the (still non-null) consumer + * closed, breaking every subsequent receive/ack/receiveAsync on the template. + */ + @Test + public void receiveAsyncShouldNotCloseSharedConsumer() throws Exception { + RocketMQClientTemplate template = new RocketMQClientTemplate(); + RecordingSimpleConsumer consumer = new RecordingSimpleConsumer(); + template.setSimpleConsumer(consumer); + + CompletableFuture> future = template.receiveAsync(1, Duration.ofSeconds(1)); + + assertNotNull(future); + assertEquals("receiveAsync should start exactly one async receive", 1, consumer.receiveAsyncCount.get()); + assertEquals("receiveAsync must not close the shared consumer", 0, consumer.closeCount.get()); + assertFalse("shared consumer must stay open after receiveAsync", consumer.closed); + // Future must complete normally (not aborted by a premature close()). + assertNotNull(future.get()); + + // The shared consumer must remain usable for subsequent calls. + CompletableFuture> second = template.receiveAsync(1, Duration.ofSeconds(1)); + assertEquals(2, consumer.receiveAsyncCount.get()); + assertNotNull(second.get()); + assertEquals("consumer must never be closed by receiveAsync", 0, consumer.closeCount.get()); + } +}