@Override
    public void notifyCheckpointComplete(long checkpointId) {
        LOG.debug("Committing cursors for checkpoint {}", checkpointId);

        Map<TopicPartition, MessageId> cursors = cursorsToCommit.get(checkpointId);
        try {
            ((PulsarOrderedFetcherManager<OUT>) splitFetcherManager).acknowledgeMessages(cursors);
            LOG.debug("Successfully acknowledge cursors for checkpoint {}", checkpointId);

            // Clean up the cursors.
            cursorsOfFinishedSplits.keySet().removeAll(cursors.keySet());
            cursorsToCommit.headMap(checkpointId + 1).clear();
            if (deserializationSchema instanceof PulsarTableDeserializationSchema) {
                PulsarTableDeserializationSchema pulsarTableDeserializationSchema =
                        (PulsarTableDeserializationSchema) deserializationSchema;
                pulsarTableDeserializationSchema.flushAudit();
                pulsarTableDeserializationSchema.updateLastCheckpointId(checkpointId);
            }
            // get the start time of the currently completed checkpoint
            if (checkpointStartTimeMap != null) {
                Long snapShotStartTimeById = checkpointStartTimeMap.remove(checkpointId);
                if (snapShotStartTimeById != null && sourceExactlyMetric != null) {
                    sourceExactlyMetric.incNumSnapshotComplete();
                    sourceExactlyMetric
                            .recordSnapshotToCheckpointDelay(System.currentTimeMillis() - snapShotStartTimeById);
                }
            } else {
                LOG.error("checkpointStartTimeMap is null, can't get the start time of checkpoint");
            }
        } catch (Exception e) {
            LOG.error("Failed to acknowledge cursors for checkpoint {}", checkpointId, e);
            cursorCommitThrowable.compareAndSet(null, 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