@Override
    public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
        // record the start time of each checkpoint
        checkpointStartTimeMap.put(functionSnapshotContext.getCheckpointId(), System.currentTimeMillis());

        updateCurrentCheckpointId(functionSnapshotContext.getCheckpointId());
        sinkManager.flush();

        if (sinkOptions.getSemantic() != StarRocksSinkSemantic.EXACTLY_ONCE) {
            if (sinkExactlyMetric != null) {
                sinkExactlyMetric.incNumSnapshotCreate();
            }
            return;
        }

        StreamLoadSnapshot snapshot = sinkManager.snapshot();

        if (sinkManager.prepare(snapshot)) {
            snapshotMap.put(functionSnapshotContext.getCheckpointId(), Collections.singletonList(snapshot));

            snapshotStates.clear();
            snapshotStates.add(StarrocksSnapshotState.of(snapshotMap));
            if (sinkExactlyMetric != null) {
                sinkExactlyMetric.incNumSnapshotCreate();
            }
        } else {
            sinkManager.abort(snapshot);
            checkpointStartTimeMap.remove(functionSnapshotContext.getCheckpointId());
            if (sinkExactlyMetric != null) {
                sinkExactlyMetric.incNumSnapshotError();
            }
            throw new RuntimeException("Snapshot state failed by prepare");
        }

        if (legacyState != null) {
            legacyState.clear();
        }
    }
The map to store the start time of each checkpoint.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/table/StarRocksDynamicSinkFunctionV2.java