@Override
    public List<PulsarPartitionSplit> snapshotState(long checkpointId) {
        try {
            // record the start time of each checkpoint
            if (checkpointStartTimeMap != null) {
                checkpointStartTimeMap.put(checkpointId, System.currentTimeMillis());
            }
            if (sourceExactlyMetric != null) {
                sourceExactlyMetric.incNumSnapshotCreate();
            }

            if (deserializationSchema instanceof PulsarTableDeserializationSchema) {
                ((PulsarTableDeserializationSchema) deserializationSchema).updateCurrentCheckpointId(checkpointId);
            }
            List<PulsarPartitionSplit> splits = super.snapshotState(checkpointId);

            // Perform a snapshot for these splits.
            Map<TopicPartition, MessageId> cursors =
                    cursorsToCommit.computeIfAbsent(checkpointId, id -> new HashMap<>());
            // Put the cursors of the active splits.
            for (PulsarPartitionSplit split : splits) {
                MessageId latestConsumedId = split.getLatestConsumedId();
                if (latestConsumedId != null) {
                    cursors.put(split.getPartition(), latestConsumedId);
                }
            }
            // Put cursors of all the finished splits.
            cursors.putAll(cursorsOfFinishedSplits);

            return splits;
        } catch (Exception e) {
            if (sourceExactlyMetric != null) {
                sourceExactlyMetric.incNumSnapshotError();
            }
            throw e;
        }
    }
The map to store the start time of each checkpoint.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/source/reader/PulsarOrderedSourceReader.java