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

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            processData(driver);

            {
                final KeyValueStore<String, String> reduced = driver.getKeyValueStore("reduce");

                assertThat(reduced.get("1"), equalTo("A+C+D"));
                assertThat(reduced.get("2"), equalTo("B"));
                assertThat(reduced.get("3"), equalTo("E+F"));
            }
            {
                final KeyValueStore<String, ValueAndTimestamp<String>> reduced = driver.getTimestampedKeyValueStore("reduce");

                assertThat(reduced.get("1"), equalTo(ValueAndTimestamp.make("A+C+D", 10L)));
                assertThat(reduced.get("2"), equalTo(ValueAndTimestamp.make("B", 1L)));
                assertThat(reduced.get("3"), equalTo(ValueAndTimestamp.make("E+F", 9L)));
            }
        }
    }