@Override
    public <K1, V1> KGroupedTable<K1, V1> groupBy(final KeyValueMapper<? super K, ? super V, KeyValue<K1, V1>> selector) {
        return groupBy(selector, Grouped.with(null, null));
    }
@Test
    public void shouldCreateSourceAndSinkNodesForRepartitioningTopic() throws Exception {
        final StreamsBuilder builder = new StreamsBuilder();
        final String topic1 = "topic1";
        final String storeName1 = "storeName1";

        final KTableImpl<String, String, String> table1 =
            (KTableImpl<String, String, String>) builder.table(
                topic1,
                consumed,
                Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as(storeName1)
                    .withKeySerde(Serdes.String())
                    .withValueSerde(Serdes.String())
            );

        table1.groupBy(MockMapper.noOpKeyValueMapper())
            .aggregate(
                MockInitializer.STRING_INIT,
                MockAggregator.TOSTRING_ADDER,
                MockAggregator.TOSTRING_REMOVER,
                Materialized.as("mock-result1"));

        table1.groupBy(MockMapper.noOpKeyValueMapper())
            .reduce(
                MockReducer.STRING_ADDER,
                MockReducer.STRING_REMOVER,
                Materialized.as("mock-result2"));

        final Topology topology = builder.build();
        try (final TopologyTestDriverWrapper driver = new TopologyTestDriverWrapper(topology, props)) {

            assertEquals(3, driver.getAllStateStores().size());

            assertTopologyContainsProcessor(topology, "KSTREAM-SINK-0000000003");
            assertTopologyContainsProcessor(topology, "KSTREAM-SOURCE-0000000004");
            assertTopologyContainsProcessor(topology, "KSTREAM-SINK-0000000007");
            assertTopologyContainsProcessor(topology, "KSTREAM-SOURCE-0000000008");

            final Field valSerializerField = ((SinkNode) driver.getProcessor("KSTREAM-SINK-0000000003"))
                .getClass()
                .getDeclaredField("valSerializer");
            final Field valDeserializerField = ((SourceNode) driver.getProcessor("KSTREAM-SOURCE-0000000004"))
                .getClass()
                .getDeclaredField("valDeserializer");
            valSerializerField.setAccessible(true);
            valDeserializerField.setAccessible(true);

            assertNotNull(((ChangedSerializer) valSerializerField.get(driver.getProcessor("KSTREAM-SINK-0000000003"))).inner());
            assertNotNull(((ChangedDeserializer) valDeserializerField.get(driver.getProcessor("KSTREAM-SOURCE-0000000004"))).inner());
            assertNotNull(((ChangedSerializer) valSerializerField.get(driver.getProcessor("KSTREAM-SINK-0000000007"))).inner());
            assertNotNull(((ChangedDeserializer) valDeserializerField.get(driver.getProcessor("KSTREAM-SOURCE-0000000008"))).inner());
        }
    }