private void commitOperation(
            SnapshotUpdate<?> operation,
            CommitSummary summary,
            String description,
            String newFlinkJobId,
            String operatorId,
            long checkpointId) {
        LOG.info(
                "Committing {} for checkpoint {} to table {} branch {} with summary: {}",
                description,
                checkpointId,
                table.name(),
                branch,
                summary);
        snapshotProperties.forEach(operation::set);
        // custom snapshot metadata properties will be overridden if they conflict with internal ones
        // used by the sink.
        operation.set(MAX_COMMITTED_CHECKPOINT_ID, Long.toString(checkpointId));
        operation.set(FLINK_JOB_ID, newFlinkJobId);
        operation.set(OPERATOR_ID, operatorId);
        operation.toBranch(branch);

        long startNano = System.nanoTime();
        operation.commit(); // abort is automatically called if this fails.
        long durationMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNano);
        LOG.info(
                "Committed {} to table: {}, branch: {}, checkpointId {} in {} ms",
                description,
                table.name(),
                branch,
                checkpointId,
                durationMs);
        committerMetrics.commitDuration(durationMs);
    }
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