@Async("asyncExecutor")
    public void addTopicConfig(Long clusterId, List<TopicTaskEntry> topicTasks) {
        if (CollectionUtils.isEmpty(topicTasks)) {
            return;
        }
        List<List<TopicTaskEntry>> taskParts = ListUtils.partition(topicTasks, MAX_CONFIG_TOPIC_NUM);

        MasterEntry masterNode = masterService.getMasterNode(clusterId);
        TubeHttpBrokerInfoList brokerInfoList = nodeService.requestBrokerStatus(masterNode);
        if (ValidateUtils.isNull(brokerInfoList)) {
            return;
        }

        for (List<TopicTaskEntry> taskPart : taskParts) {
            Map<String, TopicTaskEntry> topicEntryMap = taskPart.stream().collect(
                    Collectors.toMap(TopicTaskEntry::getTopicName, topicTaskEntry -> topicTaskEntry, (v1, v2) -> v2));
            doConfigTopics(topicEntryMap, masterNode, brokerInfoList);
        }

    }
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