@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 Named named,
                                        final Materialized<K, VAgg, KeyValueStore<Bytes, byte[]>> materialized) {
        Objects.requireNonNull(initializer, "initializer can't be null");
        Objects.requireNonNull(adder, "adder can't be null");
        Objects.requireNonNull(subtractor, "subtractor can't be null");
        Objects.requireNonNull(named, "named can't be null");
        Objects.requireNonNull(materialized, "materialized can't be null");

        final MaterializedInternal<K, VAgg, KeyValueStore<Bytes, byte[]>> materializedInternal =
            new MaterializedInternal<>(materialized, builder, AGGREGATE_NAME);

        if (materializedInternal.keySerde() == null) {
            materializedInternal.withKeySerde(keySerde);
        }
        final ProcessorSupplier<K, Change<V>, K, Change<VAgg>> aggregateSupplier = new KTableAggregate<>(
            materializedInternal.storeName(),
            initializer,
            adder,
            subtractor);
        return doAggregate(aggregateSupplier, new NamedInternal(named), AGGREGATE_NAME, materializedInternal);
    }
The implementation class of {@link KGroupedTable}.

@param <K> the key type
@param <V> the value type
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java