class WindowedKStreamImpl extends AbstractStream<K> implements WindowedKStream<K, V> {

    @SuppressWarnings("unchecked")
    @Override
    public <VR> KTable<Windowed<K>, VR> aggregate(final Initializer<VR> initializer,
                                                  final Aggregator<? super K, ? super V, VR> aggregator,
                                                  final Serde<VR> aggValueSerde) {
        Objects.requireNonNull(initializer, "initializer can't be null");
        Objects.requireNonNull(aggregator, "aggregator can't be null");
        final String storeName = builder.newStoreName(AGGREGATE_NAME);
        return (KTable<Windowed<K>, VR>) aggregateBuilder.build(new KStreamWindowAggregate<>(windows, storeName, initializer, aggregator),
                                                                AGGREGATE_NAME,
                                                                windowStoreBuilder(storeName, aggValueSerde),
                                                                false);
    }

      WindowedKStreamImpl(final Windows<W> windows,
                        final InternalStreamsBuilder builder,
                        final Set<String> sourceNodes,
                        final String name,
                        final Serde<K> keySerde,
                        final Serde<V> valSerde,
                        final boolean repartitionRequired);

}