test_class.declarator: stringSerde = Serdes.String()
test_class.modifier: final private
test_class.original_string: final private Serde<String> stringSerde = Serdes.String();
test_class.type: Serde<String>
test_class.var_name: stringSerde
test_class.declarator: driver = null
test_class.modifier: private
test_class.original_string: private KStreamTestDriver driver = null;
test_class.type: KStreamTestDriver
test_class.var_name: driver
test_class.declarator: stateDir = null
test_class.modifier: private
test_class.original_string: private File stateDir = null;
test_class.type: File
test_class.var_name: stateDir
test_class.file: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
test_class.identifier: KTableImplTest
test_case.body: @Test
      public void testRepartition() throws IOException {
          String topic1 = "topic1";
  
          final KStreamBuilder builder = new KStreamBuilder();
  
          KTableImpl<String, String, String> table1 =
                  (KTableImpl<String, String, String>) builder.table(stringSerde, stringSerde, topic1);
  
          KTableImpl<String, String, String> table1Aggregated = (KTableImpl<String, String, String>) table1
                  .groupBy(MockKeyValueMapper.<String, String>NoOpKeyValueMapper())
                  .aggregate(MockInitializer.STRING_INIT, MockAggregator.STRING_ADDER, MockAggregator.STRING_REMOVER, "mock-result1");
  
  
          KTableImpl<String, String, String> table1Reduced = (KTableImpl<String, String, String>) table1
                  .groupBy(MockKeyValueMapper.<String, String>NoOpKeyValueMapper())
                  .reduce(MockReducer.STRING_ADDER, MockReducer.STRING_REMOVER, "mock-result2");
  
          driver = new KStreamTestDriver(builder, stateDir, stringSerde, stringSerde);
          driver.setTime(0L);
  
          // three state store should be created, one for source, one for aggregate and one for reduce
          assertEquals(3, driver.allStateStores().size());
  
          // contains the corresponding repartition source / sink nodes
          assertTrue(driver.allProcessorNames().contains("KSTREAM-SINK-0000000003"));
          assertTrue(driver.allProcessorNames().contains("KSTREAM-SOURCE-0000000004"));
          assertTrue(driver.allProcessorNames().contains("KSTREAM-SINK-0000000007"));
          assertTrue(driver.allProcessorNames().contains("KSTREAM-SOURCE-0000000008"));
  
          assertNotNull(((ChangedSerializer) ((SinkNode) driver.processor("KSTREAM-SINK-0000000003")).valueSerializer()).inner());
          assertNotNull(((ChangedDeserializer) ((SourceNode) driver.processor("KSTREAM-SOURCE-0000000004")).valueDeserializer()).inner());
          assertNotNull(((ChangedSerializer) ((SinkNode) driver.processor("KSTREAM-SINK-0000000007")).valueSerializer()).inner());
          assertNotNull(((ChangedDeserializer) ((SourceNode) driver.processor("KSTREAM-SOURCE-0000000008")).valueDeserializer()).inner());
      }
test_case.class_method_signature: KTableImplTest.testRepartition()
test_case.full_signature: @Test public void testRepartition()
test_case.identifier: testRepartition
test_case.invocations: table aggregate groupBy NoOpKeyValueMapper reduce groupBy NoOpKeyValueMapper setTime assertEquals size allStateStores assertTrue contains allProcessorNames assertTrue contains allProcessorNames assertTrue contains allProcessorNames assertTrue contains allProcessorNames assertNotNull inner valueSerializer processor assertNotNull inner valueDeserializer processor assertNotNull inner valueSerializer processor assertNotNull inner valueDeserializer processor
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testRepartition()
focal_class.declarator: FILTER_NAME = "KTABLE-FILTER-"
focal_class.modifier: private static final
focal_class.original_string: private static final String FILTER_NAME = "KTABLE-FILTER-";
focal_class.type: String
focal_class.var_name: FILTER_NAME
focal_class.declarator: FOREACH_NAME = "KTABLE-FOREACH-"
focal_class.modifier: private static final
focal_class.original_string: private static final String FOREACH_NAME = "KTABLE-FOREACH-";
focal_class.type: String
focal_class.var_name: FOREACH_NAME
focal_class.declarator: JOINTHIS_NAME = "KTABLE-JOINTHIS-"
focal_class.modifier: public static final
focal_class.original_string: public static final String JOINTHIS_NAME = "KTABLE-JOINTHIS-";
focal_class.type: String
focal_class.var_name: JOINTHIS_NAME
focal_class.declarator: JOINOTHER_NAME = "KTABLE-JOINOTHER-"
focal_class.modifier: public static final
focal_class.original_string: public static final String JOINOTHER_NAME = "KTABLE-JOINOTHER-";
focal_class.type: String
focal_class.var_name: JOINOTHER_NAME
focal_class.declarator: LEFTTHIS_NAME = "KTABLE-LEFTTHIS-"
focal_class.modifier: public static final
focal_class.original_string: public static final String LEFTTHIS_NAME = "KTABLE-LEFTTHIS-";
focal_class.type: String
focal_class.var_name: LEFTTHIS_NAME
focal_class.declarator: LEFTOTHER_NAME = "KTABLE-LEFTOTHER-"
focal_class.modifier: public static final
focal_class.original_string: public static final String LEFTOTHER_NAME = "KTABLE-LEFTOTHER-";
focal_class.type: String
focal_class.var_name: LEFTOTHER_NAME
focal_class.declarator: MAPVALUES_NAME = "KTABLE-MAPVALUES-"
focal_class.modifier: private static final
focal_class.original_string: private static final String MAPVALUES_NAME = "KTABLE-MAPVALUES-";
focal_class.type: String
focal_class.var_name: MAPVALUES_NAME
focal_class.declarator: MERGE_NAME = "KTABLE-MERGE-"
focal_class.modifier: public static final
focal_class.original_string: public static final String MERGE_NAME = "KTABLE-MERGE-";
focal_class.type: String
focal_class.var_name: MERGE_NAME
focal_class.declarator: OUTERTHIS_NAME = "KTABLE-OUTERTHIS-"
focal_class.modifier: public static final
focal_class.original_string: public static final String OUTERTHIS_NAME = "KTABLE-OUTERTHIS-";
focal_class.type: String
focal_class.var_name: OUTERTHIS_NAME
focal_class.declarator: OUTEROTHER_NAME = "KTABLE-OUTEROTHER-"
focal_class.modifier: public static final
focal_class.original_string: public static final String OUTEROTHER_NAME = "KTABLE-OUTEROTHER-";
focal_class.type: String
focal_class.var_name: OUTEROTHER_NAME
focal_class.declarator: PRINTING_NAME = "KSTREAM-PRINTER-"
focal_class.modifier: private static final
focal_class.original_string: private static final String PRINTING_NAME = "KSTREAM-PRINTER-";
focal_class.type: String
focal_class.var_name: PRINTING_NAME
focal_class.declarator: SELECT_NAME = "KTABLE-SELECT-"
focal_class.modifier: private static final
focal_class.original_string: private static final String SELECT_NAME = "KTABLE-SELECT-";
focal_class.type: String
focal_class.var_name: SELECT_NAME
focal_class.declarator: SOURCE_NAME = "KTABLE-SOURCE-"
focal_class.modifier: public static final
focal_class.original_string: public static final String SOURCE_NAME = "KTABLE-SOURCE-";
focal_class.type: String
focal_class.var_name: SOURCE_NAME
focal_class.declarator: TOSTREAM_NAME = "KTABLE-TOSTREAM-"
focal_class.modifier: private static final
focal_class.original_string: private static final String TOSTREAM_NAME = "KTABLE-TOSTREAM-";
focal_class.type: String
focal_class.var_name: TOSTREAM_NAME
focal_class.declarator: processorSupplier
focal_class.modifier: public final
focal_class.original_string: public final ProcessorSupplier<?, ?> processorSupplier;
focal_class.type: ProcessorSupplier<?, ?>
focal_class.var_name: processorSupplier
focal_class.declarator: keySerde
focal_class.modifier: private final
focal_class.original_string: private final Serde<K> keySerde;
focal_class.type: Serde<K>
focal_class.var_name: keySerde
focal_class.declarator: valSerde
focal_class.modifier: private final
focal_class.original_string: private final Serde<V> valSerde;
focal_class.type: Serde<V>
focal_class.var_name: valSerde
focal_class.declarator: sendOldValues = false
focal_class.modifier: private
focal_class.original_string: private boolean sendOldValues = false;
focal_class.type: boolean
focal_class.var_name: sendOldValues
focal_class.file: streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
focal_class.identifier: KTableImpl
focal_class.interfaces: implements KTable<K, V>
focal_class.class_method_signature: KTableImpl.KTableImpl(KStreamBuilder topology,
                        String name,
                        ProcessorSupplier<?, ?> processorSupplier,
                        Set<String> sourceNodes)
focal_class.full_signature: public  KTableImpl(KStreamBuilder topology,
                        String name,
                        ProcessorSupplier<?, ?> processorSupplier,
                        Set<String> sourceNodes)
focal_class.identifier: KTableImpl
focal_class.modifiers: public
focal_class.parameters: (KStreamBuilder topology,
                        String name,
                        ProcessorSupplier<?, ?> processorSupplier,
                        Set<String> sourceNodes)
focal_class.signature:  KTableImpl(KStreamBuilder topology,
                        String name,
                        ProcessorSupplier<?, ?> processorSupplier,
                        Set<String> sourceNodes)
focal_class.class_method_signature: KTableImpl.KTableImpl(KStreamBuilder topology,
                        String name,
                        ProcessorSupplier<?, ?> processorSupplier,
                        Set<String> sourceNodes,
                        Serde<K> keySerde,
                        Serde<V> valSerde)
focal_class.full_signature: public  KTableImpl(KStreamBuilder topology,
                        String name,
                        ProcessorSupplier<?, ?> processorSupplier,
                        Set<String> sourceNodes,
                        Serde<K> keySerde,
                        Serde<V> valSerde)
focal_class.identifier: KTableImpl
focal_class.modifiers: public
focal_class.parameters: (KStreamBuilder topology,
                        String name,
                        ProcessorSupplier<?, ?> processorSupplier,
                        Set<String> sourceNodes,
                        Serde<K> keySerde,
                        Serde<V> valSerde)
focal_class.signature:  KTableImpl(KStreamBuilder topology,
                        String name,
                        ProcessorSupplier<?, ?> processorSupplier,
                        Set<String> sourceNodes,
                        Serde<K> keySerde,
                        Serde<V> valSerde)
focal_class.class_method_signature: KTableImpl.filter(Predicate<K, V> predicate)
focal_class.full_signature: @Override public KTable<K, V> filter(Predicate<K, V> predicate)
focal_class.identifier: filter
focal_class.modifiers: @Override public
focal_class.parameters: (Predicate<K, V> predicate)
focal_class.return: KTable<K, V>
focal_class.signature: KTable<K, V> filter(Predicate<K, V> predicate)
focal_class.class_method_signature: KTableImpl.filterNot(final Predicate<K, V> predicate)
focal_class.full_signature: @Override public KTable<K, V> filterNot(final Predicate<K, V> predicate)
focal_class.identifier: filterNot
focal_class.modifiers: @Override public
focal_class.parameters: (final Predicate<K, V> predicate)
focal_class.return: KTable<K, V>
focal_class.signature: KTable<K, V> filterNot(final Predicate<K, V> predicate)
focal_class.class_method_signature: KTableImpl.mapValues(ValueMapper<V, V1> mapper)
focal_class.full_signature: @Override public KTable<K, V1> mapValues(ValueMapper<V, V1> mapper)
focal_class.identifier: mapValues
focal_class.modifiers: @Override public
focal_class.parameters: (ValueMapper<V, V1> mapper)
focal_class.return: KTable<K, V1>
focal_class.signature: KTable<K, V1> mapValues(ValueMapper<V, V1> mapper)
focal_class.class_method_signature: KTableImpl.print()
focal_class.full_signature: @Override public void print()
focal_class.identifier: print
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void print()
focal_class.class_method_signature: KTableImpl.print(Serde<K> keySerde, Serde<V> valSerde)
focal_class.full_signature: @Override public void print(Serde<K> keySerde, Serde<V> valSerde)
focal_class.identifier: print
focal_class.modifiers: @Override public
focal_class.parameters: (Serde<K> keySerde, Serde<V> valSerde)
focal_class.return: void
focal_class.signature: void print(Serde<K> keySerde, Serde<V> valSerde)
focal_class.class_method_signature: KTableImpl.writeAsText(String filePath)
focal_class.full_signature: @Override public void writeAsText(String filePath)
focal_class.identifier: writeAsText
focal_class.modifiers: @Override public
focal_class.parameters: (String filePath)
focal_class.return: void
focal_class.signature: void writeAsText(String filePath)
focal_class.class_method_signature: KTableImpl.writeAsText(String filePath, Serde<K> keySerde, Serde<V> valSerde)
focal_class.full_signature: @Override public void writeAsText(String filePath, Serde<K> keySerde, Serde<V> valSerde)
focal_class.identifier: writeAsText
focal_class.modifiers: @Override public
focal_class.parameters: (String filePath, Serde<K> keySerde, Serde<V> valSerde)
focal_class.return: void
focal_class.signature: void writeAsText(String filePath, Serde<K> keySerde, Serde<V> valSerde)
focal_class.class_method_signature: KTableImpl.through(Serde<K> keySerde,
                                  Serde<V> valSerde,
                                  StreamPartitioner<K, V> partitioner,
                                  String topic)
focal_class.full_signature: @Override public KTable<K, V> through(Serde<K> keySerde,
                                  Serde<V> valSerde,
                                  StreamPartitioner<K, V> partitioner,
                                  String topic)
focal_class.identifier: through
focal_class.modifiers: @Override public
focal_class.parameters: (Serde<K> keySerde,
                                  Serde<V> valSerde,
                                  StreamPartitioner<K, V> partitioner,
                                  String topic)
focal_class.return: KTable<K, V>
focal_class.signature: KTable<K, V> through(Serde<K> keySerde,
                                  Serde<V> valSerde,
                                  StreamPartitioner<K, V> partitioner,
                                  String topic)
focal_class.class_method_signature: KTableImpl.foreach(final ForeachAction<K, V> action)
focal_class.full_signature: @Override public void foreach(final ForeachAction<K, V> action)
focal_class.identifier: foreach
focal_class.modifiers: @Override public
focal_class.parameters: (final ForeachAction<K, V> action)
focal_class.return: void
focal_class.signature: void foreach(final ForeachAction<K, V> action)
focal_class.class_method_signature: KTableImpl.through(Serde<K> keySerde, Serde<V> valSerde, String topic)
focal_class.full_signature: @Override public KTable<K, V> through(Serde<K> keySerde, Serde<V> valSerde, String topic)
focal_class.identifier: through
focal_class.modifiers: @Override public
focal_class.parameters: (Serde<K> keySerde, Serde<V> valSerde, String topic)
focal_class.return: KTable<K, V>
focal_class.signature: KTable<K, V> through(Serde<K> keySerde, Serde<V> valSerde, String topic)
focal_class.class_method_signature: KTableImpl.through(StreamPartitioner<K, V> partitioner, String topic)
focal_class.full_signature: @Override public KTable<K, V> through(StreamPartitioner<K, V> partitioner, String topic)
focal_class.identifier: through
focal_class.modifiers: @Override public
focal_class.parameters: (StreamPartitioner<K, V> partitioner, String topic)
focal_class.return: KTable<K, V>
focal_class.signature: KTable<K, V> through(StreamPartitioner<K, V> partitioner, String topic)
focal_class.class_method_signature: KTableImpl.through(String topic)
focal_class.full_signature: @Override public KTable<K, V> through(String topic)
focal_class.identifier: through
focal_class.modifiers: @Override public
focal_class.parameters: (String topic)
focal_class.return: KTable<K, V>
focal_class.signature: KTable<K, V> through(String topic)
focal_class.class_method_signature: KTableImpl.to(String topic)
focal_class.full_signature: @Override public void to(String topic)
focal_class.identifier: to
focal_class.modifiers: @Override public
focal_class.parameters: (String topic)
focal_class.return: void
focal_class.signature: void to(String topic)
focal_class.class_method_signature: KTableImpl.to(StreamPartitioner<K, V> partitioner, String topic)
focal_class.full_signature: @Override public void to(StreamPartitioner<K, V> partitioner, String topic)
focal_class.identifier: to
focal_class.modifiers: @Override public
focal_class.parameters: (StreamPartitioner<K, V> partitioner, String topic)
focal_class.return: void
focal_class.signature: void to(StreamPartitioner<K, V> partitioner, String topic)
focal_class.class_method_signature: KTableImpl.to(Serde<K> keySerde, Serde<V> valSerde, String topic)
focal_class.full_signature: @Override public void to(Serde<K> keySerde, Serde<V> valSerde, String topic)
focal_class.identifier: to
focal_class.modifiers: @Override public
focal_class.parameters: (Serde<K> keySerde, Serde<V> valSerde, String topic)
focal_class.return: void
focal_class.signature: void to(Serde<K> keySerde, Serde<V> valSerde, String topic)
focal_class.class_method_signature: KTableImpl.to(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<K, V> partitioner, String topic)
focal_class.full_signature: @Override public void to(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<K, V> partitioner, String topic)
focal_class.identifier: to
focal_class.modifiers: @Override public
focal_class.parameters: (Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<K, V> partitioner, String topic)
focal_class.return: void
focal_class.signature: void to(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<K, V> partitioner, String topic)
focal_class.class_method_signature: KTableImpl.toStream()
focal_class.full_signature: @Override public KStream<K, V> toStream()
focal_class.identifier: toStream
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: KStream<K, V>
focal_class.signature: KStream<K, V> toStream()
focal_class.class_method_signature: KTableImpl.toStream(KeyValueMapper<K, V, K1> mapper)
focal_class.full_signature: @Override public KStream<K1, V> toStream(KeyValueMapper<K, V, K1> mapper)
focal_class.identifier: toStream
focal_class.modifiers: @Override public
focal_class.parameters: (KeyValueMapper<K, V, K1> mapper)
focal_class.return: KStream<K1, V>
focal_class.signature: KStream<K1, V> toStream(KeyValueMapper<K, V, K1> mapper)
focal_class.class_method_signature: KTableImpl.join(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.full_signature: @SuppressWarnings("unchecked") @Override public KTable<K, R> join(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.identifier: join
focal_class.modifiers: @SuppressWarnings("unchecked") @Override public
focal_class.parameters: (KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.return: KTable<K, R>
focal_class.signature: KTable<K, R> join(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.class_method_signature: KTableImpl.outerJoin(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.full_signature: @SuppressWarnings("unchecked") @Override public KTable<K, R> outerJoin(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.identifier: outerJoin
focal_class.modifiers: @SuppressWarnings("unchecked") @Override public
focal_class.parameters: (KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.return: KTable<K, R>
focal_class.signature: KTable<K, R> outerJoin(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.class_method_signature: KTableImpl.leftJoin(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.full_signature: @SuppressWarnings("unchecked") @Override public KTable<K, R> leftJoin(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.identifier: leftJoin
focal_class.modifiers: @SuppressWarnings("unchecked") @Override public
focal_class.parameters: (KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.return: KTable<K, R>
focal_class.signature: KTable<K, R> leftJoin(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner)
focal_class.class_method_signature: KTableImpl.groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector,
                                                    Serde<K1> keySerde,
                                                    Serde<V1> valueSerde)
focal_class.full_signature: @Override public KGroupedTable<K1, V1> groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector,
                                                    Serde<K1> keySerde,
                                                    Serde<V1> valueSerde)
focal_class.identifier: groupBy
focal_class.modifiers: @Override public
focal_class.parameters: (KeyValueMapper<K, V, KeyValue<K1, V1>> selector,
                                                    Serde<K1> keySerde,
                                                    Serde<V1> valueSerde)
focal_class.return: KGroupedTable<K1, V1>
focal_class.signature: KGroupedTable<K1, V1> groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector,
                                                    Serde<K1> keySerde,
                                                    Serde<V1> valueSerde)
focal_class.class_method_signature: KTableImpl.groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector)
focal_class.full_signature: @Override public KGroupedTable<K1, V1> groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector)
focal_class.identifier: groupBy
focal_class.modifiers: @Override public
focal_class.parameters: (KeyValueMapper<K, V, KeyValue<K1, V1>> selector)
focal_class.return: KGroupedTable<K1, V1>
focal_class.signature: KGroupedTable<K1, V1> groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector)
focal_class.class_method_signature: KTableImpl.valueGetterSupplier()
focal_class.full_signature: @SuppressWarnings("unchecked") KTableValueGetterSupplier<K, V> valueGetterSupplier()
focal_class.identifier: valueGetterSupplier
focal_class.modifiers: @SuppressWarnings("unchecked")
focal_class.parameters: ()
focal_class.return: KTableValueGetterSupplier<K, V>
focal_class.signature: KTableValueGetterSupplier<K, V> valueGetterSupplier()
focal_class.class_method_signature: KTableImpl.enableSendingOldValues()
focal_class.full_signature: @SuppressWarnings("unchecked") void enableSendingOldValues()
focal_class.identifier: enableSendingOldValues
focal_class.modifiers: @SuppressWarnings("unchecked")
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void enableSendingOldValues()
focal_class.class_method_signature: KTableImpl.sendingOldValueEnabled()
focal_class.full_signature:  boolean sendingOldValueEnabled()
focal_class.identifier: sendingOldValueEnabled
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean sendingOldValueEnabled()
focal_class.class_method_signature: KTableImpl.materialize(KTableSource<K, ?> source)
focal_class.full_signature: private void materialize(KTableSource<K, ?> source)
focal_class.identifier: materialize
focal_class.modifiers: private
focal_class.parameters: (KTableSource<K, ?> source)
focal_class.return: void
focal_class.signature: void materialize(KTableSource<K, ?> source)
focal_class.superclass: extends AbstractStream<K>
focal_method.body: @Override
      public <K1, V1> KGroupedTable<K1, V1> groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector,
                                                    Serde<K1> keySerde,
                                                    Serde<V1> valueSerde) {
  
          String selectName = topology.newName(SELECT_NAME);
  
          KTableProcessorSupplier<K, V, KeyValue<K1, V1>> selectSupplier = new KTableRepartitionMap<>(this, selector);
  
          // select the aggregate key and values (old and new), it would require parent to send old values
          topology.addProcessor(selectName, selectSupplier, this.name);
          this.enableSendingOldValues();
  
          return new KGroupedTableImpl<>(topology, selectName, this.name, keySerde, valueSerde);
      }
focal_method.class_method_signature: KTableImpl.groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector,
                                                    Serde<K1> keySerde,
                                                    Serde<V1> valueSerde)
focal_method.full_signature: @Override public KGroupedTable<K1, V1> groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector,
                                                    Serde<K1> keySerde,
                                                    Serde<V1> valueSerde)
focal_method.identifier: groupBy
focal_method.invocations: newName addProcessor enableSendingOldValues
focal_method.modifiers: @Override public
focal_method.parameters: (KeyValueMapper<K, V, KeyValue<K1, V1>> selector,
                                                    Serde<K1> keySerde,
                                                    Serde<V1> valueSerde)
focal_method.return: KGroupedTable<K1, V1>
focal_method.signature: KGroupedTable<K1, V1> groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector,
                                                    Serde<K1> keySerde,
                                                    Serde<V1> valueSerde)