private void commitTransaction(long checkpointId) {
        boolean succeed = true;

        List<Long> commitCheckpointIds = snapshotMap.keySet().stream()
                .filter(cpId -> cpId <= checkpointId)
                .sorted(Long::compare)
                .collect(Collectors.toList());

        for (Long cpId : commitCheckpointIds) {
            try {
                for (StreamLoadSnapshot snapshot : snapshotMap.get(cpId)) {
                    if (!sinkManager.commit(snapshot)) {
                        succeed = false;
                        break;
                    }
                }

                if (!succeed) {
                    throw new RuntimeException(String.format("Failed to commit some transactions for snapshot %s, " +
                            "please check taskmanager logs for details", cpId));
                }
            } catch (Exception e) {
                log.error("Failed to notify checkpoint complete, checkpoint id : {}", checkpointId, e);
                throw new RuntimeException("Failed to notify checkpoint complete for checkpoint id " + checkpointId, e);
            }

            snapshotMap.remove(cpId);
        }

        // set legacyState to null to avoid clear it in latter snapshotState
        legacyState = null;
    }
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