@Override
    public byte[] serialize(final String topic, final TimestampedKeyAndJoinSide<K> data) {
        final byte boolByte = (byte) (data.isLeftSide() ? 1 : 0);
        final byte[] keyBytes = keySerializer.serialize(topic, data.getKey());
        final byte[] timestampBytes = timestampSerializer.serialize(topic, data.getTimestamp());

        return ByteBuffer
            .allocate(timestampBytes.length + 1 + keyBytes.length)
            .put(timestampBytes)
            .put(boolByte)
            .put(keyBytes)
            .array();
    }
@Test
    public void shouldThrowIfSerializeNullData() {
        assertThrows(NullPointerException.class,
            () -> STRING_SERDE.serializer().serialize(TOPIC, TimestampedKeyAndJoinSide.makeLeft(null, 0)));
    }