@Override
    public byte[] serialize(DeltaManifests deltaManifests) throws IOException {
        Preconditions.checkNotNull(
                deltaManifests, "DeltaManifests to be serialized should not be null");

        ByteArrayOutputStream binaryOut = new ByteArrayOutputStream();
        DataOutputStream out = new DataOutputStream(binaryOut);

        byte[] dataManifestBinary = EMPTY_BINARY;
        if (deltaManifests.dataManifest() != null) {
            dataManifestBinary = ManifestFiles.encode(deltaManifests.dataManifest());
        }

        out.writeInt(dataManifestBinary.length);
        out.write(dataManifestBinary);

        byte[] deleteManifestBinary = EMPTY_BINARY;
        if (deltaManifests.deleteManifest() != null) {
            deleteManifestBinary = ManifestFiles.encode(deltaManifests.deleteManifest());
        }

        out.writeInt(deleteManifestBinary.length);
        out.write(deleteManifestBinary);

        CharSequence[] referencedDataFiles = deltaManifests.referencedDataFiles();
        out.writeInt(referencedDataFiles.length);
        for (CharSequence referencedDataFile : referencedDataFiles) {
            out.writeUTF(referencedDataFile.toString());
        }

        return binaryOut.toByteArray();
    }
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/DeltaManifestsSerializer.java