private void connect2Broker(ConsumerEvent event) throws InterruptedException {
        Map<BrokerInfo, List<Partition>> registerInfoMap =
                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> curPartList =
                    registerInfoMap.computeIfAbsent(
                            broker, k -> new ArrayList<>());
            if (!curPartList.contains(partition)) {
                curPartList.add(partition);
            }
        }
        if ((isRebalanceStopped()) || (isShutdown())) {
            return;
        }
        List<Partition> unfinishedPartitions = new ArrayList<>();
        rmtDataCache.filterCachedPartitionInfo(registerInfoMap, unfinishedPartitions);
        registerPartitions(registerInfoMap, unfinishedPartitions);
        if (this.isFirst.get()) {
            this.isFirst.set(false);
        }
        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