@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 shouldReturnNullWhenTheInternalSerializerReturnsNull() {
        // Testing against regressions with respect to https://github.com/apache/kafka/pull/7679

        final Serializer<String> alwaysNullSerializer = (topic, data) -> null;

        final ValueAndTimestampSerializer<String> serializer =
                new ValueAndTimestampSerializer<>(alwaysNullSerializer);

        final byte[] serialized = serializer.serialize(TOPIC, "non-null-data", TIMESTAMP);

        assertThat(serialized, is(nullValue()));
    }