@Override
    public TopicFetcher subscribe() {
        Optional.ofNullable(context)
                .orElseThrow(() -> new IllegalStateException("context is null"));
        Optional.ofNullable(pulsarClient)
                .orElseThrow(() -> new IllegalStateException("pulsar client is null"));
        deserializer = Optional.ofNullable(deserializer).orElse(new MessageDeserializer());
        if (CollectionUtils.isNotEmpty(topics)) {
            return subscribeMultiTopic();
        } else if (Objects.nonNull(topic)) {
            return subscribeSingleTopic();
        } else {
            throw new IllegalArgumentException("subscribe pulsar fetcher, but never assign any topic");
        }
    }
Builder of pulsar single topic fetcher.
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/pulsar/PulsarTopicFetcherBuilderImpl.java