@Test public void testRepartition() throws Exception { String topic1 = "topic1"; String storeName1 = "storeName1"; final KStreamBuilder builder = new KStreamBuilder(); KTableImpl<String, String, String> table1 = (KTableImpl<String, String, String>) builder.table(stringSerde, stringSerde, topic1, storeName1); KTableImpl<String, String, String> table1Aggregated = (KTableImpl<String, String, String>) table1 .groupBy(MockKeyValueMapper.<String, String>NoOpKeyValueMapper()) .aggregate(MockInitializer.STRING_INIT, MockAggregator.TOSTRING_ADDER, MockAggregator.TOSTRING_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")); Field valSerializerField = ((SinkNode) driver.processor("KSTREAM-SINK-0000000003")).getClass().getDeclaredField("valSerializer"); Field valDeserializerField = ((SourceNode) driver.processor("KSTREAM-SOURCE-0000000004")).getClass().getDeclaredField("valDeserializer"); valSerializerField.setAccessible(true); valDeserializerField.setAccessible(true); assertNotNull(((ChangedSerializer) valSerializerField.get(driver.processor("KSTREAM-SINK-0000000003"))).inner()); assertNotNull(((ChangedDeserializer) valDeserializerField.get(driver.processor("KSTREAM-SOURCE-0000000004"))).inner()); assertNotNull(((ChangedSerializer) valSerializerField.get(driver.processor("KSTREAM-SINK-0000000007"))).inner()); assertNotNull(((ChangedDeserializer) valDeserializerField.get(driver.processor("KSTREAM-SOURCE-0000000008"))).inner()); }
@Override public <K1, V1> KGroupedTable<K1, V1> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<K1, V1>> selector, Serde<K1> keySerde, Serde<V1> valueSerde) { Objects.requireNonNull(selector, "selector can't be null"); String selectName = topology.newName(SELECT_NAME); KTableProcessorSupplier<K, V, KeyValue<K1, V1>> selectSupplier = new KTableRepartitionMap<K, V, K1, V1>(this, selector); topology.addProcessor(selectName, selectSupplier, this.name); this.enableSendingOldValues(); return new KGroupedTableImpl<>(topology, selectName, this.name, keySerde, valueSerde); }
KTableImpl extends AbstractStream<K> implements KTable<K, V> { @Override public <K1, V1> KGroupedTable<K1, V1> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<K1, V1>> selector, Serde<K1> keySerde, Serde<V1> valueSerde) { Objects.requireNonNull(selector, "selector can't be null"); String selectName = topology.newName(SELECT_NAME); KTableProcessorSupplier<K, V, KeyValue<K1, V1>> selectSupplier = new KTableRepartitionMap<K, V, K1, V1>(this, selector); topology.addProcessor(selectName, selectSupplier, this.name); this.enableSendingOldValues(); return new KGroupedTableImpl<>(topology, selectName, this.name, keySerde, valueSerde); } }
KTableImpl extends AbstractStream<K> implements KTable<K, V> { @Override public <K1, V1> KGroupedTable<K1, V1> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<K1, V1>> selector, Serde<K1> keySerde, Serde<V1> valueSerde) { Objects.requireNonNull(selector, "selector can't be null"); String selectName = topology.newName(SELECT_NAME); KTableProcessorSupplier<K, V, KeyValue<K1, V1>> selectSupplier = new KTableRepartitionMap<K, V, K1, V1>(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,
                      final String queryableStoreName,
                      boolean isQueryable);  KTableImpl(KStreamBuilder topology,
                      String name,
                      ProcessorSupplier<?, ?> processorSupplier,
                      final Serde<K> keySerde,
                      final Serde<V> valSerde,
                      Set<String> sourceNodes,
                      final String queryableStoreName,
                      boolean isQueryable); }
KTableImpl extends AbstractStream<K> implements KTable<K, V> { @Override public <K1, V1> KGroupedTable<K1, V1> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<K1, V1>> selector, Serde<K1> keySerde, Serde<V1> valueSerde) { Objects.requireNonNull(selector, "selector can't be null"); String selectName = topology.newName(SELECT_NAME); KTableProcessorSupplier<K, V, KeyValue<K1, V1>> selectSupplier = new KTableRepartitionMap<K, V, K1, V1>(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,
                      final String queryableStoreName,
                      boolean isQueryable);  KTableImpl(KStreamBuilder topology,
                      String name,
                      ProcessorSupplier<?, ?> processorSupplier,
                      final Serde<K> keySerde,
                      final Serde<V> valSerde,
                      Set<String> sourceNodes,
                      final String queryableStoreName,
                      boolean isQueryable); @Override String queryableStoreName(); @Override KTable<K, V> filter(final Predicate<? super K, ? super V> predicate); @Override KTable<K, V> filter(final Predicate<? super K, ? super V> predicate, final String queryableStoreName); @Override KTable<K, V> filter(final Predicate<? super K, ? super V> predicate, final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate); @Override KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate, final String queryableStoreName); @Override KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate, final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V1> mapValues(final ValueMapper<? super V, ? extends V1> mapper); @Override KTable<K, V1> mapValues(final ValueMapper<? super V, ? extends V1> mapper,
                                        final Serde<V1> valueSerde,
                                        final String queryableStoreName); @Override KTable<K, V1> mapValues(final ValueMapper<? super V, ? extends V1> mapper,
                                         final Serde<V1> valueSerde,
                                         final StateStoreSupplier<KeyValueStore> storeSupplier); @Override void print(); @Override void print(String streamName); @Override void print(Serde<K> keySerde, Serde<V> valSerde); @Override void print(Serde<K> keySerde, Serde<V> valSerde, String streamName); @Override void writeAsText(String filePath); @Override void writeAsText(String filePath, String streamName); @Override void writeAsText(String filePath, Serde<K> keySerde, Serde<V> valSerde); @Override void writeAsText(String filePath, String streamName, Serde<K> keySerde, Serde<V> valSerde); @Override void foreach(final ForeachAction<? super K, ? super V> action); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic,
                                final String queryableStoreName); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic,
                                final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final String topic,
                                final String queryableStoreName); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final String topic,
                                final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final String topic); @Override KTable<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic,
                                final String queryableStoreName); @Override KTable<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic,
                                final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic); @Override KTable<K, V> through(final String topic,
                                final String queryableStoreName); @Override KTable<K, V> through(final String topic,
                                final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> through(final String topic); @Override void to(String topic); @Override void to(StreamPartitioner<? super K, ? super 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<? super K, ? super V> partitioner, String topic); @Override KStream<K, V> toStream(); @Override KStream<K1, V> toStream(KeyValueMapper<? super K, ? super V, ? extends K1> mapper); @Override KTable<K, R> join(final KTable<K, V1> other,
                                     final ValueJoiner<? super V, ? super V1, ? extends R> joiner); @Override KTable<K, R> join(final KTable<K, V1> other,
                                     final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                     final Serde<R> joinSerde,
                                     final String queryableStoreName); @Override KTable<K, R> join(final KTable<K, V1> other,
                                     final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                     final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, R> outerJoin(final KTable<K, V1> other,
                                          final ValueJoiner<? super V, ? super V1, ? extends R> joiner); @Override KTable<K, R> outerJoin(final KTable<K, V1> other,
                                          final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                          final Serde<R> joinSerde,
                                          final String queryableStoreName); @Override KTable<K, R> outerJoin(final KTable<K, V1> other,
                                          final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                          final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, R> leftJoin(final KTable<K, V1> other,
                                         final ValueJoiner<? super V, ? super V1, ? extends R> joiner); @Override KTable<K, R> leftJoin(final KTable<K, V1> other,
                                         final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                         final Serde<R> joinSerde,
                                         final String queryableStoreName); @Override KTable<K, R> leftJoin(final KTable<K, V1> other,
                                         final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                         final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KGroupedTable<K1, V1> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<K1, V1>> selector,
                                                  Serde<K1> keySerde,
                                                  Serde<V1> valueSerde); @Override KGroupedTable<K1, V1> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<K1, V1>> selector); }
KTableImpl extends AbstractStream<K> implements KTable<K, V> { @Override public <K1, V1> KGroupedTable<K1, V1> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<K1, V1>> selector, Serde<K1> keySerde, Serde<V1> valueSerde) { Objects.requireNonNull(selector, "selector can't be null"); String selectName = topology.newName(SELECT_NAME); KTableProcessorSupplier<K, V, KeyValue<K1, V1>> selectSupplier = new KTableRepartitionMap<K, V, K1, V1>(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,
                      final String queryableStoreName,
                      boolean isQueryable);  KTableImpl(KStreamBuilder topology,
                      String name,
                      ProcessorSupplier<?, ?> processorSupplier,
                      final Serde<K> keySerde,
                      final Serde<V> valSerde,
                      Set<String> sourceNodes,
                      final String queryableStoreName,
                      boolean isQueryable); @Override String queryableStoreName(); @Override KTable<K, V> filter(final Predicate<? super K, ? super V> predicate); @Override KTable<K, V> filter(final Predicate<? super K, ? super V> predicate, final String queryableStoreName); @Override KTable<K, V> filter(final Predicate<? super K, ? super V> predicate, final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate); @Override KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate, final String queryableStoreName); @Override KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate, final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V1> mapValues(final ValueMapper<? super V, ? extends V1> mapper); @Override KTable<K, V1> mapValues(final ValueMapper<? super V, ? extends V1> mapper,
                                        final Serde<V1> valueSerde,
                                        final String queryableStoreName); @Override KTable<K, V1> mapValues(final ValueMapper<? super V, ? extends V1> mapper,
                                         final Serde<V1> valueSerde,
                                         final StateStoreSupplier<KeyValueStore> storeSupplier); @Override void print(); @Override void print(String streamName); @Override void print(Serde<K> keySerde, Serde<V> valSerde); @Override void print(Serde<K> keySerde, Serde<V> valSerde, String streamName); @Override void writeAsText(String filePath); @Override void writeAsText(String filePath, String streamName); @Override void writeAsText(String filePath, Serde<K> keySerde, Serde<V> valSerde); @Override void writeAsText(String filePath, String streamName, Serde<K> keySerde, Serde<V> valSerde); @Override void foreach(final ForeachAction<? super K, ? super V> action); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic,
                                final String queryableStoreName); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic,
                                final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final String topic,
                                final String queryableStoreName); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final String topic,
                                final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> through(final Serde<K> keySerde,
                                final Serde<V> valSerde,
                                final String topic); @Override KTable<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic,
                                final String queryableStoreName); @Override KTable<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic,
                                final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
                                final String topic); @Override KTable<K, V> through(final String topic,
                                final String queryableStoreName); @Override KTable<K, V> through(final String topic,
                                final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, V> through(final String topic); @Override void to(String topic); @Override void to(StreamPartitioner<? super K, ? super 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<? super K, ? super V> partitioner, String topic); @Override KStream<K, V> toStream(); @Override KStream<K1, V> toStream(KeyValueMapper<? super K, ? super V, ? extends K1> mapper); @Override KTable<K, R> join(final KTable<K, V1> other,
                                     final ValueJoiner<? super V, ? super V1, ? extends R> joiner); @Override KTable<K, R> join(final KTable<K, V1> other,
                                     final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                     final Serde<R> joinSerde,
                                     final String queryableStoreName); @Override KTable<K, R> join(final KTable<K, V1> other,
                                     final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                     final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, R> outerJoin(final KTable<K, V1> other,
                                          final ValueJoiner<? super V, ? super V1, ? extends R> joiner); @Override KTable<K, R> outerJoin(final KTable<K, V1> other,
                                          final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                          final Serde<R> joinSerde,
                                          final String queryableStoreName); @Override KTable<K, R> outerJoin(final KTable<K, V1> other,
                                          final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                          final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KTable<K, R> leftJoin(final KTable<K, V1> other,
                                         final ValueJoiner<? super V, ? super V1, ? extends R> joiner); @Override KTable<K, R> leftJoin(final KTable<K, V1> other,
                                         final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                         final Serde<R> joinSerde,
                                         final String queryableStoreName); @Override KTable<K, R> leftJoin(final KTable<K, V1> other,
                                         final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
                                         final StateStoreSupplier<KeyValueStore> storeSupplier); @Override KGroupedTable<K1, V1> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<K1, V1>> selector,
                                                  Serde<K1> keySerde,
                                                  Serde<V1> valueSerde); @Override KGroupedTable<K1, V1> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<K1, V1>> selector); static final String JOINTHIS_NAME; static final String JOINOTHER_NAME; static final String MERGE_NAME; static final String SOURCE_NAME; static final String STATE_STORE_NAME; }