private TopicFetcher onlineNewTopic(InLongTopic topic) {
        try {
            if (InlongTopicTypeEnum.PULSAR.getName().equalsIgnoreCase(topic.getTopicType())) {
                LOGGER.info("the topic is pulsar {}", topic);
                return TopicFetcherBuilder.newPulsarBuilder()
                        .pulsarClient(pulsarClients.get(topic.getInLongCluster().getClusterId()))
                        .topic(topic)
                        .context(context)
                        .subscribe();
            } else if (InlongTopicTypeEnum.KAFKA.getName().equalsIgnoreCase(topic.getTopicType())) {
                LOGGER.info("the topic is kafka {}", topic);
                return TopicFetcherBuilder.newKafkaBuilder()
                        .bootstrapServers(topic.getInLongCluster().getBootstraps())
                        .topic(topic)
                        .context(context)
                        .subscribe();
            } else if (InlongTopicTypeEnum.TUBE.getName().equalsIgnoreCase(topic.getTopicType())) {
                LOGGER.info("the topic is tube {}", topic);
                return TopicFetcherBuilder.newTubeBuilder()
                        .tubeConsumerCreater(tubeFactories.get(topic.getInLongCluster().getClusterId()))
                        .topic(topic)
                        .context(context)
                        .subscribe();
            } else {
                LOGGER.error("topic type not support " + topic.getTopicType());
                return null;
            }
        } catch (Exception e) {
            LOGGER.error("failed to subscribe new topic={}", topic, e);
            return null;
        }

    }
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