@Override
    public HeartBeatResponseB2C consumerHeartbeatC2B(HeartBeatRequestC2B request,
            final String rmtAddress,
            boolean overtls) throws Throwable {
        ProcessResult result = new ProcessResult();
        final StringBuilder strBuffer = new StringBuilder(512);
        final HeartBeatResponseB2C.Builder builder = HeartBeatResponseB2C.newBuilder();
        builder.setSuccess(false);
        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();
        }
        CertifiedInfo certifiedInfo = (CertifiedInfo) result.getRetData();
        // 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 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();
        int reqQryPriorityId = request.hasQryPriorityId()
                ? request.getQryPriorityId()
                : TBaseConstants.META_VALUE_UNDEFINED;
        List<Partition> partitions =
                DataConverterUtil.convertPartitionInfo(request.getPartitionInfoList());
        boolean isAuthorized = false;
        List<String> failureInfo = new ArrayList<>();
        for (Partition partition : partitions) {
            String topic = partition.getTopic();
            int partitionId = partition.getPartitionId();
            String partStr = getPartStr(groupName, topic, partitionId);
            ConsumerNodeInfo consumerNodeInfo = consumerRegisterMap.get(partStr);
            if (consumerNodeInfo == null) {
                failureInfo.add(strBuffer.append(TErrCodeConstants.HB_NO_NODE)
                        .append(TokenConstants.ATTR_SEP)
                        .append(partition).toString());
                strBuffer.delete(0, strBuffer.length());
                logger.warn(strBuffer.append("[Heartbeat Check] UnRegistered Consumer:")
                        .append(clientId).append(TokenConstants.SEGMENT_SEP)
                        .append(partStr).toString());
                strBuffer.delete(0, strBuffer.length());
                continue;
            }
            if (!clientId.equals(consumerNodeInfo.getConsumerId())) {
                failureInfo.add(strBuffer.append(TErrCodeConstants.DUPLICATE_PARTITION)
                        .append(TokenConstants.ATTR_SEP).append(partition).toString());
                strBuffer.delete(0, strBuffer.length());
                strBuffer.append("[Heartbeat Check] Duplicated partition: Partition ").append(partStr)
                        .append(" has been consumed by ").append(consumerNodeInfo.getConsumerId())
                        .append(";Current consumer ").append(clientId);
                logger.warn(strBuffer.toString());
                strBuffer.delete(0, strBuffer.length());
                continue;
            }
            if (!isAuthorized) {
                if (!serverAuthHandler.validConsumeAuthorizeInfo(certifiedInfo.getUserName(),
                        groupName, topic, consumerNodeInfo.getFilterCondStrs(), true, rmtAddress, result)) {
                    builder.setErrCode(result.getErrCode());
                    builder.setErrMsg(result.getErrMsg());
                    return builder.build();
                }
                isAuthorized = true;
            }
            try {
                heartbeatManager.updConsumerNode(
                        getHeartbeatNodeId(clientId, partStr));
            } catch (HeartbeatException e) {
                failureInfo.add(strBuffer.append(TErrCodeConstants.HB_NO_NODE)
                        .append(TokenConstants.ATTR_SEP).append(partition).toString());
                strBuffer.delete(0, strBuffer.length());
                logger.warn(strBuffer.append("[Heartbeat Check] Invalid Request")
                        .append(clientId).append(TokenConstants.SEGMENT_SEP)
                        .append(topic).append(TokenConstants.ATTR_SEP).append(partitionId).toString());
                strBuffer.delete(0, strBuffer.length());
                continue;
            }
            if (consumerNodeInfo.getQryPriorityId() != reqQryPriorityId) {
                consumerNodeInfo.setQryPriorityId(reqQryPriorityId);
            }
        }
        builder.setRequireAuth(certifiedInfo.isReAuth());
        builder.setSuccess(true);
        builder.setErrCode(TErrCodeConstants.SUCCESS);
        builder.setHasPartFailure(false);
        if (!failureInfo.isEmpty()) {
            builder.setHasPartFailure(true);
            builder.addAllFailureInfo(failureInfo);
        }
        builder.setErrMsg("OK!");
        return builder.build();
    }
Handle consumer's heartbeat 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