public Map<TopicRange, MessageId> offsetForEachTopic(
            Set<TopicRange> topics,
            StartupMode mode,
            Map<TopicRange, MessageId> specificStartupOffsets) {

        switch (mode) {
            case LATEST:
                return topics.stream()
                        .collect(Collectors.toMap(k -> k, k -> MessageId.latest));
            case EARLIEST:
                return topics.stream()
                        .collect(Collectors.toMap(k -> k, k -> MessageId.earliest));
            case SPECIFIC_OFFSETS:
                checkArgument(topics.containsAll(specificStartupOffsets.keySet()),
                        String.format(
                                "Topics designated in startingOffsets should appear in %s, topics:" +
                                        "%s, topics in offsets: %s",
                                StringUtils.join(PulsarOptions.TOPIC_OPTION_KEYS),
                                StringUtils.join(topics.toArray()),
                                StringUtils.join(specificStartupOffsets.entrySet().toArray())));

                Map<TopicRange, MessageId> specificOffsets = new HashMap<>();
                for (TopicRange topic : topics) {
                    if (specificStartupOffsets.containsKey(topic)) {
                        specificOffsets.put(topic, specificStartupOffsets.get(topic));
                    } else {
                        specificOffsets.put(topic, MessageId.latest);
                    }
                }
                return specificOffsets;
            case EXTERNAL_SUBSCRIPTION:
                Map<TopicRange, MessageId> offsetsFromSubs = new HashMap<>();
                for (TopicRange topic : topics) {
                    offsetsFromSubs.put(
                            topic,
                            metadataReader.getPositionFromSubscription(
                                    topic, subscriptionPosition));
                }
                log.info("offset for each topic: {}", offsetsFromSubs);
                return offsetsFromSubs;
        }
        return null;
    }
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