@Override
    public void notifyCheckpointComplete(long checkpointId) throws Exception {
        LOG.debug("Committing transactions for checkpoint {}", checkpointId);
        if (coordinatorClient == null) {
            return;
        }
        List<Long> checkpointIds =
                transactionsToCommit.keySet().stream()
                        .filter(id -> id <= checkpointId)
                        .collect(toList());
        for (Long id : checkpointIds) {
            List<TxnID> transactions = transactionsToCommit.remove(id);
            if (transactions != null) {
                for (TxnID transaction : transactions) {
                    coordinatorClient.commit(transaction);
                    transactionsOfFinishedSplits.remove(transaction);
                }
            }
        }
        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");
        }
    }
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/PulsarUnorderedSourceReader.java