RocketMQ 重平衡后，会导致什么异常吗？

public void executePullRequestImmediately(final PullRequest pullRequest) {  try {    this.pullRequestQueue.put(pullRequest);  } catch (InterruptedException e) {    log.error("executePullRequestImmediately pullRequestQueue.put", e);  }}

// 如果本次拉取消息为空，则继续将pullRequest放入阻塞队列中if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {  DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);} else {  // 将消息放入消费者消费线程去执行  DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(//    pullResult.getMsgFoundList(), //    processQueue, //    pullRequest.getMessageQueue(), //    dispathToConsume);  // 将pullRequest放入阻塞队列中  DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);  }

pullRequest.setNextOffset(pullResult.getNextBeginOffset());DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);

final ProcessQueue processQueue = pullRequest.getProcessQueue();if (processQueue.isDropped()) {  log.info("the pull request[{}] is dropped.", pullRequest.toString());  return;}

Iterator<Entry<MessageQueue, ProcessQueue>> it = this.processQueueTable.entrySet().iterator();while (it.hasNext()) {  Entry<MessageQueue, ProcessQueue> next = it.next();  MessageQueue mq = next.getKey();  ProcessQueue pq = next.getValue();
  if (mq.getTopic().equals(topic)) {    // 判断当前缓存 MessageQueue 是否包含在最新的 mqSet 中，如果不存在则将队列丢弃    if (!mqSet.contains(mq)) {      pq.setDropped(true);      if (this.removeUnnecessaryMessageQueue(mq, pq)) {        it.remove();        changed = true;        log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);      }    } else if (pq.isPullExpired()) {      // 如果队列拉取过期则丢弃      switch (this.consumeType()) {        case CONSUME_ACTIVELY:          break;        case CONSUME_PASSIVELY:          pq.setDropped(true);          if (this.removeUnnecessaryMessageQueue(mq, pq)) {            it.remove();            changed = true;            log.error("[BUG]doRebalance, {}, remove unnecessary mq, {}, because pull is pause, so try to fixed it",                      consumerGroup, mq);          }          break;        default:          break;      }    }  }}

// 新建 ProcessQueue ProcessQueue pq = new ProcessQueue();long nextOffset = this.computePullFromWhere(mq);if (nextOffset >= 0) {  // 将ProcessQueue放入processQueueTable中  ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);  if (pre != null) {    log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);  } else {    log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);    PullRequest pullRequest = new PullRequest();    pullRequest.setConsumerGroup(consumerGroup);    pullRequest.setNextOffset(nextOffset);    pullRequest.setMessageQueue(mq);    // 将ProcessQueue放入pullRequest拉取任务对象中    pullRequest.setProcessQueue(pq);    pullRequestList.add(pullRequest);    changed = true;  }}

DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(//  pullResult.getMsgFoundList(), //  processQueue, //  pullRequest.getMessageQueue(), //  dispathToConsume);

// 并发消息消费逻辑实现类org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService;// 顺序消息消费逻辑实现类org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService;

if (this.processQueue.isDropped()) {  log.info("the message queue not be able to consume, because it's dropped. group={} {}", ConsumeMessageConcurrentlyService.this.consumerGroup, this.messageQueue);  return;}
// 消息消费逻辑// ...
// 如果队列被设置为丢弃状态，则不提交消息消费进度if (!processQueue.isDropped()) {    ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);} else {    log.warn("processQueue is dropped without process consume result. messageQueue={}, msgs={}", messageQueue, msgs);}

public void run() {  // 判断队列是否被丢弃  if (this.processQueue.isDropped()) {    log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);    return;  }
  final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);  synchronized (objLock) {    // 如果不是广播模式，且队列已加锁且锁没有过期    if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())        || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {      final long beginTime = System.currentTimeMillis();      for (boolean continueConsume = true; continueConsume; ) {        // 再次判断队列是否被丢弃        if (this.processQueue.isDropped()) {          log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);          break;        }
        // 消息消费处理逻辑        // ...
          continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);        } else {          continueConsume = false;        }      }    } else {      if (this.processQueue.isDropped()) {        log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);        return;      }      ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);    }  }}