private void procConfigFromHeartBeat(StringBuilder strBuff,
            HeartResponseM2B response) {
        // process service status
        ServiceStatusHolder
                .setReadWriteServiceStatus(response.getStopRead(),
                        response.getStopWrite(), "Master");
        // update default flow controller rules
        FlowCtrlRuleHandler defFlowCtrlHandler =
                metadataManager.getFlowCtrlRuleHandler();
        if (response.hasFlowCheckId()
                && response.getFlowCheckId() >= 0
                && response.getFlowCheckId() != defFlowCtrlHandler.getFlowCtrlId()) {
            try {
                defFlowCtrlHandler.updateFlowCtrlInfo(response.getQryPriorityId(),
                        response.getFlowCheckId(), response.getFlowControlInfo(), strBuff);
            } catch (Exception e1) {
                logger.warn("[HeartBeat response] found update flowCtrl rules failure", e1);
            }
        }
        // update configure report requirement
        requireReportConf = response.getNeedReportData();
        // update cluster setting
        if (response.hasClsConfig()) {
            long configId = response.getClsConfig().getConfigId();
            if (configId != ClusterConfigHolder.getConfigId()) {
                ClusterConfigHolder.updClusterSetting(response.getClsConfig());
                logger.info(strBuff
                        .append("[HeartBeat response] received cluster configure changed")
                        .append(",hasClsConfig=").append(response.hasClsConfig())
                        .append(",curClusterConfigId=").append(ClusterConfigHolder.getConfigId())
                        .append(",curMaxMsgSize=").append(ClusterConfigHolder.getMaxMsgSize())
                        .append(",minMemCacheSize=")
                        .append(ClusterConfigHolder.getMinMemCacheSize())
                        .toString());
                strBuff.delete(0, strBuff.length());
            }
        }
        if (response.getTakeConfInfo()) {
            logger.info(strBuff
                    .append("[HeartBeat response] received broker metadata info: brokerConfId=")
                    .append(response.getCurBrokerConfId())
                    .append(",stopWrite=").append(response.getStopWrite())
                    .append(",stopRead=").append(response.getStopRead())
                    .append(",configCheckSumId=").append(response.getConfCheckSumId())
                    .append(",hasFlowCtrl=").append(response.hasFlowCheckId())
                    .append(",curFlowCtrlId=").append(defFlowCtrlHandler.getFlowCtrlId())
                    .append(",curQryPriorityId=").append(defFlowCtrlHandler.getQryPriorityId())
                    .append(",curMaxMsgSize=").append(ClusterConfigHolder.getMaxMsgSize())
                    .append(",brokerDefaultConfInfo=")
                    .append(response.getBrokerDefaultConfInfo())
                    .append(",brokerTopicSetConfList=")
                    .append(response.getBrokerTopicSetConfInfoList()).toString());
            strBuff.delete(0, strBuff.length());
            metadataManager
                    .updateBrokerTopicConfigMap(response.getCurBrokerConfId(),
                            response.getConfCheckSumId(), response.getBrokerDefaultConfInfo(),
                            response.getBrokerTopicSetConfInfoList(), false, strBuff);
        }
        // update auth info
        if (response.hasBrokerAuthorizedInfo()) {
            serverAuthHandler.appendVisitToken(response.getBrokerAuthorizedInfo());
        }
        // process topic deletion
        boolean needProcess =
                metadataManager.updateBrokerRemoveTopicMap(
                        response.getTakeRemoveTopicInfo(),
                        response.getRemoveTopicConfInfoList(), strBuff);
        if (needProcess) {
            new Thread() {

                @Override
                public void run() {
                    Set<String> removedTopics = storeManager.removeTopicStore();
                    offsetManager.cleanRmvTopicInfo(removedTopics);
                }
            }.start();
        }
    }
Start broker service.

@throws Exception  the exception during processing
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/TubeBroker.java