@Test public void shouldReduce() throws Exception { final String topic = "input"; final KeyValueMapper<String, Number, KeyValue<String, Integer>> intProjection = new KeyValueMapper<String, Number, KeyValue<String, Integer>>() { @Override public KeyValue<String, Integer> apply(String key, Number value) { return KeyValue.pair(key, value.intValue()); } }; final KTable<String, Integer> reduced = builder.table(Serdes.String(), Serdes.Double(), topic, "store") .groupBy(intProjection) .reduce(MockReducer.INTEGER_ADDER, MockReducer.INTEGER_SUBTRACTOR, "reduced"); doShouldReduce(reduced, topic); assertEquals(reduced.queryableStoreName(), "reduced"); }
@Override public KTable<K, V> reduce(final Reducer<V> adder, final Reducer<V> subtractor, final String queryableStoreName) { determineIsQueryable(queryableStoreName); return reduce(adder, subtractor, keyValueStore(keySerde, valSerde, getOrCreateName(queryableStoreName, REDUCE_NAME))); }
KGroupedTableImpl extends AbstractStream<K> implements KGroupedTable<K, V> { @Override public KTable<K, V> reduce(final Reducer<V> adder, final Reducer<V> subtractor, final String queryableStoreName) { determineIsQueryable(queryableStoreName); return reduce(adder, subtractor, keyValueStore(keySerde, valSerde, getOrCreateName(queryableStoreName, REDUCE_NAME))); } }
KGroupedTableImpl extends AbstractStream<K> implements KGroupedTable<K, V> { @Override public KTable<K, V> reduce(final Reducer<V> adder, final Reducer<V> subtractor, final String queryableStoreName) { determineIsQueryable(queryableStoreName); return reduce(adder, subtractor, keyValueStore(keySerde, valSerde, getOrCreateName(queryableStoreName, REDUCE_NAME))); } KGroupedTableImpl(final KStreamBuilder topology,
                             final String name,
                             final String sourceName,
                             final Serde<? extends K> keySerde,
                             final Serde<? extends V> valSerde); }
KGroupedTableImpl extends AbstractStream<K> implements KGroupedTable<K, V> { @Override public KTable<K, V> reduce(final Reducer<V> adder, final Reducer<V> subtractor, final String queryableStoreName) { determineIsQueryable(queryableStoreName); return reduce(adder, subtractor, keyValueStore(keySerde, valSerde, getOrCreateName(queryableStoreName, REDUCE_NAME))); } KGroupedTableImpl(final KStreamBuilder topology,
                             final String name,
                             final String sourceName,
                             final Serde<? extends K> keySerde,
                             final Serde<? extends V> valSerde); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> adder,
                                      final Aggregator<? super K, ? super V, T> subtractor,
                                      final Serde<T> aggValueSerde,
                                      final String queryableStoreName); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> adder,
                                      final Aggregator<? super K, ? super V, T> subtractor,
                                      final Serde<T> aggValueSerde); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> adder,
                                      final Aggregator<? super K, ? super V, T> subtractor,
                                      final String queryableStoreName); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> adder,
                                      final Aggregator<? super K, ? super V, T> subtractor); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> adder,
                                      final Aggregator<? super K, ? super V, T> subtractor,
                                      final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> reduce(final Reducer<V> adder,
                               final Reducer<V> subtractor,
                               final String queryableStoreName); @Override KTable<K, V> reduce(final Reducer<V> adder,
                               final Reducer<V> subtractor); @Override KTable<K, V> reduce(final Reducer<V> adder,
                               final Reducer<V> subtractor,
                               final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, Long> count(final String queryableStoreName); @Override KTable<K, Long> count(); @Override KTable<K, Long> count(final StateStoreSupplier<KeyValueStore> storeSupplier); }
KGroupedTableImpl extends AbstractStream<K> implements KGroupedTable<K, V> { @Override public KTable<K, V> reduce(final Reducer<V> adder, final Reducer<V> subtractor, final String queryableStoreName) { determineIsQueryable(queryableStoreName); return reduce(adder, subtractor, keyValueStore(keySerde, valSerde, getOrCreateName(queryableStoreName, REDUCE_NAME))); } KGroupedTableImpl(final KStreamBuilder topology,
                             final String name,
                             final String sourceName,
                             final Serde<? extends K> keySerde,
                             final Serde<? extends V> valSerde); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> adder,
                                      final Aggregator<? super K, ? super V, T> subtractor,
                                      final Serde<T> aggValueSerde,
                                      final String queryableStoreName); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> adder,
                                      final Aggregator<? super K, ? super V, T> subtractor,
                                      final Serde<T> aggValueSerde); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> adder,
                                      final Aggregator<? super K, ? super V, T> subtractor,
                                      final String queryableStoreName); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> adder,
                                      final Aggregator<? super K, ? super V, T> subtractor); @Override KTable<K, T> aggregate(final Initializer<T> initializer,
                                      final Aggregator<? super K, ? super V, T> adder,
                                      final Aggregator<? super K, ? super V, T> subtractor,
                                      final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> reduce(final Reducer<V> adder,
                               final Reducer<V> subtractor,
                               final String queryableStoreName); @Override KTable<K, V> reduce(final Reducer<V> adder,
                               final Reducer<V> subtractor); @Override KTable<K, V> reduce(final Reducer<V> adder,
                               final Reducer<V> subtractor,
                               final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, Long> count(final String queryableStoreName); @Override KTable<K, Long> count(); @Override KTable<K, Long> count(final StateStoreSupplier<KeyValueStore> storeSupplier);  }