Copilot commented on code in PR #781: URL: https://github.com/apache/rocketmq-spring/pull/781#discussion_r3923650881
########## 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<String, FilterExpression> getSubscriptionExpressions() { + return Collections.emptyMap(); + } + + @Override + public List<MessageView> receive(int maxMessageNum, Duration invisibleDuration) throws ClientException { + if (closed) { + throw new IllegalStateException("consumer already closed"); + } + return Collections.emptyList(); + } + + @Override + public CompletableFuture<List<MessageView>> receiveAsync(int maxMessageNum, Duration invisibleDuration) { + receiveAsyncCount.incrementAndGet(); + CompletableFuture<List<MessageView>> 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<Void> ackAsync(MessageView messageView) { + return CompletableFuture.completedFuture(null); + } + + @Override + public void changeInvisibleDuration(MessageView messageView, Duration invisibleDuration) { + } + + @Override + public CompletableFuture<Void> 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<List<MessageView>> 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<List<MessageView>> second = template.receiveAsync(1, Duration.ofSeconds(1)); + assertEquals(2, consumer.receiveAsyncCount.get()); + assertNotNull(second.get()); Review Comment: Using `CompletableFuture#get()` without a timeout can cause the test to hang indefinitely if the future never completes (including in regressions or CI resource issues). Use a bounded wait (e.g., `get(timeout, unit)`) or a test-level timeout to ensure failures are reported promptly rather than stalling the test suite. ########## rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/core/RocketMQClientTemplate.java: ########## @@ -372,9 +372,11 @@ public List<MessageView> receive(int maxMessageNum, Duration invisibleDuration) public CompletableFuture<List<MessageView>> receiveAsync(int maxMessageNum, Duration invisibleDuration) throws ClientException, IOException { SimpleConsumer simpleConsumer = this.getSimpleConsumer(); - CompletableFuture<List<MessageView>> 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); Review Comment: Now that `receiveAsync` no longer closes the consumer, this method no longer appears to throw `IOException`. Keeping `throws IOException` in a public API forces callers to handle/declare an exception that can't occur here anymore. Consider removing `IOException` from the throws clause (and any related imports/docs) to keep the API accurate. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
