[ISSUE #780] Stop closing the shared SimpleConsumer in RocketMQClientTemplate.receiveAsync - #781
Conversation
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
LGTM. Critical bug fix — receiveAsync was prematurely closing the shared SimpleConsumer singleton, aborting the in-flight future and breaking all subsequent operations. Removing the close() call is the correct fix since the consumer lifecycle is owned by destroy().
The regression test is excellent: the RecordingSimpleConsumer fake directly verifies that close() is never invoked by receiveAsync, and that the consumer remains usable for subsequent calls. The test would have caught the original bug (expected:<0> but was:<1>).
Automated review by github-manager-bot
There was a problem hiding this comment.
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Pull request overview
Fixes a lifecycle bug where RocketMQClientTemplate.receiveAsync() closed the shared SimpleConsumer immediately after starting an async receive, breaking the returned future and all subsequent consumer operations on the same template.
Changes:
- Removed the
simpleConsumer.close()call fromRocketMQClientTemplate.receiveAsync(). - Added a regression test with a recording fake
SimpleConsumerto verify the consumer remains open and reusable across multiplereceiveAsync()calls.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/core/RocketMQClientTemplate.java | Stops prematurely closing the shared SimpleConsumer during receiveAsync(). |
| rocketmq-v5-client-spring-boot/src/test/java/org/apache/rocketmq/client/core/RocketMQClientTemplateReceiveAsyncTest.java | Adds a regression test to ensure receiveAsync() does not close the shared consumer and remains reusable. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| 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); |
| 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()); |
Which Issue(s) This PR Fixes
Fixes #780
Brief Description
receiveAsyncclosed the sharedSimpleConsumersingleton right after starting an async receive, which disrupted the in-flight future and permanently broke every subsequentreceive/receiveAsync/ack/changeInvisibleDurationon the same template ("consumer already closed"). The consumer's lifecycle belongs to the bean'sdestroy(), which already closes it — the synchronousreceive()sibling never closes it either.Fix: remove the
simpleConsumer.close()call fromreceiveAsync.How Did You Test This Change?
New regression test
RocketMQClientTemplateReceiveAsyncTestusing a recording fake consumer (the module has no mockito): on the unfixed code it fails withexpected:<0> but was:<1>(close was invoked); with the fix it passes — close count stays 0, the consumer remains usable, and a secondreceiveAsyncsucceeds. 1/1 pass on current master.