private void resetCsmFromMaxOffset(StringBuilder sBuffer) {
            Set<String> regPartSet = clientRmtDataCache.getCurRegisteredPartSet();
            if (regPartSet.isEmpty()) {
                return;
            }
            Partition partition;
            long boostrapOffset;
            for (String partitionKey : regPartSet) {
                if (TStringUtils.isBlank(partitionKey)) {
                    continue;
                }
                if (isShutdown()) {
                    break;
                }
                boostrapOffset =
                        clientRmtDataCache.getMaxOffsetOfPartition(partitionKey);
                final String uniqueId =
                        sBuffer.append(consumerConfig.getConsumerGroup())
                                .append("#").append(partitionKey).toString();
                sBuffer.delete(0, sBuffer.length());
                synchronized (uniqueId) {
                    partition = clientRmtDataCache.getPartitionByKey(partitionKey);
                    if (partition == null) {
                        continue;
                    }
                    try {
                        ClientBroker.RegisterResponseB2C response =
                                getBrokerService(partition.getBroker()).consumerRegisterC2B(
                                        createBrokerRegisterRequest(partition, boostrapOffset),
                                        AddressUtils.getLocalAddress(), consumerConfig.isTlsEnable());
                        if (response == null) {
                            continue;
                        }
                        if (response.getSuccess()) {
                            long currOffset = response.hasCurrOffset()
                                    ? response.getCurrOffset()
                                    : TBaseConstants.META_VALUE_UNDEFINED;
                            long maxOffset = response.hasMaxOffset()
                                    ? response.getMaxOffset()
                                    : TBaseConstants.META_VALUE_UNDEFINED;
                            clientRmtDataCache.updPartOffsetInfo(
                                    partitionKey, currOffset, maxOffset);
                            logger.info(sBuffer.append("[Admin Reset] consumer is ")
                                    .append(consumerId).append(", partition=")
                                    .append(partition.toString()).append(", consume from max=")
                                    .append(currOffset).toString());
                            sBuffer.delete(0, sBuffer.length());
                        } else {
                            if (response.getErrCode() == TErrCodeConstants.PARTITION_OCCUPIED
                                    || response.getErrCode() == TErrCodeConstants.CERTIFICATE_FAILURE) {
                                clientRmtDataCache.removePartition(partition);
                            }
                        }
                    } catch (Throwable e) {
                        sBuffer.delete(0, sBuffer.length());
                        logger.info(sBuffer.append("register ").append(partition.toString())
                                .append(" throw exception ").append(e.toString()).toString());
                        sBuffer.delete(0, sBuffer.length());
                    }
                }
            }
        }
Start heartbeat thread threads.

@param result  process result
@param sBuffer  string buffer

@return  process result
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/SimpleClientBalanceConsumer.java