@SuppressWarnings("unchecked") @Test public void testGroupedCountOccurences() throws IOException { final KStreamBuilder builder = new KStreamBuilder(); final String input = "count-test-input"; final MockProcessorSupplier processorSupplier = new MockProcessorSupplier<>(); builder.table(Serdes.String(), Serdes.String(), input) .groupBy(new KeyValueMapper<String, String, KeyValue<String, String>>() { @Override public KeyValue<String, String> apply(final String key, final String value) { return new KeyValue<>(value, value); } }, Serdes.String(), Serdes.String()) .count("count") .toStream() .process(processorSupplier); final KStreamTestDriver driver = new KStreamTestDriver(builder, stateDir); driver.process(input, "A", "green"); driver.process(input, "B", "green"); driver.process(input, "A", "blue"); driver.process(input, "C", "yellow"); driver.process(input, "D", "green"); final List<String> expected = Arrays.asList("green:1", "green:2", "blue:1", "green:1", "yellow:1", "green:2"); final List<String> actual = processorSupplier.processed; assertEquals(expected, actual); }
@Override public KTable<K, Long> count(String name) { return this.aggregate( new Initializer<Long>() { @Override public Long apply() { return 0L; } }, new Aggregator<K, V, Long>() { @Override public Long apply(K aggKey, V value, Long aggregate) { return aggregate + 1L; } }, new Aggregator<K, V, Long>() { @Override public Long apply(K aggKey, V value, Long aggregate) { return aggregate - 1L; } }, Serdes.Long(), name); }
KGroupedTableImpl extends AbstractStream<K> implements KGroupedTable<K, V> { @Override public KTable<K, Long> count(String name) { return this.aggregate( new Initializer<Long>() { @Override public Long apply() { return 0L; } }, new Aggregator<K, V, Long>() { @Override public Long apply(K aggKey, V value, Long aggregate) { return aggregate + 1L; } }, new Aggregator<K, V, Long>() { @Override public Long apply(K aggKey, V value, Long aggregate) { return aggregate - 1L; } }, Serdes.Long(), name); } }
KGroupedTableImpl extends AbstractStream<K> implements KGroupedTable<K, V> { @Override public KTable<K, Long> count(String name) { return this.aggregate( new Initializer<Long>() { @Override public Long apply() { return 0L; } }, new Aggregator<K, V, Long>() { @Override public Long apply(K aggKey, V value, Long aggregate) { return aggregate + 1L; } }, new Aggregator<K, V, Long>() { @Override public Long apply(K aggKey, V value, Long aggregate) { return aggregate - 1L; } }, Serdes.Long(), name); } KGroupedTableImpl(KStreamBuilder topology,
                             String name,
                             String sourceName,
                             Serde<K> keySerde,
                             Serde<V> valSerde); }
KGroupedTableImpl extends AbstractStream<K> implements KGroupedTable<K, V> { @Override public KTable<K, Long> count(String name) { return this.aggregate( new Initializer<Long>() { @Override public Long apply() { return 0L; } }, new Aggregator<K, V, Long>() { @Override public Long apply(K aggKey, V value, Long aggregate) { return aggregate + 1L; } }, new Aggregator<K, V, Long>() { @Override public Long apply(K aggKey, V value, Long aggregate) { return aggregate - 1L; } }, Serdes.Long(), name); } KGroupedTableImpl(KStreamBuilder topology,
                             String name,
                             String sourceName,
                             Serde<K> keySerde,
                             Serde<V> valSerde); @Override KTable<K, T> aggregate(Initializer<T> initializer,
                                      Aggregator<K, V, T> adder,
                                      Aggregator<K, V, T> subtractor,
                                      Serde<T> aggValueSerde,
                                      String name); @Override KTable<K, T> aggregate(Initializer<T> initializer,
                            Aggregator<K, V, T> adder,
                            Aggregator<K, V, T> substractor,
                            String name); @Override KTable<K, V> reduce(Reducer<V> adder,
                               Reducer<V> subtractor,
                               String name); @Override KTable<K, Long> count(String name); }
KGroupedTableImpl extends AbstractStream<K> implements KGroupedTable<K, V> { @Override public KTable<K, Long> count(String name) { return this.aggregate( new Initializer<Long>() { @Override public Long apply() { return 0L; } }, new Aggregator<K, V, Long>() { @Override public Long apply(K aggKey, V value, Long aggregate) { return aggregate + 1L; } }, new Aggregator<K, V, Long>() { @Override public Long apply(K aggKey, V value, Long aggregate) { return aggregate - 1L; } }, Serdes.Long(), name); } KGroupedTableImpl(KStreamBuilder topology,
                             String name,
                             String sourceName,
                             Serde<K> keySerde,
                             Serde<V> valSerde); @Override KTable<K, T> aggregate(Initializer<T> initializer,
                                      Aggregator<K, V, T> adder,
                                      Aggregator<K, V, T> subtractor,
                                      Serde<T> aggValueSerde,
                                      String name); @Override KTable<K, T> aggregate(Initializer<T> initializer,
                            Aggregator<K, V, T> adder,
                            Aggregator<K, V, T> substractor,
                            String name); @Override KTable<K, V> reduce(Reducer<V> adder,
                               Reducer<V> subtractor,
                               String name); @Override KTable<K, Long> count(String name);  }