@Override
    public void notifyCheckpointComplete(long checkpointId) throws Exception {
        super.notifyCheckpointComplete(checkpointId);
        // It's possible that we have the following events:
        // 1. snapshotState(ckpId);
        // 2. snapshotState(ckpId+1);
        // 3. notifyCheckpointComplete(ckpId+1);
        // 4. notifyCheckpointComplete(ckpId);
        // For step#4, we don't need to commit iceberg table again because in step#3 we've committed all
        // the files,
        // Besides, we need to maintain the max-committed-checkpoint-id to be increasing.
        if (checkpointId > maxCommittedCheckpointId) {
            LOG.info("Checkpoint {} completed. Attempting commit.", checkpointId);
            commitUpToCheckpoint(dataFilesPerCheckpoint, flinkJobId, operatorUniqueId, checkpointId);
            this.maxCommittedCheckpointId = checkpointId;
        } else {
            LOG.info(
                    "Skipping committing checkpoint {}. {} is already committed.",
                    checkpointId,
                    maxCommittedCheckpointId);
        }
    }
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