@Override
    public Map<String, Map<String, List<Partition>>> bukAssign(
            ConsumerInfoHolder consumerHolder,
            BrokerRunManager brokerRunManager,
            List<String> groupSet,
            MetaDataService defMetaDataService,
            StringBuilder strBuffer) {
        // #lizard forgives
        // regular consumer allocate operation
        Map<String, Map<String, List<Partition>>> finalSubInfoMap =
                new HashMap<>();
        for (String group : groupSet) {
            ConsumeGroupInfo consumeGroupInfo =
                    consumerHolder.getConsumeGroupInfo(group);
            if (consumeGroupInfo == null
                    || consumeGroupInfo.isClientBalance()
                    || consumeGroupInfo.isUnReadyServerBalance()) {
                continue;
            }
            List<ConsumerInfo> consumerList = consumeGroupInfo.getConsumerInfoList();
            if (CollectionUtils.isEmpty(consumerList)) {
                continue;
            }
            // check if current client meet minimal requirements
            Set<String> topicSet = consumeGroupInfo.getTopicSet();
            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 / consumerList.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 1] 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);
            }
            // sort and mod
            Collections.sort(consumerList);
            for (String topic : topicSet) {
                List<Partition> partPubList =
                        brokerRunManager.getSubBrokerAcceptSubParts(topic);
                Collections.sort(partPubList);
                int partsPerConsumer = partPubList.size() / consumerList.size();
                int consumersWithExtraPart = partPubList.size() % consumerList.size();
                for (int i = 0; i < consumerList.size(); i++) {
                    String consumerId = consumerList.get(i).getConsumerId();
                    Map<String, List<Partition>> topicSubPartMap =
                            finalSubInfoMap.computeIfAbsent(consumerId, k -> new HashMap<>());
                    List<Partition> partList =
                            topicSubPartMap.computeIfAbsent(topic, k -> new ArrayList<>());
                    int startIndex = partsPerConsumer * i + Math.min(i, consumersWithExtraPart);
                    int parts = partsPerConsumer + ((i + 1) > consumersWithExtraPart ? 0 : 1);
                    for (int j = startIndex; j < startIndex + parts; j++) {
                        final Partition part = partPubList.get(j);
                        partList.add(part);
                    }
                }
            }
        }
        return finalSubInfoMap;
    }
Assign consumer partitions

@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