(rocketmq) branch develop updated: [ISSUE #11427] fix(client): stop the pop consume service when the push consumer shuts down (#11428)

[jinrongtong]

This is an automated email from the ASF dual-hosted git repository.

RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git


The following commit(s) were added to refs/heads/develop by this push:
     new 57c2dc04f7 [ISSUE #11427] fix(client): stop the pop consume service 
when the push consumer shuts down (#11428)
57c2dc04f7 is described below

commit 57c2dc04f78cc7e73e6b8c9436ee0fff0d0f46c3
Author: bai mengxuan <2013373...@qq.com>
AuthorDate: Tue Oct 20 09:19:19 2026 +0800

    [ISSUE #11427] fix(client): stop the pop consume service when the push 
consumer shuts down (#11428)
---
 .../client/impl/consumer/DefaultMQPushConsumerImpl.java      |  2 ++
 .../client/impl/consumer/DefaultMQPushConsumerImplTest.java  | 12 ++++++++++++
 2 files changed, 14 insertions(+)

diff --git 
a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
 
b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
index 0ae779971c..9229fb42a8 100644
--- 
a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
+++ 
b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
@@ -906,6 +906,7 @@ public class DefaultMQPushConsumerImpl implements 
MQConsumerInner {
                 break;
             case RUNNING:
                 this.consumeMessageService.shutdown(awaitTerminateMillis);
+                this.consumeMessagePopService.shutdown(awaitTerminateMillis);
                 this.persistConsumerOffset();
                 
this.mQClientFactory.unregisterConsumer(this.defaultMQPushConsumer.getConsumerGroup());
                 this.mQClientFactory.shutdown();
@@ -989,6 +990,7 @@ public class DefaultMQPushConsumerImpl implements 
MQConsumerInner {
                 if (!registerOK) {
                     this.serviceState = ServiceState.CREATE_JUST;
                     
this.consumeMessageService.shutdown(defaultMQPushConsumer.getAwaitTerminationMillisWhenShutdown());
+                    
this.consumeMessagePopService.shutdown(defaultMQPushConsumer.getAwaitTerminationMillisWhenShutdown());
                     throw new MQClientException("The consumer group[" + 
this.defaultMQPushConsumer.getConsumerGroup()
                         + "] has been created before, specify another name 
please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                         null);
diff --git 
a/client/src/test/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImplTest.java
 
b/client/src/test/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImplTest.java
index 2bc9c5a18d..0bb759b65c 100644
--- 
a/client/src/test/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImplTest.java
+++ 
b/client/src/test/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImplTest.java
@@ -96,6 +96,7 @@ import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -701,6 +702,17 @@ public class DefaultMQPushConsumerImplTest {
         assertEquals(ServiceState.SHUTDOWN_ALREADY, 
defaultMQPushConsumerImpl.getServiceState());
     }
 
+    @Test
+    public void testShutdownStopsThePopConsumeService() throws Exception {
+        ConsumeMessageService popService = mock(ConsumeMessageService.class);
+        FieldUtils.writeDeclaredField(defaultMQPushConsumerImpl, 
"consumeMessagePopService", popService, true);
+
+        defaultMQPushConsumerImpl.setServiceState(ServiceState.RUNNING);
+        defaultMQPushConsumerImpl.shutdown();
+
+        verify(popService, times(1)).shutdown(anyLong());
+    }
+
     @Test
     public void testSubscribe() throws MQClientException {
         defaultMQPushConsumerImpl.subscribe(defaultTopic, "fullClassname", 
"filterClassSource");