@Override
    public <VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                                        final Aggregator<? super K, ? super V, VR> aggregator,
                                        final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
        return aggregate(initializer, aggregator, NamedInternal.empty(), materialized);
    }
@Test
    public void shouldAggregateWithDefaultSerdes() {
        final MockApiProcessorSupplier<String, String, Void, Void> supplier = new MockApiProcessorSupplier<>();
        groupedStream
            .aggregate(MockInitializer.STRING_INIT, MockAggregator.TOSTRING_ADDER)
            .toStream()
            .process(supplier);

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

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