private void createTubeConsumerCreator(CacheZoneCluster cluster) {
        LOGGER.info("start to init tube creator for cluster={}", cluster);
        if (cluster.getBootstraps() != null) {
            try {
                TubeClientConfig tubeConfig = new TubeClientConfig(cluster.getBootstraps());
                MessageSessionFactory messageSessionFactory = new TubeSingleSessionFactory(tubeConfig);
                TubeConsumerCreator tubeConsumerCreator = new TubeConsumerCreator(messageSessionFactory,
                        tubeConfig);
                TubeConsumerCreator oldCreator = tubeFactories.putIfAbsent(cluster.getClusterId(), tubeConsumerCreator);
                if (oldCreator != null) {
                    LOGGER.warn("close new tube creator for cluster={}", cluster);
                    tubeConsumerCreator.getMessageSessionFactory().shutdown();
                }
            } catch (Exception e) {
                LOGGER.error("create tube creator error for cluster={}", cluster, e);
                return;
            }
            LOGGER.info("success to init tube creatorfor cluster={}", cluster);
        } else {
            LOGGER.error("bootstrap is null for cluster={}", cluster);
        }
    }
Inlong manager that maintain the single topic fetchers.
It is suitable to the cases that each topic has its own configurations.
And each consumer only consume the very one topic.
InlongMultiTopicManager was used since 1.9.0
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongTopicManager.java