@Test(expected = InvalidTopicException.class) public void shouldNotAllowInvalidStoreNameOnAggregate() throws Exception { groupedTable.aggregate(MockInitializer.STRING_INIT, MockAggregator.TOSTRING_ADDER, MockAggregator.TOSTRING_REMOVER, INVALID_STORE_NAME); }
@Override public <T> 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) { determineIsQueryable(queryableStoreName); return aggregate(initializer, adder, subtractor, keyValueStore(keySerde, aggValueSerde, getOrCreateName(queryableStoreName, AGGREGATE_NAME))); }
KGroupedTableImpl extends AbstractStream<K> implements KGroupedTable<K, V> { @Override public <T> 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) { determineIsQueryable(queryableStoreName); return aggregate(initializer, adder, subtractor, keyValueStore(keySerde, aggValueSerde, getOrCreateName(queryableStoreName, AGGREGATE_NAME))); } }
KGroupedTableImpl extends AbstractStream<K> implements KGroupedTable<K, V> { @Override public <T> 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) { determineIsQueryable(queryableStoreName); return aggregate(initializer, adder, subtractor, keyValueStore(keySerde, aggValueSerde, getOrCreateName(queryableStoreName, AGGREGATE_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 <T> 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) { determineIsQueryable(queryableStoreName); return aggregate(initializer, adder, subtractor, keyValueStore(keySerde, aggValueSerde, getOrCreateName(queryableStoreName, AGGREGATE_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 <T> 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) { determineIsQueryable(queryableStoreName); return aggregate(initializer, adder, subtractor, keyValueStore(keySerde, aggValueSerde, getOrCreateName(queryableStoreName, AGGREGATE_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);  }