@Override
    public KTable<K, V> reduce(final Reducer<V> adder,
                               final Reducer<V> subtractor,
                               final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized) {
        return reduce(adder, subtractor, NamedInternal.empty(), materialized);
    }
@Test
    public void shouldReduceAndMaterializeResults() {
        final KeyValueMapper<String, Number, KeyValue<String, Integer>> intProjection =
            (key, value) -> KeyValue.pair(key, value.intValue());

        final KTable<String, Integer> reduced = builder
            .table(
                topic,
                Consumed.with(Serdes.String(), Serdes.Double()))
            .groupBy(intProjection)
            .reduce(
                MockReducer.INTEGER_ADDER,
                MockReducer.INTEGER_SUBTRACTOR,
                Materialized.<String, Integer, KeyValueStore<Bytes, byte[]>>as("reduce")
                    .withKeySerde(Serdes.String())
                    .withValueSerde(Serdes.Integer()));

        final MockApiProcessorSupplier<String, Integer, Void, Void> supplier = getReducedResults(reduced);
        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            assertReduced(supplier.theCapturedProcessor().lastValueAndTimestampPerKey(), topic, driver);
            {
                final KeyValueStore<String, Integer> reduce = driver.getKeyValueStore("reduce");
                assertThat(reduce.get("A"), equalTo(5));
                assertThat(reduce.get("B"), equalTo(6));
            }
            {
                final KeyValueStore<String, ValueAndTimestamp<Integer>> reduce = driver.getTimestampedKeyValueStore("reduce");
                assertThat(reduce.get("A"), equalTo(ValueAndTimestamp.make(5, 50L)));
                assertThat(reduce.get("B"), equalTo(ValueAndTimestamp.make(6, 30L)));
            }
        }
    }