private void doSnapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
        if (handover.hasError()) {
            LOG.debug("snapshotState() called on closed source");
            throw new FlinkRuntimeException(
                    "Call snapshotState() on closed source, checkpoint failed.");
        } else {
            snapshotOffsetState(functionSnapshotContext.getCheckpointId());
            snapshotHistoryRecordsState();
        }
        if (deserializer instanceof RowDataDebeziumDeserializeSchema) {
            ((RowDataDebeziumDeserializeSchema) deserializer)
                    .updateCurrentCheckpointId(functionSnapshotContext.getCheckpointId());
        }
        // record the start time of each checkpoint
        if (checkpointStartTimeMap != null) {
            checkpointStartTimeMap.put(functionSnapshotContext.getCheckpointId(), System.currentTimeMillis());
        } else {
            LOG.error("checkpointStartTimeMap is null, can't record the start time of checkpoint");
        }
        if (sourceExactlyMetric != null) {
            sourceExactlyMetric.incNumSnapshotCreate();
        }
    }
The map to store the start time of each checkpoint.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/DebeziumSourceFunction.java