@Override
    public RegisterResponseM2B brokerRegisterB2M(RegisterRequestB2M request,
            final String rmtAddress,
            boolean overtls) throws Exception {
        // #lizard forgives
        final ProcessResult result = new ProcessResult();
        final StringBuilder strBuff = new StringBuilder(512);
        RegisterResponseM2B.Builder builder = RegisterResponseM2B.newBuilder();
        builder.setSuccess(false);
        builder.setStopRead(false);
        builder.setStopWrite(false);
        builder.setTakeConfInfo(false);
        // auth
        if (!serverAuthHandler.identityValidBrokerInfo(request.getAuthInfo(), result)) {
            builder.setErrCode(result.getErrCode());
            builder.setErrMsg(result.getErrMsg());
            return builder.build();
        }
        // get clientId and check valid
        if (!PBParameterUtils.checkClientId(request.getClientId(), strBuff, result)) {
            builder.setErrCode(result.getErrCode());
            builder.setErrMsg(result.getErrMsg());
            return builder.build();
        }
        final String clientId = (String) result.getRetData();
        // check authority
        checkNodeStatus(clientId, strBuff);
        // get optional filed
        ClusterSettingEntity defSetting =
                defMetaDataService.getClusterDefSetting(false);
        final long reFlowCtrlId = request.hasFlowCheckId()
                ? request.getFlowCheckId()
                : TBaseConstants.META_VALUE_UNDEFINED;
        final int qryPriorityId = request.hasQryPriorityId()
                ? request.getQryPriorityId()
                : TBaseConstants.META_VALUE_UNDEFINED;
        int tlsPort = request.hasTlsPort()
                ? request.getTlsPort()
                : defSetting.getBrokerTLSPort();
        // build broker info
        BrokerInfo brokerInfo =
                new BrokerInfo(clientId, request.getEnableTls(), tlsPort);
        // register broker run status info
        if (!brokerRunManager.brokerRegister2M(clientId, brokerInfo,
                request.getCurBrokerConfId(), request.getConfCheckSumId(),
                true, request.getBrokerDefaultConfInfo(),
                request.getBrokerTopicSetConfInfoList(), request.getBrokerOnline(),
                overtls, strBuff, result)) {
            builder.setErrCode(result.getErrCode());
            builder.setErrMsg(result.getErrMsg());
            return builder.build();
        }
        // print broker register log
        logger.info(strBuff.append("[Broker Register] ").append(clientId)
                .append(" report, configureId=").append(request.getCurBrokerConfId())
                .append(",readStatusRpt=").append(request.getReadStatusRpt())
                .append(",writeStatusRpt=").append(request.getWriteStatusRpt())
                .append(",isTlsEnable=").append(brokerInfo.isEnableTLS())
                .append(",TLSport=").append(brokerInfo.getTlsPort())
                .append(",FlowCtrlId=").append(reFlowCtrlId)
                .append(",qryPriorityId=").append(qryPriorityId)
                .append(",checksumId=").append(request.getConfCheckSumId()).toString());
        strBuff.delete(0, strBuff.length());
        // response
        builder.setSuccess(true);
        builder.setErrCode(TErrCodeConstants.SUCCESS);
        builder.setErrMsg("OK!");
        // begin: deprecated when brokers version equal to current master version
        builder.setAuthorizedInfo(genAuthorizedInfo(null, true));
        // end deprecated
        builder.setBrokerAuthorizedInfo(genBrokerAuthorizedInfo(null));
        EnableBrokerFunInfo.Builder enableInfo = EnableBrokerFunInfo.newBuilder();
        enableInfo.setEnableVisitTokenCheck(masterConfig.isStartVisitTokenCheck());
        enableInfo.setEnableProduceAuthenticate(masterConfig.isStartProduceAuthenticate());
        enableInfo.setEnableProduceAuthorize(masterConfig.isStartProduceAuthorize());
        enableInfo.setEnableConsumeAuthenticate(masterConfig.isStartConsumeAuthenticate());
        enableInfo.setEnableConsumeAuthorize(masterConfig.isStartConsumeAuthorize());
        builder.setEnableBrokerInfo(enableInfo);
        brokerRunManager.setRegisterDownConfInfo(brokerInfo.getBrokerId(), strBuff, builder);
        builder.setSsdStoreId(TBaseConstants.META_VALUE_UNDEFINED);
        ClientMaster.ClusterConfig.Builder clusterConfigBuilder =
                buildClusterConfig(request.getClsConfig());
        if (clusterConfigBuilder != null) {
            builder.setClsConfig(clusterConfigBuilder);
        }
        if (request.hasFlowCheckId()) {
            builder.setQryPriorityId(defSetting.getQryPriorityId());
            builder.setFlowCheckId(defSetting.getSerialId());
            if (reFlowCtrlId != defSetting.getSerialId()) {
                if (defSetting.enableFlowCtrl()) {
                    builder.setFlowControlInfo(defSetting.getGloFlowCtrlRuleInfo());
                } else {
                    builder.setFlowControlInfo(" ");
                }
            }
        }
        logger.info(strBuff.append("[Broker Register] ").append(clientId)
                .append(", isOverTLS=").append(overtls).toString());
        return builder.build();
    }
Broker register request with master

@param request
@param rmtAddress
@param overtls
@return register response
@throws Exception
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/TMaster.java