private Map<String, Map<String, Map<String, Partition>>> inReBalanceCluster(
            boolean isResetRebalance,
            ConsumerInfoHolder consumerHolder,
            BrokerRunManager brokerRunManager,
            List<String> groupSet,
            MetaDataService defMetaDataService,
            StringBuilder strBuffer) {
        // band consume reset offset
        Map<String, Map<String, Map<String, Partition>>> finalSubInfoMap =
                new HashMap<>();
        // according group
        for (String group : groupSet) {
            if (group == null) {
                continue;
            }
            // filter band consumer
            ConsumeGroupInfo consumeGroupInfo =
                    consumerHolder.getConsumeGroupInfo(group);
            if (consumeGroupInfo == null
                    || consumeGroupInfo.isGroupEmpty()
                    || consumeGroupInfo.isNotNeedBoundBalance()) {
                continue;
            }
            if (!consumeGroupInfo.isGroupFullSize()) {
                // check if client size meet minimal requirements
                Long checkCycle = consumerHolder.addCurCheckCycle(group);
                if (isResetRebalance) {
                    if (checkCycle != null && checkCycle % 15 == 0) {
                        logger.info(strBuffer.append("[Bound Alloc 2] Not allocate partition :group(")
                                .append(group).append(")'s consumer getCachedSize(")
                                .append(consumeGroupInfo.getGroupCnt())
                                .append(") low than required source count:")
                                .append(consumeGroupInfo.getSourceCount())
                                .append(", checked cycle is ")
                                .append(checkCycle).toString());
                        strBuffer.delete(0, strBuffer.length());
                    }
                } else {
                    logger.info(strBuffer.append("[Bound Alloc 1] Not allocate partition :group(")
                            .append(group).append(")'s consumer getCachedSize(")
                            .append(consumeGroupInfo.getGroupCnt())
                            .append(") low than required source count:")
                            .append(consumeGroupInfo.getSourceCount()).toString());
                    strBuffer.delete(0, strBuffer.length());
                }
                continue;
            }
            Map<String, Partition> partPubMap =
                    brokerRunManager.getSubBrokerAcceptSubParts(consumeGroupInfo.getTopicSet());
            Map<String, Partition> partitionMap = new HashMap<>();
            for (Partition partition : partPubMap.values()) {
                partitionMap.put(partition.getPartitionKey(), partition);
            }
            Map<String, String> partsConsumerMap =
                    consumeGroupInfo.getPartitionInfoMap();
            for (Entry<String, String> entry : partsConsumerMap.entrySet()) {
                Partition foundPart = partitionMap.get(entry.getKey());
                if (foundPart != null) {
                    String consumerId = entry.getValue();
                    Map<String, Map<String, Partition>> topicSubPartMap =
                            finalSubInfoMap.computeIfAbsent(consumerId, k -> new HashMap<>());
                    Map<String, Partition> partMap =
                            topicSubPartMap.computeIfAbsent(
                                    foundPart.getTopic(), k -> new HashMap<>());
                    partMap.put(foundPart.getPartitionKey(), foundPart);
                    partitionMap.remove(entry.getKey());
                }
            }
            if (consumeGroupInfo.addAllocatedTimes() > 0) {
                MasterSrvStatsHolder.updSvrBalResetDurations(
                        System.currentTimeMillis() - consumeGroupInfo.getCreateTime());
            }
        }
        return finalSubInfoMap;
    }
Reset balance

@param clusterState
@param consumerHolder
@param brokerRunManager
@param groupSet
@param defMetaDataService
@param strBuffer
@return
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/balance/DefaultLoadBalancer.java