public Map<TopicRange, ReaderThread<T>> createAndStartReaderThread(
            List<PulsarTopicState<T>> states,
            ExceptionProxy exceptionProxy) {
        Map<TopicRange, ReaderThread<T>> topic2Threads = new HashMap<>();

        for (PulsarTopicState state : states) {
            ReaderThread<T> readerT = createReaderThread(exceptionProxy, state);
            readerT.setName(String.format(
                    "Pulsar Reader for %s in task %s",
                    state.getTopicRange(),
                    runtimeContext.getTaskName()));
            readerT.setDaemon(true);
            readerT.start();
            log.info("Starting Thread {}", readerT.getName());
            topic2Threads.put(state.getTopicRange(), readerT);
        }
        return topic2Threads;
    }
Takes a snapshot of the partition offsets.

<p>Important: This method must be called under the checkpoint lock.

@return A map from partition to current offset.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/PulsarFetcher.java