@Override
    public KTable<K, Long> count(final Named named, final Materialized<K, Long, KeyValueStore<Bytes, byte[]>> materialized) {
        final MaterializedInternal<K, Long, KeyValueStore<Bytes, byte[]>> materializedInternal =
            new MaterializedInternal<>(materialized, builder, AGGREGATE_NAME);

        if (materializedInternal.keySerde() == null) {
            materializedInternal.withKeySerde(keySerde);
        }
        if (materializedInternal.valueSerde() == null) {
            materializedInternal.withValueSerde(Serdes.Long());
        }

        final ProcessorSupplier<K, Change<V>, K, Change<Long>> aggregateSupplier = new KTableAggregate<>(
            materializedInternal.storeName(),
            countInitializer,
            countAdder,
            countSubtractor);

        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