@Override
    public <VAgg> KTable<K, VAgg> aggregate(final Initializer<VAgg> initializer,
                                        final Aggregator<? super K, ? super V, VAgg> adder,
                                        final Aggregator<? super K, ? super V, VAgg> subtractor,
                                        final Materialized<K, VAgg, KeyValueStore<Bytes, byte[]>> materialized) {
        return aggregate(initializer, adder, subtractor, NamedInternal.empty(), materialized);
    }
@Test
    public void shouldAggregateAndMaterializeResults() {
        builder
            .table(
                topic,
                Consumed.with(Serdes.String(), Serdes.String()))
            .groupBy(
                MockMapper.selectValueKeyValueMapper(),
                Grouped.with(Serdes.String(), Serdes.String()))
            .aggregate(
                MockInitializer.STRING_INIT,
                MockAggregator.TOSTRING_ADDER,
                MockAggregator.TOSTRING_REMOVER,
                Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("aggregate")
                    .withValueSerde(Serdes.String())
                    .withKeySerde(Serdes.String()));

        try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
            processData(topic, driver);
            {
                {
                    final KeyValueStore<String, String> aggregate = driver.getKeyValueStore("aggregate");
                    assertThat(aggregate.get("1"), equalTo("0+1+1+1"));
                    assertThat(aggregate.get("2"), equalTo("0+2+2"));
                }
                {
                    final KeyValueStore<String, ValueAndTimestamp<String>> aggregate = driver.getTimestampedKeyValueStore("aggregate");
                    assertThat(aggregate.get("1"), equalTo(ValueAndTimestamp.make("0+1+1+1", 50L)));
                    assertThat(aggregate.get("2"), equalTo(ValueAndTimestamp.make("0+2+2", 60L)));
                }
            }
        }
    }