@Override
    public TopicFetcher addTopic(InLongTopic topic) {

        try {
            TopicFetcher result = fetchers.get(topic.getTopicKey());
            if (result == null) {
                // create fetcher (pulsar,tube,kafka)
                TopicFetcher topicFetcher = createInLongTopicFetcher(topic);
                TopicFetcher preValue = fetchers.putIfAbsent(topic.getTopicKey(), topicFetcher);
                LOGGER.info("addFetcher :{}", topic.getTopicKey());
                result = topicFetcher;
                if (preValue != null) {
                    result = preValue;
                    if (topicFetcher != null) {
                        topicFetcher.close();
                    }
                    LOGGER.info("addFetcher create same fetcher {}", topic);
                }
            }
            return result;
        } catch (Throwable t) {
            LOGGER.error("got error when add fetcher: {}", t.getMessage(), t);
            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 deprecated since 1.9.0
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManager.java