private void deleteCommittedManifests(
            List<ManifestFile> manifests, String newFlinkJobId, long checkpointId) {
        for (ManifestFile manifest : manifests) {
            try {
                table.io().deleteFile(manifest.path());
            } catch (Exception e) {
                // The flink manifests cleaning failure shouldn't abort the completed checkpoint.
                String details =
                        MoreObjects.toStringHelper(this)
                                .add("flinkJobId", newFlinkJobId)
                                .add("checkpointId", checkpointId)
                                .add("manifestPath", manifest.path())
                                .toString();
                LOG.warn(
                        "The iceberg transaction has been committed, but we failed to clean the temporary flink manifests: {}",
                        details,
                        e);
            }
        }
    }
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