@Override
  public byte[] toBytes(KV<K, V> obj) {
    if (obj != null) {
      byte[] keyBytes = keySerde.toBytes(obj.key);
      byte[] valueBytes = valueSerde.toBytes(obj.value);
      byte[] bytes = new byte[8 + keyBytes.length + 8 + valueBytes.length];
      ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
      byteBuffer.putInt(keyBytes.length);
      byteBuffer.put(keyBytes);
      byteBuffer.putInt(valueBytes.length);
      byteBuffer.put(valueBytes);
      return byteBuffer.array();
    } else {
      return null;
    }
  }
A serializer that serves one of the following two purposes:
<ol>
  <li>When used for input/output streams, serves as an indication to the framework that messages are keyed and
  should be deserialized as K-V pairs.</li>
  <li>When used for a key or value of type KV, serializes the key and value pair using the provided serdes.</li>
</ol>

@param <K> type of the key in the message
@param <V> type of the value in the message
samza-api/src/main/java/org/apache/samza/serializers/KVSerde.java