private void handleAddingTopic(MasterEntry masterEntry,
            TubeHttpBrokerInfoList brokerInfoList,
            Map<String, TopicTaskEntry> pendingTopic) {
        // check tubemq cluster by topic name, remove pending topic if has added.
        Set<String> brandNewTopics = new HashSet<>();
        for (String topic : pendingTopic.keySet()) {
            TubeHttpTopicInfoList topicInfoList = topicService.requestTopicConfigInfo(masterEntry, topic);
            if (ValidateUtils.isNull(topicInfoList)) {
                return;
            }
            // get broker list by topic request
            List<Integer> topicBrokerList = topicInfoList.getTopicBrokerIdList();
            if (topicBrokerList.isEmpty()) {
                brandNewTopics.add(topic);
            } else {
                handleAddingExistTopics(masterEntry, brokerInfoList, topic, topicBrokerList);
            }
        }
        handleAddingNewTopics(masterEntry, brokerInfoList, brandNewTopics);
    }
Add topic info

@param clusterId this is the cluster id
@param topicTasks topic info
inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/executors/AddTopicExecutor.java