private void disconnectFromBroker(ConsumerEvent event) throws InterruptedException {
        List<String> partKeys = new ArrayList<>();
        HashMap<BrokerInfo, List<Partition>> unRegisterInfoMap =
                new HashMap<>();
        List<SubscribeInfo> subscribeInfoList = event.getSubscribeInfoList();
        for (SubscribeInfo info : subscribeInfoList) {
            BrokerInfo broker =
                    new BrokerInfo(info.getBrokerId(), info.getHost(), info.getPort());
            Partition partition =
                    new Partition(broker, info.getTopic(), info.getPartitionId());
            List<Partition> unRegisterPartitionList =
                    unRegisterInfoMap.get(broker);
            if (unRegisterPartitionList == null) {
                unRegisterPartitionList = new ArrayList<>();
                unRegisterInfoMap.put(broker, unRegisterPartitionList);
            }
            if (!unRegisterPartitionList.contains(partition)) {
                unRegisterPartitionList.add(partition);
                partKeys.add(partition.getPartitionKey());
            }
        }
        if (isShutdown() || isRebalanceStopped()) {
            return;
        }
        Map<BrokerInfo, List<PartitionSelectResult>> unNewRegisterInfoMap =
                new HashMap<>();
        try {
            if (this.isPullConsume) {
                unNewRegisterInfoMap =
                        rmtDataCache.removeAndGetPartition(unRegisterInfoMap, partKeys,
                                this.consumerConfig.getPullRebConfirmWaitPeriodMs(),
                                this.consumerConfig.isPullRebConfirmTimeoutRollBack());

            } else {
                unNewRegisterInfoMap =
                        rmtDataCache.removeAndGetPartition(unRegisterInfoMap, partKeys,
                                this.consumerConfig.getPushListenerWaitPeriodMs(),
                                this.consumerConfig.isPushListenerWaitTimeoutRollBack());
            }
        } finally {
            unregisterPartitions(unNewRegisterInfoMap);
            event.setStatus(EventStatus.DONE);
        }
    }
Start heartbeat thread and rebalance thread.

@throws TubeClientException
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/BaseMessageConsumer.java