TimestampedValueSerde implements Serde<TimestampedValue<V>> { @Override public byte[] toBytes(TimestampedValue<V> tv) { byte[] vBytes = vSerde.toBytes(tv.getValue()); int vBytesLength = vBytes != null ? vBytes.length : 0; ByteBuffer bb = ByteBuffer.allocate(vBytesLength + TIMESTAMP_BYTES); if (vBytes != null) { bb.put(vBytes); } bb.putLong(tv.getTimestamp()); return bb.array(); } TimestampedValueSerde(Serde<V> vSerde); @Override TimestampedValue<V> fromBytes(byte[] bytes); @Override byte[] toBytes(TimestampedValue<V> tv);  }
@Test public void testEmptyValueSerialization() { byte[] expectedBytes = new byte[8]; ByteBuffer.wrap(expectedBytes).putLong(1234L); TimestampedValueSerde<Integer> timestampedValueSerde = new TimestampedValueSerde<>(new IntegerSerde()); TimestampedValue<Integer> timestampedValue = new TimestampedValue<>(null, 1234L); assertTrue(Arrays.equals(expectedBytes, timestampedValueSerde.toBytes(timestampedValue))); }