opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-19 02:23:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
* [ISSUE #8725]clean DefaultMQPushConsumer after start fail

* clean DefaultLitePullConsumerImpl after start fail
This commit is contained in:
yuz10
2024-10-30 20:08:28 +08:00
committed by GitHub
parent fe8077250b
commit 5600684eb4
2 changed files with 16 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
+6 -1
View File
@@ -307,7 +307,12 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
log.info("the consumer [{}] start OK", this.defaultLitePullConsumer.getConsumerGroup());
operateAfterRunning();
try {
operateAfterRunning();
} catch (Exception e) {
shutdown();
throw e;
}
break;
case RUNNING:
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
+10 -4
View File
@@ -1006,10 +1006,16 @@ public class DefaultMQPushConsumerImpl implements MQConsumerInner {
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;
}
}