@Override
    public Map<String, Map<String, List<Partition>>> balanceCluster(
            Map<String, Map<String, Map<String, Partition>>> clusterState,
            ConsumerInfoHolder consumerHolder,
            BrokerRunManager brokerRunManager,
            List<String> groupSet,
            MetaDataService defMetaDataService,
            StringBuilder strBuffer) {
        // #lizard forgives
        // load balance according to group
        Map<String/* consumer */, Map<String/* topic */, List<Partition>>> finalSubInfoMap =
                new HashMap<>();
        Map<String, RebProcessInfo> rejGroupClientInfoMap = new HashMap<>();
        Set<String> onlineOfflineGroupSet = new HashSet<>();
        Set<String> boundGroupSet = new HashSet<>();
        for (String group : groupSet) {
            if (group == null) {
                continue;
            }
            ConsumeGroupInfo consumeGroupInfo = consumerHolder.getConsumeGroupInfo(group);
            if (consumeGroupInfo == null || consumeGroupInfo.isClientBalance()) {
                continue;
            }
            List<ConsumerInfo> consumerList =
                    consumeGroupInfo.getConsumerInfoList();
            if (CollectionUtils.isEmpty(consumerList)) {
                continue;
            }
            // deal with regular consumer allocation, bound consume not in this part
            if (consumeGroupInfo.isUnReadyServerBalance()) {
                boundGroupSet.add(group);
                continue;
            }
            List<ConsumerInfo> newConsumerList = new ArrayList<>();
            for (ConsumerInfo consumerInfo : consumerList) {
                if (consumerInfo != null) {
                    newConsumerList.add(consumerInfo);
                }
            }
            if (CollectionUtils.isEmpty(newConsumerList)) {
                continue;
            }
            Set<String> topicSet = consumeGroupInfo.getTopicSet();
            if (consumeGroupInfo.needResourceCheck()) {
                // check if current client meet minimal requirements
                GroupResCtrlEntity offsetResetGroupEntity =
                        defMetaDataService.getGroupCtrlConf(group);
                int confAllowBClientRate = (offsetResetGroupEntity != null
                        && offsetResetGroupEntity.getAllowedBrokerClientRate() > 0)
                                ? offsetResetGroupEntity.getAllowedBrokerClientRate()
                                : -2;
                int allowRate = confAllowBClientRate > 0
                        ? confAllowBClientRate
                        : consumerHolder.getDefResourceRate();
                int maxBrokerCount =
                        brokerRunManager.getSubTopicMaxBrokerCount(topicSet);
                int curBClientRate = (int) Math.floor(maxBrokerCount / newConsumerList.size());
                if (curBClientRate > allowRate) {
                    int minClientCnt = maxBrokerCount / allowRate;
                    if (maxBrokerCount % allowRate != 0) {
                        minClientCnt += 1;
                    }
                    consumeGroupInfo.setConsumeResourceInfo(confAllowBClientRate,
                            curBClientRate, minClientCnt, false);
                    if (consumeGroupInfo.isEnableBalanceChkPrint()) {
                        logger.info(strBuffer.append("[UnBound Alloc 2] Not allocate partition :group(")
                                .append(group).append(")'s consumer getCachedSize(")
                                .append(consumeGroupInfo.getGroupCnt())
                                .append(") low than min required client count:")
                                .append(minClientCnt).toString());
                        strBuffer.delete(0, strBuffer.length());
                    }
                    continue;
                } else {
                    consumeGroupInfo.setConsumeResourceInfo(confAllowBClientRate,
                            curBClientRate, -2, true);
                }
            }
            RebProcessInfo rebProcessInfo = new RebProcessInfo();
            if (!consumeGroupInfo.isBalanceMapEmpty()) {
                rebProcessInfo = consumerHolder.getNeedRebNodeList(group);
                if (!rebProcessInfo.isProcessInfoEmpty()) {
                    rejGroupClientInfoMap.put(group, rebProcessInfo);
                }
            }
            List<ConsumerInfo> newConsumerList2 = new ArrayList<>();
            Map<String, Partition> partMap =
                    brokerRunManager.getSubBrokerAcceptSubParts(topicSet);
            Map<String, NodeRebInfo> rebProcessInfoMap = consumeGroupInfo.getBalanceMap();
            for (ConsumerInfo consumer : newConsumerList) {
                Map<String, List<Partition>> partitions = new HashMap<>();
                finalSubInfoMap.put(consumer.getConsumerId(), partitions);
                Map<String, Map<String, Partition>> relation = clusterState.get(consumer.getConsumerId());
                if (relation != null) {
                    // filter client which can not meet requirements
                    if (rebProcessInfo.needProcessList.contains(consumer.getConsumerId())
                            || rebProcessInfo.needEscapeList.contains(consumer.getConsumerId())) {
                        NodeRebInfo tmpNodeRegInfo =
                                rebProcessInfoMap.get(consumer.getConsumerId());
                        if (tmpNodeRegInfo != null
                                && tmpNodeRegInfo.getReqType() == 0) {
                            newConsumerList2.add(consumer);
                        }
                        for (Entry<String, Map<String, Partition>> entry : relation.entrySet()) {
                            partitions.put(entry.getKey(), new ArrayList<>());
                        }
                        continue;
                    }
                    newConsumerList2.add(consumer);
                    for (Entry<String, Map<String, Partition>> entry : relation.entrySet()) {
                        List<Partition> ps = new ArrayList<>();
                        partitions.put(entry.getKey(), ps);
                        Map<String, Partition> partitionMap = entry.getValue();
                        if (partitionMap != null && !partitionMap.isEmpty()) {
                            for (Partition partition : partitionMap.values()) {
                                Partition curPart = partMap.remove(partition.getPartitionKey());
                                if (curPart != null) {
                                    ps.add(curPart);
                                }
                            }
                        }
                    }
                }
            }
            // random allocate
            if (!partMap.isEmpty()) {
                onlineOfflineGroupSet.add(group);
                if (!newConsumerList2.isEmpty()) {
                    this.randomAssign(partMap, newConsumerList2,
                            finalSubInfoMap, clusterState, rebProcessInfo.needProcessList);
                }
            }
        }
        List<String> groupsNeedToBalance = new ArrayList<>();
        if (onlineOfflineGroupSet.isEmpty()) {
            for (String group : groupSet) {
                if (group == null) {
                    continue;
                }
                groupsNeedToBalance.add(group);
            }
        } else {
            for (String group : groupSet) {
                if (group == null) {
                    continue;
                }
                if (!onlineOfflineGroupSet.contains(group)) {
                    groupsNeedToBalance.add(group);
                }
            }
        }
        if (!boundGroupSet.isEmpty()) {
            for (String group : boundGroupSet) {
                groupsNeedToBalance.remove(group);
            }
        }
        if (!groupsNeedToBalance.isEmpty()) {
            balance(finalSubInfoMap, consumerHolder, brokerRunManager,
                    groupsNeedToBalance, clusterState, rejGroupClientInfoMap);
        }
        if (!rejGroupClientInfoMap.isEmpty()) {
            for (Entry<String, RebProcessInfo> entry : rejGroupClientInfoMap.entrySet()) {
                consumerHolder.setRebNodeProcessed(entry.getKey(),
                        entry.getValue().needProcessList);
            }
        }
        return finalSubInfoMap;
    }
Load 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