private void commitUpToCheckpoint(
            NavigableMap<Long, byte[]> deltaManifestsMap,
            String newFlinkJobId,
            String operatorId,
            long checkpointId)
            throws IOException {
        NavigableMap<Long, byte[]> pendingMap = deltaManifestsMap.headMap(checkpointId, true);
        List<ManifestFile> manifests = Lists.newArrayList();
        NavigableMap<Long, WriteResult> pendingResults = Maps.newTreeMap();
        for (Map.Entry<Long, byte[]> e : pendingMap.entrySet()) {
            if (Arrays.equals(EMPTY_MANIFEST_DATA, e.getValue())) {
                // Skip the empty flink manifest.
                continue;
            }

            DeltaManifests deltaManifests =
                    SimpleVersionedSerialization.readVersionAndDeSerialize(
                            DeltaManifestsSerializer.INSTANCE, e.getValue());
            pendingResults.put(
                    e.getKey(),
                    FlinkManifestUtil.readCompletedFiles(deltaManifests, table.io(), table.specs()));
            manifests.addAll(deltaManifests.manifests());
        }

        CommitSummary summary = new CommitSummary(pendingResults);
        commitPendingResult(pendingResults, summary, newFlinkJobId, operatorId, checkpointId);
        committerMetrics.updateCommitSummary(summary);
        pendingMap.clear();
        deleteCommittedManifests(manifests, newFlinkJobId, 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