ds_state_processor_blog_2.java

@Override
public byte[] serialize(KafkaPartitionSplit split) throws IOException {
    try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
            DataOutputStream out = new DataOutputStream(baos)) {
        out.writeUTF(split.getTopic());
        out.writeInt(split.getPartition());
        out.writeLong(split.getStartingOffset());
        out.writeLong(split.getStoppingOffset().orElse(KafkaPartitionSplit.NO_STOPPING_OFFSET));
        out.flush();
        return baos.toByteArray();
    }
}