private void startMasterAndBrokerThreads() throws TubeClientException {
        int registerRetryTimes = 0;
        StringBuilder strBuffer = new StringBuilder(256);
        while (registerRetryTimes < consumerConfig.getMaxRegisterRetryTimes()) {
            try {
                ClientMaster.RegisterResponseM2C response =
                        masterService.consumerRegisterC2M(createMasterRegisterRequest(),
                                AddressUtils.getLocalAddress(), consumerConfig.isTlsEnable());
                if (response != null && response.getSuccess()) {
                    processRegisterAllocAndRspFlowRules(response, strBuffer);
                    processRegAuthorizedToken(response);
                    break;
                }
                if (response == null) {
                    logger.warn(strBuffer.append("[Register Failed] ")
                            .append("response return null!").toString());
                } else {
                    if (response.getErrCode() == TErrCodeConstants.CONSUME_GROUP_FORBIDDEN) {
                        throw new TubeClientException(strBuffer
                                .append("Register to master failed! ConsumeGroup forbidden, ")
                                .append(response.getErrMsg()).toString());
                    } else if (response.getErrCode() == TErrCodeConstants.CONSUME_CONTENT_FORBIDDEN) {
                        throw new TubeClientException(strBuffer
                                .append("Register to master failed! Restricted consume content, ")
                                .append(response.getErrMsg()).toString());
                    } else {
                        logger.warn(strBuffer.append("[Register Failed] ")
                                .append(response.getErrMsg()).toString());
                    }
                }
                strBuffer.delete(0, strBuffer.length());
            } catch (Throwable e) {
                logger.warn("Register to master failed.", e);
                ThreadUtils.sleep(this.consumerConfig.getRegFailWaitPeriodMs());
            }
            if (++registerRetryTimes >= consumerConfig.getMaxRegisterRetryTimes()) {
                subStatus.compareAndSet(0, -1);
                logger.error(
                        "Register to master failed! please check and retry later.");
                throw new TubeClientException(
                        "Register to master failed! please check and retry later.");
            }
        }
        // to master heartbeat
        this.lastHeartbeatTime2Master = System.currentTimeMillis();
        this.heartService2Master.scheduleWithFixedDelay(new HeartTask2MasterWorker(),
                0, consumerConfig.getHeartbeatPeriodMs(), TimeUnit.MILLISECONDS);
        // to broker
        this.lastHeartbeatTime2Broker = System.currentTimeMillis();
        this.heartBeatThread2Broker = new Thread(new HeartTask2BrokerWorker());
        heartBeatThread2Broker
                .setName(strBuffer.append("Broker-Heartbeat-Thread-").append(consumerId).toString());
        heartBeatThread2Broker.setPriority(Thread.MAX_PRIORITY);
        heartBeatThread2Broker.start();
        rebalanceThread.start();
    }
Start heartbeat thread and rebalance thread.

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