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

        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));
            default:
                throw new IllegalArgumentException(
                        "Unknown startup mode option: " + mode);
        }
    }
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/FlinkPulsarSourceWithoutAdmin.java