@Override
    public byte[] serialize(final String topic,
                            final ValueAndTimestamp<V> data) {
        if (data == null) {
            return null;
        }
        return serialize(topic, data.value(), data.timestamp());
    }
@Test
    public void shouldSerializeNonNullDataUsingTheInternalSerializer() {
        final String value = "some-string";

        final ValueAndTimestamp<String> valueAndTimestamp = ValueAndTimestamp.make(value, TIMESTAMP);

        final byte[] serialized =
                STRING_SERDE.serializer().serialize(TOPIC, valueAndTimestamp);

        assertThat(serialized, is(notNullValue()));

        final ValueAndTimestamp<String> deserialized =
                STRING_SERDE.deserializer().deserialize(TOPIC, serialized);

        assertThat(deserialized, is(valueAndTimestamp));
    }