private TopicFetcher subscribeMultiTopic() {
        InLongTopic firstTopic = topics.stream().findFirst().get();
        interceptor = Optional.ofNullable(interceptor).orElse(new MsgTimeInterceptor());
        interceptor.configure(firstTopic);
        String key = Optional.ofNullable(fetchKey)
                .orElse(firstTopic.getInLongCluster().getClusterId() + new Random().nextLong());
        TopicFetcher fetcher =
                new PulsarMultiTopicsFetcher(topics, context, interceptor, deserializer, pulsarClient, key);
        if (!fetcher.init()) {
            throw new IllegalStateException("init pulsar multi topic fetcher failed");
        }
        return fetcher;
    }
Builder of pulsar single topic fetcher.
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/pulsar/PulsarTopicFetcherBuilderImpl.java