@Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        if (!running) {
            log.debug("snapshotState() called on closed source");
        } else {

            flushAudit();

            unionOffsetStates.clear();

            PulsarFetcher<T> fetcher = this.pulsarFetcher;

            if (fetcher == null) {
                // the fetcher has not yet been initialized, which means we need to return the
                // originally restored offsets or the assigned partitions
                for (Map.Entry<TopicRange, MessageId> entry : ownedTopicStarts.entrySet()) {
                    final TopicSubscription topicSubscription = TopicSubscription.builder()
                            .topic(entry.getKey().getTopic())
                            .range(entry.getKey().getRange())
                            .subscriptionName(getSubscriptionName())
                            .build();
                    unionOffsetStates.add(Tuple2.of(topicSubscription, entry.getValue()));
                }
                pendingOffsetsToCommit.put(context.getCheckpointId(), restoredState);
            } else {
                Map<TopicRange, MessageId> currentOffsets = fetcher.snapshotCurrentState();
                pendingOffsetsToCommit.put(context.getCheckpointId(), currentOffsets);
                for (Map.Entry<TopicRange, MessageId> entry : currentOffsets.entrySet()) {
                    final TopicSubscription topicSubscription = TopicSubscription.builder()
                            .topic(entry.getKey().getTopic())
                            .range(entry.getKey().getRange())
                            .subscriptionName(getSubscriptionName())
                            .build();
                    unionOffsetStates.add(Tuple2.of(topicSubscription, entry.getValue()));
                }

                int exceed = pendingOffsetsToCommit.size() - MAX_NUM_PENDING_CHECKPOINTS;
                Iterator<Long> iterator = pendingOffsetsToCommit.keySet().iterator();

                while (iterator.hasNext() && exceed > 0) {
                    iterator.next();
                    iterator.remove();
                }
            }
        }
    }
Try to restore the old save point.

@param stateStore state store
@return state data
@throws Exception Type incompatibility, serialization failure
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSource.java