[ISSUE #8725] clean DefaultMQPushConsumer after start fail (#8726)

* [ISSUE #8725]clean DefaultMQPushConsumer after start fail * clean DefaultLitePullConsumerImpl after start fail
apache · Oct 30, 2024 · 5600684 · 5600684
1 parent fe80772
commit 5600684
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 5 deletions.
diff --git a/...t/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java b/...t/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
@@ -307,7 +307,12 @@ public synchronized void start() throws MQClientException {
 
                 log.info("the consumer [{}] start OK", this.defaultLitePullConsumer.getConsumerGroup());
 
-                operateAfterRunning();
+                try {
+                    operateAfterRunning();
+                } catch (Exception e) {
+                    shutdown();
+                    throw e;
+                }
 
                 break;
             case RUNNING:

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
@@ -1006,10 +1006,16 @@ public synchronized void start() throws MQClientException {
                 break;
         }
 
-        this.updateTopicSubscribeInfoWhenSubscriptionChanged();
-        this.mQClientFactory.checkClientInBroker();
-        if (this.mQClientFactory.sendHeartbeatToAllBrokerWithLock()) {
-            this.mQClientFactory.rebalanceImmediately();
+        try {
+            this.updateTopicSubscribeInfoWhenSubscriptionChanged();
+            this.mQClientFactory.checkClientInBroker();
+            if (this.mQClientFactory.sendHeartbeatToAllBrokerWithLock()) {
+                this.mQClientFactory.rebalanceImmediately();
+            }
+        } catch (Exception e) {
+            log.warn("Start the consumer {} fail.", this.defaultMQPushConsumer.getConsumerGroup(), e);
+            shutdown();
+            throw e;
         }
     }