test_class.declarator: stateDir
test_class.modifier: private
test_class.original_string: private File stateDir;
test_class.type: File
test_class.var_name: stateDir
test_class.file: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImplTest.java
test_class.identifier: KGroupedTableImplTest
test_case.body: @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);
      }
test_case.class_method_signature: KGroupedTableImplTest.testGroupedCountOccurences()
test_case.full_signature: @SuppressWarnings("unchecked") @Test public void testGroupedCountOccurences()
test_case.identifier: testGroupedCountOccurences
test_case.invocations: process toStream count groupBy table String String String String process process process process process asList assertEquals
test_case.modifiers: @SuppressWarnings("unchecked") @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testGroupedCountOccurences()
focal_class.declarator: AGGREGATE_NAME = "KTABLE-AGGREGATE-"
focal_class.modifier: private static final
focal_class.original_string: private static final String AGGREGATE_NAME = "KTABLE-AGGREGATE-";
focal_class.type: String
focal_class.var_name: AGGREGATE_NAME
focal_class.declarator: REDUCE_NAME = "KTABLE-REDUCE-"
focal_class.modifier: private static final
focal_class.original_string: private static final String REDUCE_NAME = "KTABLE-REDUCE-";
focal_class.type: String
focal_class.var_name: REDUCE_NAME
focal_class.declarator: REPARTITION_TOPIC_SUFFIX = "-repartition"
focal_class.modifier: private static final
focal_class.original_string: private static final String REPARTITION_TOPIC_SUFFIX = "-repartition";
focal_class.type: String
focal_class.var_name: REPARTITION_TOPIC_SUFFIX
focal_class.declarator: keySerde
focal_class.modifier: protected final
focal_class.original_string: protected final Serde<K> keySerde;
focal_class.type: Serde<K>
focal_class.var_name: keySerde
focal_class.declarator: valSerde
focal_class.modifier: protected final
focal_class.original_string: protected final Serde<V> valSerde;
focal_class.type: Serde<V>
focal_class.var_name: valSerde
focal_class.file: streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java
focal_class.identifier: KGroupedTableImpl
focal_class.interfaces: implements KGroupedTable<K, V>
focal_class.class_method_signature: KGroupedTableImpl.KGroupedTableImpl(KStreamBuilder topology,
                               String name,
                               String sourceName,
                               Serde<K> keySerde,
                               Serde<V> valSerde)
focal_class.full_signature: public  KGroupedTableImpl(KStreamBuilder topology,
                               String name,
                               String sourceName,
                               Serde<K> keySerde,
                               Serde<V> valSerde)
focal_class.identifier: KGroupedTableImpl
focal_class.modifiers: public
focal_class.parameters: (KStreamBuilder topology,
                               String name,
                               String sourceName,
                               Serde<K> keySerde,
                               Serde<V> valSerde)
focal_class.signature:  KGroupedTableImpl(KStreamBuilder topology,
                               String name,
                               String sourceName,
                               Serde<K> keySerde,
                               Serde<V> valSerde)
focal_class.class_method_signature: KGroupedTableImpl.aggregate(Initializer<T> initializer,
                                        Aggregator<K, V, T> adder,
                                        Aggregator<K, V, T> subtractor,
                                        Serde<T> aggValueSerde,
                                        String name)
focal_class.full_signature: @Override public KTable<K, T> aggregate(Initializer<T> initializer,
                                        Aggregator<K, V, T> adder,
                                        Aggregator<K, V, T> subtractor,
                                        Serde<T> aggValueSerde,
                                        String name)
focal_class.identifier: aggregate
focal_class.modifiers: @Override public
focal_class.parameters: (Initializer<T> initializer,
                                        Aggregator<K, V, T> adder,
                                        Aggregator<K, V, T> subtractor,
                                        Serde<T> aggValueSerde,
                                        String name)
focal_class.return: KTable<K, T>
focal_class.signature: KTable<K, T> aggregate(Initializer<T> initializer,
                                        Aggregator<K, V, T> adder,
                                        Aggregator<K, V, T> subtractor,
                                        Serde<T> aggValueSerde,
                                        String name)
focal_class.class_method_signature: KGroupedTableImpl.aggregate(Initializer<T> initializer,
                              Aggregator<K, V, T> adder,
                              Aggregator<K, V, T> substractor,
                              String name)
focal_class.full_signature: @Override public KTable<K, T> aggregate(Initializer<T> initializer,
                              Aggregator<K, V, T> adder,
                              Aggregator<K, V, T> substractor,
                              String name)
focal_class.identifier: aggregate
focal_class.modifiers: @Override public
focal_class.parameters: (Initializer<T> initializer,
                              Aggregator<K, V, T> adder,
                              Aggregator<K, V, T> substractor,
                              String name)
focal_class.return: KTable<K, T>
focal_class.signature: KTable<K, T> aggregate(Initializer<T> initializer,
                              Aggregator<K, V, T> adder,
                              Aggregator<K, V, T> substractor,
                              String name)
focal_class.class_method_signature: KGroupedTableImpl.doAggregate(ProcessorSupplier<K, Change<V>> aggregateSupplier,
                                           Serde<T> aggValueSerde,
                                           String functionName,
                                           String name)
focal_class.full_signature: private KTable<K, T> doAggregate(ProcessorSupplier<K, Change<V>> aggregateSupplier,
                                           Serde<T> aggValueSerde,
                                           String functionName,
                                           String name)
focal_class.identifier: doAggregate
focal_class.modifiers: private
focal_class.parameters: (ProcessorSupplier<K, Change<V>> aggregateSupplier,
                                           Serde<T> aggValueSerde,
                                           String functionName,
                                           String name)
focal_class.return: KTable<K, T>
focal_class.signature: KTable<K, T> doAggregate(ProcessorSupplier<K, Change<V>> aggregateSupplier,
                                           Serde<T> aggValueSerde,
                                           String functionName,
                                           String name)
focal_class.class_method_signature: KGroupedTableImpl.reduce(Reducer<V> adder,
                                 Reducer<V> subtractor,
                                 String name)
focal_class.full_signature: @Override public KTable<K, V> reduce(Reducer<V> adder,
                                 Reducer<V> subtractor,
                                 String name)
focal_class.identifier: reduce
focal_class.modifiers: @Override public
focal_class.parameters: (Reducer<V> adder,
                                 Reducer<V> subtractor,
                                 String name)
focal_class.return: KTable<K, V>
focal_class.signature: KTable<K, V> reduce(Reducer<V> adder,
                                 Reducer<V> subtractor,
                                 String name)
focal_class.class_method_signature: KGroupedTableImpl.count(String name)
focal_class.full_signature: @Override public KTable<K, Long> count(String name)
focal_class.identifier: count
focal_class.modifiers: @Override public
focal_class.parameters: (String name)
focal_class.return: KTable<K, Long>
focal_class.signature: KTable<K, Long> count(String name)
focal_class.superclass: extends AbstractStream<K>
focal_method.body: @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);
      }
focal_method.class_method_signature: KGroupedTableImpl.count(String name)
focal_method.full_signature: @Override public KTable<K, Long> count(String name)
focal_method.identifier: count
focal_method.invocations: aggregate Long
focal_method.modifiers: @Override public
focal_method.parameters: (String name)
focal_method.return: KTable<K, Long>
focal_method.signature: KTable<K, Long> count(String name)