@Test public void shouldAllowNullStoreNameWithWindowedReduce() throws Exception { groupedStream.reduce(MockReducer.STRING_ADDER, TimeWindows.of(10), (String) null); }
@Override public KTable<K, V> reduce(final Reducer<V> reducer, final String queryableStoreName) { determineIsQueryable(queryableStoreName); return reduce(reducer, keyValueStore(keySerde, valSerde, getOrCreateName(queryableStoreName, REDUCE_NAME))); }
KGroupedStreamImpl extends AbstractStream<K> implements KGroupedStream<K, V> { @Override public KTable<K, V> reduce(final Reducer<V> reducer, final String queryableStoreName) { determineIsQueryable(queryableStoreName); return reduce(reducer, keyValueStore(keySerde, valSerde, getOrCreateName(queryableStoreName, REDUCE_NAME))); } }
KGroupedStreamImpl extends AbstractStream<K> implements KGroupedStream<K, V> { @Override public KTable<K, V> reduce(final Reducer<V> reducer, final String queryableStoreName) { determineIsQueryable(queryableStoreName); return reduce(reducer, keyValueStore(keySerde, valSerde, getOrCreateName(queryableStoreName, REDUCE_NAME))); } KGroupedStreamImpl(final KStreamBuilder topology,
                       final String name,
                       final Set<String> sourceNodes,
                       final Serde<K> keySerde,
                       final Serde<V> valSerde,
                       final boolean repartitionRequired); }
KGroupedStreamImpl extends AbstractStream<K> implements KGroupedStream<K, V> { @Override public KTable<K, V> reduce(final Reducer<V> reducer, final String queryableStoreName) { determineIsQueryable(queryableStoreName); return reduce(reducer, keyValueStore(keySerde, valSerde, getOrCreateName(queryableStoreName, REDUCE_NAME))); } KGroupedStreamImpl(final KStreamBuilder topology,
                       final String name,
                       final Set<String> sourceNodes,
                       final Serde<K> keySerde,
                       final Serde<V> valSerde,
                       final boolean repartitionRequired); @Override KTable<K, V> reduce(final Reducer<V> reducer,
                               final String queryableStoreName); @Override KTable<K, V> reduce(final Reducer<V> reducer); @Override KTable<K, V> reduce(final Reducer<V> reducer,
                               final StateStoreSupplier<KeyValueStore> storeSupplier); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                                            final Windows<W> windows,
                                                            final String queryableStoreName); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                                            final Windows<W> windows); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                                            final Windows<W> windows,
                                                            final StateStoreSupplier<WindowStore> storeSupplier); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> aggregator,
                                      final Serde<T> aggValueSerde,
                                      final String queryableStoreName); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> aggregator,
                                      final Serde<T> aggValueSerde); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> aggregator,
                                      final StateStoreSupplier<KeyValueStore> storeSupplier); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                                  final Aggregator<? super K, ? super V, T> aggregator,
                                                                  final Windows<W> windows,
                                                                  final Serde<T> aggValueSerde,
                                                                  final String queryableStoreName); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                                  final Aggregator<? super K, ? super V, T> aggregator,
                                                                  final Windows<W> windows,
                                                                  final Serde<T> aggValueSerde); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                                  final Aggregator<? super K, ? super V, T> aggregator,
                                                                  final Windows<W> windows,
                                                                  final StateStoreSupplier<WindowStore> storeSupplier); @Override KTable<K, Long> count(final String queryableStoreName); @Override KTable<K, Long> count(); @Override KTable<K, Long> count(final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<Windowed<K>, Long> count(final Windows<W> windows,
                                                              final String queryableStoreName); @Override KTable<Windowed<K>, Long> count(final Windows<W> windows); @Override KTable<Windowed<K>, Long> count(final Windows<W> windows,
                                                              final StateStoreSupplier<WindowStore> storeSupplier); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                final Aggregator<? super K, ? super V, T> aggregator,
                                                final Merger<? super K, T> sessionMerger,
                                                final SessionWindows sessionWindows,
                                                final Serde<T> aggValueSerde,
                                                final String queryableStoreName); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                final Aggregator<? super K, ? super V, T> aggregator,
                                                final Merger<? super K, T> sessionMerger,
                                                final SessionWindows sessionWindows,
                                                final Serde<T> aggValueSerde); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                final Aggregator<? super K, ? super V, T> aggregator,
                                                final Merger<? super K, T> sessionMerger,
                                                final SessionWindows sessionWindows,
                                                final Serde<T> aggValueSerde,
                                                final StateStoreSupplier<SessionStore> storeSupplier); @SuppressWarnings("unchecked") KTable<Windowed<K>, Long> count(final SessionWindows sessionWindows, final String queryableStoreName); KTable<Windowed<K>, Long> count(final SessionWindows sessionWindows); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, Long> count(final SessionWindows sessionWindows,
                                           final StateStoreSupplier<SessionStore> storeSupplier); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                         final SessionWindows sessionWindows,
                                         final String queryableStoreName); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                         final SessionWindows sessionWindows); @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                         final SessionWindows sessionWindows,
                                         final StateStoreSupplier<SessionStore> storeSupplier); }
KGroupedStreamImpl extends AbstractStream<K> implements KGroupedStream<K, V> { @Override public KTable<K, V> reduce(final Reducer<V> reducer, final String queryableStoreName) { determineIsQueryable(queryableStoreName); return reduce(reducer, keyValueStore(keySerde, valSerde, getOrCreateName(queryableStoreName, REDUCE_NAME))); } KGroupedStreamImpl(final KStreamBuilder topology,
                       final String name,
                       final Set<String> sourceNodes,
                       final Serde<K> keySerde,
                       final Serde<V> valSerde,
                       final boolean repartitionRequired); @Override KTable<K, V> reduce(final Reducer<V> reducer,
                               final String queryableStoreName); @Override KTable<K, V> reduce(final Reducer<V> reducer); @Override KTable<K, V> reduce(final Reducer<V> reducer,
                               final StateStoreSupplier<KeyValueStore> storeSupplier); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                                            final Windows<W> windows,
                                                            final String queryableStoreName); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                                            final Windows<W> windows); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                                            final Windows<W> windows,
                                                            final StateStoreSupplier<WindowStore> storeSupplier); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> aggregator,
                                      final Serde<T> aggValueSerde,
                                      final String queryableStoreName); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> aggregator,
                                      final Serde<T> aggValueSerde); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> aggregator,
                                      final StateStoreSupplier<KeyValueStore> storeSupplier); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                                  final Aggregator<? super K, ? super V, T> aggregator,
                                                                  final Windows<W> windows,
                                                                  final Serde<T> aggValueSerde,
                                                                  final String queryableStoreName); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                                  final Aggregator<? super K, ? super V, T> aggregator,
                                                                  final Windows<W> windows,
                                                                  final Serde<T> aggValueSerde); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                                  final Aggregator<? super K, ? super V, T> aggregator,
                                                                  final Windows<W> windows,
                                                                  final StateStoreSupplier<WindowStore> storeSupplier); @Override KTable<K, Long> count(final String queryableStoreName); @Override KTable<K, Long> count(); @Override KTable<K, Long> count(final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<Windowed<K>, Long> count(final Windows<W> windows,
                                                              final String queryableStoreName); @Override KTable<Windowed<K>, Long> count(final Windows<W> windows); @Override KTable<Windowed<K>, Long> count(final Windows<W> windows,
                                                              final StateStoreSupplier<WindowStore> storeSupplier); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                final Aggregator<? super K, ? super V, T> aggregator,
                                                final Merger<? super K, T> sessionMerger,
                                                final SessionWindows sessionWindows,
                                                final Serde<T> aggValueSerde,
                                                final String queryableStoreName); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                final Aggregator<? super K, ? super V, T> aggregator,
                                                final Merger<? super K, T> sessionMerger,
                                                final SessionWindows sessionWindows,
                                                final Serde<T> aggValueSerde); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,
                                                final Aggregator<? super K, ? super V, T> aggregator,
                                                final Merger<? super K, T> sessionMerger,
                                                final SessionWindows sessionWindows,
                                                final Serde<T> aggValueSerde,
                                                final StateStoreSupplier<SessionStore> storeSupplier); @SuppressWarnings("unchecked") KTable<Windowed<K>, Long> count(final SessionWindows sessionWindows, final String queryableStoreName); KTable<Windowed<K>, Long> count(final SessionWindows sessionWindows); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, Long> count(final SessionWindows sessionWindows,
                                           final StateStoreSupplier<SessionStore> storeSupplier); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                         final SessionWindows sessionWindows,
                                         final String queryableStoreName); @SuppressWarnings("unchecked") @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                         final SessionWindows sessionWindows); @Override KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                         final SessionWindows sessionWindows,
                                         final StateStoreSupplier<SessionStore> storeSupplier);  }