Hello everybody. I kinda stuck with a problem: In our original Kafka project, we have implemented our functions for serialization and deserialization. In the Flink project (in Java), I will need to follow deserialization approach. Everything is fine except binary chunk:

dataFileChunk.data

. In Python deserialization, they used

"unpackb"

and I used the same component in Java for deserialization but I get different results vs Python deserialization: This is my deserialization class in Java:

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.json.JsonMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;

import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
import org.msgpack.core.MessageBufferPacker;
import org.msgpack.core.MessageFormat;
import org.msgpack.core.MessagePack;

import org.msgpack.core.MessageUnpacker;
import org.msgpack.value.ImmutableValue;
import org.msgpack.value.Value;
import org.msgpack.value.ValueType;


public class DataFileChunkDeserializer extends AbstractDeserializationSchema<DataFileChunk> {

    private static final long serialVersionUID = 1L;

    private transient ObjectMapper objectMapper;

    @Override
    public void open(InitializationContext context) {
        objectMapper = JsonMapper.builder().build().registerModule(new JavaTimeModule());
    }

    @Override
    public DataFileChunk deserialize(byte[] message) throws IOException {
//        List<DataFileChunk> dataFileChunks = new ArrayList<>();
        DataFileChunk dataFileChunk = new DataFileChunk();
        MessageDigest md = null;
        MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(message);
        if (unpacker.hasNext()) {
            try {
                md = MessageDigest.getInstance("SHA-512");
            } catch (NoSuchAlgorithmException e) {
                e.printStackTrace();
            }
            ImmutableValue value = unpacker.unpackValue();

            List<Value> list = value.asArrayValue().list();

            dataFileChunk.filename = "" + list.get(0);
            dataFileChunk.chunk_hash = "" + list.get(2);
            dataFileChunk.chunk_i = Long.parseLong(String.valueOf(list.get(4)));
            dataFileChunk.n_total_chunks = Long.parseLong(String.valueOf(list.get(5)));
            dataFileChunk.subdir_str = "" + list.get(6);
            dataFileChunk.filename_append = "" + list.get(7);
            ValueType vt = list.get(8).getValueType();

            dataFileChunk.data = list.get(8).asBinaryValue();
            dataFileChunk.dataByteArray = dataFileChunk.data.asByteArray();
            assert md != null;

            try {
                byte[] sh1 = MessageDigest.getInstance("SHA-512").digest(dataFileChunk.dataByteArray);
                byte[] sh2 = dataFileChunk.chunk_hash.getBytes(StandardCharsets.UTF_8);
                System.out.println();
            } catch (NoSuchAlgorithmException e) {
                e.printStackTrace();
            }

        }

        return dataFileChunk;

    }
}

and this is part of my pom.xml:

<dependency>
    <groupId>org.msgpack</groupId>
    <artifactId>msgpack-core</artifactId>
    <version>0.8.18</version>
</dependency>

DataFileChunk class:

import org.msgpack.value.BinaryValue;
import org.msgpack.value.RawValue;

public class DataFileChunk {
    public long chunk_i;
    public String filename;
    public String chunk_hash;
    public String chunk_offset_write;
    public long n_total_chunks;
    public String subdir_str;
    public String filename_append;
    public RawValue data;
    public byte[] dataByteArray;

    public DataFileChunk() {
    }

    public DataFileChunk(long chunk_i, String filename) {
        this.chunk_i = chunk_i;
        this.filename = filename;
    }
    public DataFileChunk(long chunk_i, String filename, String chunk_hash, String chunk_offset_write,
                         long n_total_chunks, String subdir_str, String filename_append, BinaryValue data) {
        this.chunk_i = chunk_i;
        this.filename = filename;
        this.chunk_hash = chunk_hash;
        this.chunk_offset_write = chunk_offset_write;
        this.n_total_chunks = n_total_chunks;
        this.subdir_str = subdir_str;
        this.filename_append = filename_append;
        this.data = data;
    }

    @Override
    public String toString() {
        return "Event{" + "chunk_i=" + chunk_i + ", n_total_chunks=" + n_total_chunks +
                ", filename='" + filename + '\'' +
                ", chunk_hash='" + chunk_hash + '\'' +
                ", chunk_offset_write='" + chunk_offset_write + '\'' +
                ", subdir_str='" + subdir_str + '\'' +
                ", filename_append='" + filename_append + '\'' +
                ", data='" + data + '\'' +
                '}';
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        DataFileChunk dataFileChunk = (DataFileChunk) o;
        return chunk_i == dataFileChunk.chunk_i &&
                n_total_chunks == dataFileChunk.n_total_chunks &&
                filename.equals(dataFileChunk.filename) &&
                chunk_hash.equals(dataFileChunk.chunk_hash) &&
                chunk_offset_write.equals(dataFileChunk.chunk_offset_write) &&
                subdir_str.equals(dataFileChunk.subdir_str) &&
                filename_append.equals(dataFileChunk.filename_append) &&
                data.equals(dataFileChunk.data);
    }


}

Does anyone have experience of deserialization with msgpack?