@Override
    public RegisterResponseB2C consumerRegisterC2B(RegisterRequestC2B request,
            final String rmtAddress,
            boolean overtls) throws Throwable {
        final ProcessResult result = new ProcessResult();
        RegisterResponseB2C.Builder builder = RegisterResponseB2C.newBuilder();
        builder.setSuccess(false);
        builder.setCurrOffset(-1);
        if (!this.started.get()) {
            builder.setErrCode(TErrCodeConstants.SERVICE_UNAVAILABLE);
            builder.setErrMsg("StoreService temporary unavailable!");
            return builder.build();
        }
        if (!serverAuthHandler.identityValidUserInfo(request.getAuthInfo(), false, result)) {
            builder.setErrCode(result.getErrCode());
            builder.setErrMsg(result.getErrMsg());
            return builder.build();
        }
        final CertifiedInfo certifiedInfo = (CertifiedInfo) result.getRetData();
        final StringBuilder strBuffer = new StringBuilder(512);
        // get and check clientId field
        if (!PBParameterUtils.checkClientId(request.getClientId(), strBuffer, result)) {
            builder.setErrCode(result.getErrCode());
            builder.setErrMsg(result.getErrMsg());
            return builder.build();
        }
        final String clientId = (String) result.getRetData();
        // get and check topicName field
        if (!PBParameterUtils.getTopicNameParameter(request.getTopicName(),
                this.metadataManager, strBuffer, result)) {
            builder.setErrCode(result.getErrCode());
            builder.setErrMsg(result.getErrMsg());
            return builder.build();
        }
        // get consumer info
        final String topicName = (String) result.getRetData();
        // get and check groupName field
        if (!PBParameterUtils.checkGroupName(request.getGroupName(), strBuffer, result)) {
            builder.setErrCode(result.getErrCode());
            builder.setErrMsg(result.getErrMsg());
            return builder.build();
        }
        final String groupName = (String) result.getRetData();
        boolean isRegister = (request.getOpType() == RpcConstants.MSG_OPTYPE_REGISTER);
        Set<String> filterCondSet = new HashSet<>();
        if (request.getFilterCondStrList() != null && !request.getFilterCondStrList().isEmpty()) {
            for (String filterCond : request.getFilterCondStrList()) {
                if (TStringUtils.isNotBlank(filterCond)) {
                    filterCondSet.add(filterCond.trim());
                }
            }
        }
        if (!serverAuthHandler.validConsumeAuthorizeInfo(certifiedInfo.getUserName(),
                groupName, topicName, filterCondSet, isRegister, rmtAddress, result)) {
            builder.setErrCode(result.getErrCode());
            builder.setErrMsg(result.getErrMsg());
            return builder.build();
        }
        Integer lid = null;
        Integer partLock = null;
        String partStr = getPartStr(groupName, topicName, request.getPartitionId());
        try {
            lid = brokerRowLock.getLock(null, StringUtils.getBytesUtf8(clientId), true);
            try {
                partLock = brokerRowLock.getLock(null, StringUtils.getBytesUtf8(partStr), true);
                if (request.getOpType() == RpcConstants.MSG_OPTYPE_REGISTER) {
                    return inProcessConsumerRegister(clientId, groupName,
                            topicName, partStr, filterCondSet, rmtAddress,
                            overtls, request, builder, strBuffer);
                } else if (request.getOpType() == RpcConstants.MSG_OPTYPE_UNREGISTER) {
                    return inProcessConsumerUnregister(clientId, groupName,
                            topicName, partStr, request, overtls, builder, strBuffer);
                } else {
                    String message = strBuffer.append("Invalid request:").append(request.getOpType()).toString();
                    logger.info(message);
                    builder.setErrCode(TErrCodeConstants.BAD_REQUEST);
                    builder.setErrMsg(message);
                    return builder.build();
                }
            } finally {
                if (partLock != null) {
                    brokerRowLock.releaseRowLock(partLock);
                }
            }
        } catch (IOException e) {
            strBuffer.delete(0, strBuffer.length());
            String message = "Failed to lock.";
            logger.warn(message, e);
            builder.setErrCode(TErrCodeConstants.BAD_REQUEST);
            builder.setErrMsg(strBuffer.append(message).append(e.getMessage()).toString());
            return builder.build();
        } finally {
            if (lid != null) {
                brokerRowLock.releaseRowLock(lid);
            }
        }
    }
Handle consumer register request.

@param request       the request
@param rmtAddress    the remote address
@param overtls       whether transfer over TLS
@return              the response
@throws Throwable    the exception during processing
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/BrokerServiceServer.java