private void batchAddTopic() {
        pendingTopics.forEach((clusterId, queue) -> {
            Map<String, TopicFuture> pendingTopicList = new HashMap<>(32);
            if (notSatisfiedCount.get() > queueMaxWait || queue.size() > queueMaxRunningSize) {
                notSatisfiedCount.set(0);
                List<TopicFuture> tmpTopicList = new ArrayList<>();
                queue.drainTo(tmpTopicList, queueMaxRunningSize);
                for (TopicFuture topicFuture : tmpTopicList) {
                    pendingTopicList.put(topicFuture.getEntry().getTopic(), topicFuture);
                }
            } else {
                notSatisfiedCount.incrementAndGet();
            }
            // update broker status
            nodeService.updateBrokerStatus(clusterId, pendingTopicList);
        });

    }
batch executing adding topic, wait util max n seconds or max size satisfied.
inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TopicBackendWorker.java