private RegisterResponseB2C inProcessConsumerRegister(String clientId, String groupName,
            String topicName, String partStr,
            Set<String> filterCondSet, String msgRcvFrom,
            boolean overtls, RegisterRequestC2B request,
            RegisterResponseB2C.Builder builder,
            StringBuilder strBuffer) {
        String consumerId = null;
        ConsumerNodeInfo consumerNodeInfo = consumerRegisterMap.get(partStr);
        if (consumerNodeInfo != null) {
            consumerId = consumerNodeInfo.getConsumerId();
        }
        if (TStringUtils.isEmpty(consumerId) || consumerId.equals(clientId)) {
            final long reqOffset = request.hasCurrOffset() ? request.getCurrOffset() : -1;
            long reqSessionTime = request.hasSessionTime() ? request.getSessionTime() : -1;
            String reqSessionKey = request.hasSessionKey() ? request.getSessionKey() : null;
            int reqQryPriorityId = request.hasQryPriorityId()
                    ? request.getQryPriorityId()
                    : TBaseConstants.META_VALUE_UNDEFINED;
            consumerNodeInfo =
                    new ConsumerNodeInfo(storeManager, reqQryPriorityId, groupName,
                            clientId, filterCondSet, reqSessionKey, reqSessionTime,
                            true, partStr, msgRcvFrom);
            if (consumerRegisterMap.put(partStr, consumerNodeInfo) == null) {
                BrokerSrvStatsHolder.incConsumeOnlineCnt();
            }
            heartbeatManager.regConsumerNode(getHeartbeatNodeId(clientId, partStr), clientId, partStr);
            MessageStore dataStore = null;
            try {
                dataStore = this.storeManager.getOrCreateMessageStore(topicName, request.getPartitionId());
                if (dataStore == null) {
                    builder.setErrCode(TErrCodeConstants.FORBIDDEN);
                    builder.setErrMsg(strBuffer.append("Topic ").append(topicName).append("-")
                            .append(request.getPartitionId())
                            .append(" not existed, please check your configure").toString());
                    return builder.build();
                }
            } catch (Throwable e0) {
                strBuffer.delete(0, strBuffer.length());
                String message = "Register broker failure!";
                logger.warn(message, e0);
                builder.setErrCode(TErrCodeConstants.INTERNAL_SERVER_ERROR);
                builder.setErrMsg(strBuffer.append(message).append(", exception is ")
                        .append(e0.getMessage()).toString());
                return builder.build();
            }
            OffsetStorageInfo offsetInfo =
                    offsetManager.loadOffset(dataStore, groupName, topicName,
                            request.getPartitionId(), request.getReadStatus(), reqOffset, strBuffer);
            logger.info(strBuffer.append("[Consumer Register]").append(clientId)
                    .append(TokenConstants.SEGMENT_SEP).append(partStr)
                    .append(TokenConstants.SEGMENT_SEP).append(offsetInfo)
                    .append(", reqOffset=").append(reqOffset)
                    .append(", reqQryPriorityId=").append(reqQryPriorityId)
                    .append(", isOverTLS=").append(overtls).toString());
            builder.setSuccess(true);
            builder.setErrCode(TErrCodeConstants.SUCCESS);
            builder.setErrMsg("OK!");
            builder.setCurrOffset(offsetInfo.getOffset());
            if (getRealQryPriorityId(consumerNodeInfo) <= 1) {
                builder.setMaxOffset(dataStore.getFileIndexMaxOffset());
            } else {
                builder.setMaxOffset(dataStore.getIndexMaxOffset());
            }
            return builder.build();
        } else {
            TimeoutInfo timeoutInfo =
                    heartbeatManager.getConsumerRegMap().get(getHeartbeatNodeId(consumerId, partStr));
            if (timeoutInfo == null || System.currentTimeMillis() >= timeoutInfo.getTimeoutTime()) {
                if (consumerRegisterMap.remove(partStr) != null) {
                    BrokerSrvStatsHolder.decConsumeOnlineCnt(true);
                }
                strBuffer.append("[Duplicated Register] Remove Invalid Consumer Register ")
                        .append(consumerId).append(TokenConstants.SEGMENT_SEP).append(partStr);
            } else {
                strBuffer.append("[Duplicated Register] Partition ").append(tubeConfig.getBrokerId())
                        .append(TokenConstants.SEGMENT_SEP).append(partStr)
                        .append(" has been registered by ").append(consumerId);
            }
            logger.warn(strBuffer.toString());
            builder.setErrCode(TErrCodeConstants.PARTITION_OCCUPIED);
            builder.setErrMsg(strBuffer.toString());
            return builder.build();
        }
    }
Handle consumer's register request.

@param clientId        the client id
@param groupName       the group name
@param topicName       the topic name
@param partStr         the group-topic-partitionId key
@param filterCondSet   the filter condition set
@param msgRcvFrom      the address message received
@param overtls      whether transfer over TLS
@param request         the request
@param builder      the response builder
@param strBuffer    the string buffer
@return             the response
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/BrokerServiceServer.java