@Override
    public void snapshotState(StateSnapshotContext context) throws Exception {
        super.snapshotState(context);
        long checkpointId = context.getCheckpointId();
        LOG.info(
                "Start to flush snapshot state to state backend, table: {}, checkpointId: {}",
                table,
                checkpointId);

        // Update the checkpoint state.
        long startNano = System.nanoTime();
        dataFilesPerCheckpoint.put(checkpointId, writeToManifest(checkpointId));
        // Reset the snapshot state to the latest state.
        checkpointsState.clear();
        checkpointsState.add(dataFilesPerCheckpoint);

        jobIdState.clear();
        jobIdState.add(flinkJobId);

        // Clear the local buffer for current checkpoint.
        writeResultsOfCurrentCkpt.clear();
        committerMetrics.checkpointDuration(
                TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNano));
    }
Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergFilesCommitter.java