public static DeltaManifests writeCompletedFiles(WriteResult result,
            Supplier<OutputFile> outputFileSupplier,
            PartitionSpec spec) throws IOException {

        ManifestFile dataManifest = null;
        ManifestFile deleteManifest = null;

        // Write the completed data files into a newly created data manifest file.
        if (result.dataFiles() != null && result.dataFiles().length > 0) {
            dataManifest = writeDataFiles(outputFileSupplier.get(), spec, Lists.newArrayList(result.dataFiles()));
        }

        // Write the completed delete files into a newly created delete manifest file.
        if (result.deleteFiles() != null && result.deleteFiles().length > 0) {
            OutputFile deleteManifestFile = outputFileSupplier.get();

            ManifestWriter<DeleteFile> deleteManifestWriter = ManifestFiles.writeDeleteManifest(FORMAT_V2, spec,
                    deleteManifestFile, DUMMY_SNAPSHOT_ID);
            try (ManifestWriter<DeleteFile> writer = deleteManifestWriter) {
                for (DeleteFile deleteFile : result.deleteFiles()) {
                    writer.add(deleteFile);
                }
            }

            deleteManifest = deleteManifestWriter.toManifestFile();
        }

        return new DeltaManifests(dataManifest, deleteManifest, result.referencedDataFiles());
    }
Copy from iceberg-flink:iceberg-flink-1.13:0.13.2
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkManifestUtil.java