TimestampedValueSerde implements Serde<TimestampedValue<V>> { @Override public TimestampedValue<V> fromBytes(byte[] bytes) { ByteBuffer bb = ByteBuffer.wrap(bytes); byte[] vBytes = new byte[bytes.length - TIMESTAMP_BYTES]; bb.get(vBytes, 0, vBytes.length); V v = vSerde.fromBytes(vBytes); long ts = bb.getLong(); return new TimestampedValue<>(v, ts); } TimestampedValueSerde(Serde<V> vSerde); @Override TimestampedValue<V> fromBytes(byte[] bytes); @Override byte[] toBytes(TimestampedValue<V> tv);  }
@Test public void testEmptyValueDeserialization() { byte[] bytesWithNoValue = new byte[8]; ByteBuffer.wrap(bytesWithNoValue).putLong(1234L); TimestampedValueSerde<byte[]> timestampedValueSerde = new TimestampedValueSerde<>(new ByteSerde()); TimestampedValue<byte[]> timestampedValue = timestampedValueSerde.fromBytes(bytesWithNoValue); assertEquals(1234L, timestampedValue.getTimestamp()); assertEquals(0, timestampedValue.getValue().length); }