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

    @SuppressWarnings("unchecked")
    @Override
    public KTable<Windowed<K>, V> reduce(final Reducer<V> reducer) {
        Objects.requireNonNull(reducer, "reducer can't be null");
        final String storeName = builder.newStoreName(REDUCE_NAME);
        return (KTable<Windowed<K>, V>) aggregateBuilder.build(new KStreamWindowReduce<K, V, W>(windows, storeName, reducer),
                                                               REDUCE_NAME,
                                                               windowStoreBuilder(storeName, valSerde),
                                                               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);

}