tubemq-manager/src/main/java/org/apache/tubemq/manager/service/NodeService.java
/**
     * Adding topic is an async operation, so this method should
     * 1. check whether pendingTopic contains topic that has failed/succeeded to be added.
     * 2. async add topic to tubemq cluster
     *
     * @param brokerInfoList - broker list
     * @param pendingTopic - topicMap
     */
Adding topic is an async operation, so this method should
1. check whether pendingTopic contains topic that has failed/succeeded to be added.
2. async add topic to tubemq cluster

@param brokerInfoList - broker list
@param pendingTopic - topicMap
Adding topic is an async operation so this method should 1 . check whether pendingTopic contains topic that has failed / succeeded to be added . 2 . async add topic to tubemq cluster @param brokerInfoList - broker list @param pendingTopic - topicMap
private void handleAddingTopic(NodeEntry nodeEntry,
            TubeHttpBrokerInfoList brokerInfoList,
            Map<String, TopicFuture> pendingTopic) {
        Set<String> brandNewTopics = new HashSet<>();
        for (String topic : pendingTopic.keySet()) {
            TubeHttpTopicInfoList topicInfoList = requestTopicConfigInfo(nodeEntry, topic);
            if (topicInfoList != null) {
                List<Integer> topicBrokerList = topicInfoList.getTopicBrokerIdList();
                if (topicBrokerList.isEmpty()) {
                    brandNewTopics.add(topic);
                } else {
                    List<Integer> configurableBrokerIdList =
                            brokerInfoList.getConfigurableBrokerIdList();
                    configurableBrokerIdList.removeAll(topicBrokerList);
                    Set<String> singleTopic = new HashSet<>();
                    singleTopic.add(topic);
                    int maxBrokers = maxConfigurableBrokerSize - topicBrokerList.size();
                    boolean isSuccess = configBrokersForTopics(nodeEntry, singleTopic,
                            configurableBrokerIdList, maxBrokers);
                    handleAddingResult(isSuccess, singleTopic, pendingTopic);
                }
            }
        }
        List<Integer> configurableBrokerIdList = brokerInfoList.getConfigurableBrokerIdList();
        int maxBrokers = Math.min(maxConfigurableBrokerSize, configurableBrokerIdList.size());
        boolean isSuccess = configBrokersForTopics(nodeEntry, brandNewTopics,
                configurableBrokerIdList, maxBrokers);
        handleAddingResult(isSuccess, brandNewTopics, pendingTopic);
    }
private void handleAddingTopic ( NodeEntry nodeEntry , TubeHttpBrokerInfoList brokerInfoList , Map < String , TopicFuture > pendingTopic ) { Set < String > brandNewTopics = new HashSet < > ( ) ; for ( String topic : pendingTopic . keySet ( ) ) { TubeHttpTopicInfoList topicInfoList = requestTopicConfigInfo ( nodeEntry , topic ) ; if ( topicInfoList != null ) { List < Integer > topicBrokerList = topicInfoList . getTopicBrokerIdList ( ) ; if ( topicBrokerList . isEmpty ( ) ) { brandNewTopics . add ( topic ) ; } else { List < Integer > configurableBrokerIdList = brokerInfoList . getConfigurableBrokerIdList ( ) ; configurableBrokerIdList . removeAll ( topicBrokerList ) ; Set < String > singleTopic = new HashSet < > ( ) ; singleTopic . add ( topic ) ; int maxBrokers = maxConfigurableBrokerSize - topicBrokerList . size ( ) ; boolean isSuccess = configBrokersForTopics ( nodeEntry , singleTopic , configurableBrokerIdList , maxBrokers ) ; handleAddingResult ( isSuccess , singleTopic , pendingTopic ) ; } } } List < Integer > configurableBrokerIdList = brokerInfoList . getConfigurableBrokerIdList ( ) ; int maxBrokers = Math . min ( maxConfigurableBrokerSize , configurableBrokerIdList . size ( ) ) ; boolean isSuccess = configBrokersForTopics ( nodeEntry , brandNewTopics , configurableBrokerIdList , maxBrokers ) ; handleAddingResult ( isSuccess , brandNewTopics , pendingTopic ) ; }
Adding topic is an async operation, so this method should
 1. check whether pendingTopic contains topic that has failed/succeeded to be added.
Adding topic is an async operation so this method should 1 . check whether pendingTopic contains topic that has failed / succeeded to be added .
// 1. check tubemq cluster by topic name, remove pending topic if has added. // get broker list by topic request // remove brokers which have been added. // add topic to satisfy max broker number. // 2. add new topics to cluster