@Override
    public KTable<K, V> reduce(final Reducer<V> reducer) {
        return reduce(reducer, Materialized.with(keySerde, valueSerde));
    }
@Test
    public void shouldLogAndMeasureSkipsInReduce() {
        groupedStream.reduce(
            MockReducer.STRING_ADDER,
            Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("reduce")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.String())
        );

        try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(KStreamReduce.class);
             final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {

            processData(driver);

            assertThat(
                appender.getMessages(),
                hasItem("Skipping record due to null key or value. topic=[topic] partition=[0] "
                    + "offset=[6]")
            );
        }
    }