@Override
    public TopicFetcher subscribe() {
        Optional.ofNullable(topic)
                .orElseThrow(() -> new IllegalStateException("subscribe tube single topic, but never assign topic"));
        Optional.ofNullable(context)
                .orElseThrow(() -> new IllegalStateException("context is null"));
        Optional.ofNullable(tubeConsumerCreator)
                .orElseThrow(() -> new IllegalStateException("tube consumer creator is null"));
        interceptor = Optional.ofNullable(interceptor).orElse(new MsgTimeInterceptor());
        interceptor.configure(topic);
        deserializer = Optional.ofNullable(deserializer).orElse(new MessageDeserializer());
        TubeSingleTopicFetcher fetcher =
                new TubeSingleTopicFetcher(topic, context, interceptor, deserializer, tubeConsumerCreator);
        if (!fetcher.init()) {
            throw new IllegalStateException("init tube single topic fetcher failed");
        }
        return fetcher;
    }
Builder of tube single topic fetcher.
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/tube/TubeTopicFetcherBuilderImpl.java