@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());
}