private TopicFetcher createInLongTopicFetcher(InLongTopic topic) {
        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;
        }
    }
create fetcher (pulsar,tube,kafka)

@param topic {@link InLongTopic}
@return {@link TopicFetcher}
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManager.java