KTableImpl extends AbstractStream<K> implements KTable<K, V> { @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); topology.addProcessor(selectName, selectSupplier, this.name); this.enableSendingOldValues(); return new KGroupedTableImpl<>(topology, selectName, this.name, keySerde, valueSerde); } KTableImpl(KStreamBuilder topology,
                      String name,
                      ProcessorSupplier<?, ?> processorSupplier,
                      Set<String> sourceNodes);  KTableImpl(KStreamBuilder topology,
                      String name,
                      ProcessorSupplier<?, ?> processorSupplier,
                      Set<String> sourceNodes,
                      Serde<K> keySerde,
                      Serde<V> valSerde); @Override KTable<K, V> filter(Predicate<K, V> predicate); @Override KTable<K, V> filterNot(final Predicate<K, V> predicate); @Override KTable<K, V1> mapValues(ValueMapper<V, V1> mapper); @Override void print(); @Override void print(Serde<K> keySerde, Serde<V> valSerde); @Override void writeAsText(String filePath); @Override void writeAsText(String filePath, Serde<K> keySerde, Serde<V> valSerde); @Override KTable<K, V> through(Serde<K> keySerde,
                                Serde<V> valSerde,
                                StreamPartitioner<K, V> partitioner,
                                String topic); @Override void foreach(final ForeachAction<K, V> action); @Override KTable<K, V> through(Serde<K> keySerde, Serde<V> valSerde, String topic); @Override KTable<K, V> through(StreamPartitioner<K, V> partitioner, String topic); @Override KTable<K, V> through(String topic); @Override void to(String topic); @Override void to(StreamPartitioner<K, V> partitioner, String topic); @Override void to(Serde<K> keySerde, Serde<V> valSerde, String topic); @Override void to(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<K, V> partitioner, String topic); @Override KStream<K, V> toStream(); @Override KStream<K1, V> toStream(KeyValueMapper<K, V, K1> mapper); @SuppressWarnings("unchecked") @Override KTable<K, R> join(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner); @SuppressWarnings("unchecked") @Override KTable<K, R> outerJoin(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner); @SuppressWarnings("unchecked") @Override KTable<K, R> leftJoin(KTable<K, V1> other, ValueJoiner<V, V1, R> joiner); @Override KGroupedTable<K1, V1> groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector,
                                                  Serde<K1> keySerde,
                                                  Serde<V1> valueSerde); @Override KGroupedTable<K1, V1> groupBy(KeyValueMapper<K, V, KeyValue<K1, V1>> selector); static final String JOINTHIS_NAME; static final String JOINOTHER_NAME; static final String LEFTTHIS_NAME; static final String LEFTOTHER_NAME; static final String MERGE_NAME; static final String OUTERTHIS_NAME; static final String OUTEROTHER_NAME; static final String SOURCE_NAME; final ProcessorSupplier<?, ?> processorSupplier; }
@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); assertEquals(3, driver.allStateStores().size()); 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()); }