@Override
  public KV<K, V> fromBytes(byte[] bytes) {
    if (bytes != null) {
      ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
      int keyLength = byteBuffer.getInt();
      byte[] keyBytes = new byte[keyLength];
      byteBuffer.get(keyBytes);
      int valueLength = byteBuffer.getInt();
      byte[] valueBytes = new byte[valueLength];
      byteBuffer.get(valueBytes);
      K key = keySerde.fromBytes(keyBytes);
      V value = valueSerde.fromBytes(valueBytes);
      return KV.of(key, value);
    } 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