private <VAgg> KTable<K, VAgg> doAggregate(final ProcessorSupplier<K, Change<V>, K, Change<VAgg>> aggregateSupplier,
                                         final NamedInternal named,
                                         final String functionName,
                                         final MaterializedInternal<K, VAgg, KeyValueStore<Bytes, byte[]>> materialized) {

        final String sinkName = named.suffixWithOrElseGet("-sink", builder, KStreamImpl.SINK_NAME);
        final String sourceName = named.suffixWithOrElseGet("-source", builder, KStreamImpl.SOURCE_NAME);
        final String funcName = named.orElseGenerateWithPrefix(builder, functionName);
        final String repartitionTopic = (userProvidedRepartitionTopicName != null ? userProvidedRepartitionTopicName : materialized.storeName())
            + KStreamImpl.REPARTITION_TOPIC_SUFFIX;

        if (repartitionGraphNode == null || userProvidedRepartitionTopicName == null) {
            repartitionGraphNode = createRepartitionNode(sinkName, sourceName, repartitionTopic);
        }


        // the passed in StreamsGraphNode must be the parent of the repartition node
        builder.addGraphNode(this.graphNode, repartitionGraphNode);

        final StatefulProcessorNode statefulProcessorNode = new StatefulProcessorNode<>(
            funcName,
            new ProcessorParameters<>(aggregateSupplier, funcName),
            new KeyValueStoreMaterializer<>(materialized)
        );
        statefulProcessorNode.setOutputVersioned(materialized.storeSupplier() instanceof VersionedBytesStoreSupplier);

        // now the repartition node must be the parent of the StateProcessorNode
        builder.addGraphNode(repartitionGraphNode, statefulProcessorNode);

        // return the KTable representation with the intermediate topic as the sources
        return new KTableImpl<>(funcName,
                                materialized.keySerde(),
                                materialized.valueSerde(),
                                Collections.singleton(sourceName),
                                materialized.queryableStoreName(),
                                aggregateSupplier,
                                statefulProcessorNode,
                                builder);
    }
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