@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();
            }
            LOG.debug("Trigger the new transaction for downstream readers.");
            if (deserializationSchema instanceof PulsarTableDeserializationSchema) {
                ((PulsarTableDeserializationSchema) deserializationSchema).updateCurrentCheckpointId(checkpointId);
            }
            List<PulsarPartitionSplit> splits =
                    ((PulsarUnorderedFetcherManager<OUT>) splitFetcherManager).snapshotState();

            if (coordinatorClient == null) {
                return splits;
            }
            // Snapshot the transaction status and commit it after checkpoint finishing.
            List<TxnID> txnIDs =
                    transactionsToCommit.computeIfAbsent(checkpointId, id -> new ArrayList<>());
            for (PulsarPartitionSplit split : splits) {
                TxnID uncommittedTransactionId = split.getUncommittedTransactionId();
                if (uncommittedTransactionId != null) {
                    txnIDs.add(uncommittedTransactionId);
                }
            }
            return splits;
        } catch (Exception e) {
            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/PulsarUnorderedSourceReader.java