private void commitPendingResult(
            NavigableMap<Long, WriteResult> pendingResults,
            CommitSummary summary,
            String newFlinkJobId,
            String operatorId,
            long checkpointId) {
        long totalFiles = summary.dataFilesCount() + summary.deleteFilesCount();
        continuousEmptyCheckpoints = totalFiles == 0 ? continuousEmptyCheckpoints + 1 : 0;
        if (totalFiles != 0 || continuousEmptyCheckpoints % maxContinuousEmptyCommits == 0) {
            if (replacePartitions) {
                replacePartitions(pendingResults, summary, newFlinkJobId, operatorId, checkpointId);
            } else {
                commitDeltaTxn(pendingResults, summary, newFlinkJobId, operatorId, checkpointId);
            }
            continuousEmptyCheckpoints = 0;
        } else {
            LOG.info("Skip commit for checkpoint {} due to no data files or delete files.", checkpointId);
        }
    }
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