public synchronized void createProducersByTopicSet(Set<String> cfgTopicSet) throws Exception {
        if (cfgTopicSet == null || cfgTopicSet.isEmpty()) {
            return;
        }
        // filter published topics
        List<String> filteredTopics = new ArrayList<>(cfgTopicSet.size());
        for (String topicName : cfgTopicSet) {
            if (StringUtils.isBlank(topicName)
                    || producerMap.get(topicName) != null) {
                continue;
            }
            filteredTopics.add(topicName);
        }
        if (filteredTopics.isEmpty()) {
            return;
        }
        // alloc topic count
        Collections.sort(filteredTopics);
        long startTime = System.currentTimeMillis();
        int maxPublishTopicCnt = clusterConfig.getMaxTopicsEachProducerHold();
        int allocTotalCnt = filteredTopics.size();
        List<Integer> topicGroupCnt = new ArrayList<>();
        int paddingCnt = (lastPubTopicCnt.get() <= 0)
                ? 0
                : (maxPublishTopicCnt - lastPubTopicCnt.get());
        while (allocTotalCnt > 0) {
            if (paddingCnt > 0) {
                topicGroupCnt.add(Math.min(allocTotalCnt, paddingCnt));
                allocTotalCnt -= paddingCnt;
                paddingCnt = 0;
            } else {
                topicGroupCnt.add(Math.min(allocTotalCnt, maxPublishTopicCnt));
                allocTotalCnt -= maxPublishTopicCnt;
            }
        }
        // create producer
        int startPos = 0;
        int endPos = 0;
        Set<String> subTopicSet = new HashSet<>();
        for (Integer dltCnt : topicGroupCnt) {
            // allocate topic items
            subTopicSet.clear();
            endPos = startPos + dltCnt;
            for (int index = startPos; index < endPos; index++) {
                subTopicSet.add(filteredTopics.get(index));
            }
            startPos = endPos;
            // create producer
            if (lastProducer == null
                    || lastPubTopicCnt.get() == maxPublishTopicCnt) {
                lastProducer = sessionFactory.createProducer();
                lastPubTopicCnt.set(0);
            }
            try {
                lastProducer.publish(subTopicSet);
            } catch (Throwable e) {
                logger.info(sinkName + " meta sink publish fail.", e);
            }
            lastPubTopicCnt.addAndGet(subTopicSet.size());
            for (String topicItem : subTopicSet) {
                producerMap.put(topicItem, lastProducer);
            }
        }
        logger.info(sinkName + " initializes producers for topics:"
                + producerMap.keySet() + ", cost: " + (System.currentTimeMillis() - startTime)
                + "ms");
    }
Create sink producers by configured topic set
group topicSet to different group, each group is associated with a producer

@param cfgTopicSet  the configured topic set
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/TubeProducerHolder.java