@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 shouldReduce() {
        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()),
                Materialized.<String, Double, KeyValueStore<Bytes, byte[]>>as("store")
                    .withKeySerde(Serdes.String())
                    .withValueSerde(Serdes.Double()))
            .groupBy(intProjection)
            .reduce(
                MockReducer.INTEGER_ADDER,
                MockReducer.INTEGER_SUBTRACTOR,
                Materialized.as("reduced"));

        final MockApiProcessorSupplier<String, Integer, Void, Void> supplier = getReducedResults(reduced);
        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            assertReduced(supplier.theCapturedProcessor().lastValueAndTimestampPerKey(), topic, driver);
            assertEquals(reduced.queryableStoreName(), "reduced");
        }
    }