private void registerPartitions(Map<BrokerInfo, List<Partition>> registerInfoMap,
            List<Partition> unRegPartitions) throws InterruptedException {
        int retryTimesRegister2Broker = 0;
        StringBuilder strBuffer = new StringBuilder(512);
        while ((!unRegPartitions.isEmpty())
                && (retryTimesRegister2Broker < this.consumerConfig.getMaxRegisterRetryTimes())) {
            for (Map.Entry<BrokerInfo, List<Partition>> entry : registerInfoMap.entrySet()) {
                ConcurrentLinkedQueue<Partition> regedPartitions =
                        rmtDataCache.getPartitionByBroker(entry.getKey());
                for (Partition partition : entry.getValue()) {
                    if ((isRebalanceStopped()) || (isShutdown())) {
                        return;
                    }
                    try {
                        if (regedPartitions != null && regedPartitions.contains(partition)) {
                            unRegPartitions.remove(partition);
                            continue;
                        }
                        ClientBroker.RegisterResponseB2C responseB2C =
                                getBrokerService(partition.getBroker())
                                        .consumerRegisterC2B(createBrokerRegisterRequest(partition),
                                                AddressUtils.getLocalAddress(), consumerConfig.isTlsEnable());
                        if (responseB2C != null && responseB2C.getSuccess()) {
                            clientStatsInfo.bookReg2Broker(false);
                            long currOffset =
                                    responseB2C.hasCurrOffset() ? responseB2C.getCurrOffset()
                                            : TBaseConstants.META_VALUE_UNDEFINED;
                            long maxOffset =
                                    responseB2C.hasMaxOffset() ? responseB2C.getMaxOffset()
                                            : TBaseConstants.META_VALUE_UNDEFINED;
                            rmtDataCache.addPartition(partition, currOffset, maxOffset);
                            unRegPartitions.remove(partition);
                            logger.info(strBuffer.append("Registered partition: consumer is ")
                                    .append(consumerId).append(", partition is:")
                                    .append(partition.toString()).toString());
                            strBuffer.delete(0, strBuffer.length());
                        } else {
                            clientStatsInfo.bookReg2Broker(true);
                            if (responseB2C == null) {
                                logger.warn(strBuffer.append("register2broker error! ")
                                        .append(retryTimesRegister2Broker).append(" register ")
                                        .append(partition.toString()).append(" return null!")
                                        .toString());
                            } else {
                                if (responseB2C.getErrCode() == TErrCodeConstants.PARTITION_OCCUPIED
                                        || responseB2C.getErrCode() == TErrCodeConstants.CERTIFICATE_FAILURE) {
                                    unRegPartitions.remove(partition);
                                    if (responseB2C.getErrCode() == TErrCodeConstants.PARTITION_OCCUPIED) {
                                        if (logger.isDebugEnabled()) {
                                            logger.debug(strBuffer
                                                    .append("[Partition occupied], curr consumerId: ")
                                                    .append(consumerId).append(", returned message : ")
                                                    .append(responseB2C.getErrMsg()).toString());
                                        }
                                    } else {
                                        logger.warn(strBuffer
                                                .append("[Certificate failure], curr consumerId: ")
                                                .append(consumerId).append(", returned message : ")
                                                .append(responseB2C.getErrMsg()).toString());
                                    }
                                } else {
                                    logger.warn(strBuffer.append("register2broker error! ")
                                            .append(retryTimesRegister2Broker).append(" register ")
                                            .append(partition.toString()).append(" return ")
                                            .append(responseB2C.getErrMsg()).toString());
                                }
                            }
                            strBuffer.delete(0, strBuffer.length());
                        }
                    } catch (IOException e) {
                        strBuffer.delete(0, strBuffer.length());
                        logger.warn(strBuffer.append("register2broker error1 ! ")
                                .append(retryTimesRegister2Broker).append(" ")
                                .append(partition.toString()).toString(), e);
                        strBuffer.delete(0, strBuffer.length());
                    } catch (Throwable ee) {
                        strBuffer.delete(0, strBuffer.length());
                        logger.warn(strBuffer.append("register2broker error2 ! ")
                                .append(retryTimesRegister2Broker).append(" ")
                                .append(partition.toString()).toString(), ee);
                        strBuffer.delete(0, strBuffer.length());
                    }
                }
            }
            retryTimesRegister2Broker++;
            Thread.sleep(1000);
        }
        for (Partition partition : unRegPartitions) {
            boolean result = removePartition(partition);
            logger.info(strBuffer.append("[Remove Partition] ")
                    .append(partition.toString()).append(" ")
                    .append(result).toString());
            strBuffer.delete(0, strBuffer.length());
        }
    }
Register partitions.

@param registerInfoMap broker partition mapping
@param unRegPartitions unregister partition list
@throws InterruptedException
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/BaseMessageConsumer.java