private boolean createConsumer(PulsarClient client) {
        if (null == client) {
            LOGGER.error("pulsar client is null");
            return false;
        }
        try {
            SubscriptionInitialPosition position = SubscriptionInitialPosition.Latest;
            SortClientConfig.ConsumeStrategy offsetResetStrategy = context.getConfig().getOffsetResetStrategy();
            if (offsetResetStrategy == SortClientConfig.ConsumeStrategy.earliest
                    || offsetResetStrategy == SortClientConfig.ConsumeStrategy.earliest_absolutely) {
                LOGGER.info("the subscription initial position is earliest!");
                position = SubscriptionInitialPosition.Earliest;
            }

            consumer = client.newConsumer(Schema.BYTES)
                    .topic(topic.getTopic())
                    .subscriptionName(context.getConfig().getSubscription())
                    .subscriptionType(SubscriptionType.Shared)
                    .startMessageIdInclusive()
                    .subscriptionInitialPosition(position)
                    .ackTimeout(context.getConfig().getAckTimeoutSec(), TimeUnit.SECONDS)
                    .receiverQueueSize(context.getConfig().getPulsarReceiveQueueSize())
                    .subscribe();

            this.seeker = SeekerFactory.createPulsarSeeker(consumer, topic);
            String threadName = String.format("sort_sdk_pulsar_single_topic_fetch_thread_%s_%s_%d",
                    this.topic.getInLongCluster().getClusterId(), topic.getTopic(), this.hashCode());
            this.fetchThread = new Thread(new PulsarSingleTopicFetcher.Fetcher(), threadName);
            this.fetchThread.setDaemon(true);
            this.fetchThread.start();
        } catch (Exception e) {
            LOGGER.error("fail to create consumer", e);
            return false;
        }
        return true;
    }
create Consumer and fetch thread

@return boolean
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/pulsar/PulsarSingleTopicFetcher.java