diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java index 7acc25cc8f..1057e83dff 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java @@ -241,7 +241,7 @@ public abstract class RebalanceImpl { case BROADCASTING: { Set mqSet = this.topicSubscribeInfoTable.get(topic); if (mqSet != null) { - boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder); + boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, false); if (changed) { this.messageQueueChanged(topic, mqSet, mqSet); log.info("messageQueueChanged {} {} {} {}", @@ -327,7 +327,7 @@ public abstract class RebalanceImpl { } private boolean updateProcessQueueTableInRebalance(final String topic, final Set mqSet, - final boolean isOrder) { + final boolean needLockMq) { boolean changed = false; Iterator> it = this.processQueueTable.entrySet().iterator(); @@ -367,7 +367,7 @@ public abstract class RebalanceImpl { List pullRequestList = new ArrayList(); for (MessageQueue mq : mqSet) { if (!this.processQueueTable.containsKey(mq)) { - if (isOrder && !this.lock(mq)) { + if (needLockMq && !this.lock(mq)) { log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq); continue; }