@Test
public void testJoin() throws Exception {

    final KStreamBuilder builder = new KStreamBuilder();
    final int[] expectedKeys = new int[]{0, 1, 2, 3};
    KTable<Integer, String> table1;
    KTable<Integer, String> table2;
    KTable<Integer, String> joined;
    MockProcessorSupplier<Integer, String> processor;
    processor = new MockProcessorSupplier<>();
    table1 = builder.table(intSerde, stringSerde, topic1);
    table2 = builder.table(intSerde, stringSerde, topic2);
    joined = table1.join(table2, MockValueJoiner.STRING_JOINER);
    joined.toStream().process(processor);
    Collection<Set<String>> copartitionGroups = builder.copartitionGroups();
    assertEquals(1, copartitionGroups.size());
    assertEquals(new HashSet<>(Arrays.asList(topic1, topic2)), copartitionGroups.iterator().next());
    KTableValueGetterSupplier<Integer, String> getterSupplier = ((KTableImpl<Integer, String, String>) joined).valueGetterSupplier();
    driver = new KStreamTestDriver(builder, stateDir);
    driver.setTime(0L);
    KTableValueGetter<Integer, String> getter = getterSupplier.get();
    getter.init(driver.context());
    // push two items to the primary stream. the other table is empty
    for (int i = 0; i < 2; i++) {
        driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
    }
    processor.checkAndClearProcessResult("0:null", "1:null");
    checkJoinedValues(getter, kv(0, null), kv(1, null), kv(2, null), kv(3, null));
    // push two items to the other stream. this should produce two items.
    for (int i = 0; i < 2; i++) {
        driver.process(topic2, expectedKeys[i], "Y" + expectedKeys[i]);
    }
    processor.checkAndClearProcessResult("0:X0+Y0", "1:X1+Y1");
    checkJoinedValues(getter, kv(0, "X0+Y0"), kv(1, "X1+Y1"), kv(2, null), kv(3, null));
    // push all four items to the primary stream. this should produce four items.
    for (int i = 0; i < expectedKeys.length; i++) {
        driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
    }
    processor.checkAndClearProcessResult("0:X0+Y0", "1:X1+Y1", "2:null", "3:null");
    checkJoinedValues(getter, kv(0, "X0+Y0"), kv(1, "X1+Y1"), kv(2, null), kv(3, null));
    // push all items to the other stream. this should produce four items.
    for (int i = 0; i < expectedKeys.length; i++) {
        driver.process(topic2, expectedKeys[i], "YY" + expectedKeys[i]);
    }
    processor.checkAndClearProcessResult("0:X0+YY0", "1:X1+YY1", "2:X2+YY2", "3:X3+YY3");
    checkJoinedValues(getter, kv(0, "X0+YY0"), kv(1, "X1+YY1"), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
    // push all four items to the primary stream. this should produce four items.
    for (int i = 0; i < expectedKeys.length; i++) {
        driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
    }
    processor.checkAndClearProcessResult("0:X0+YY0", "1:X1+YY1", "2:X2+YY2", "3:X3+YY3");
    checkJoinedValues(getter, kv(0, "X0+YY0"), kv(1, "X1+YY1"), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
    // push two items with null to the other stream as deletes. this should produce two item.
    for (int i = 0; i < 2; i++) {
        driver.process(topic2, expectedKeys[i], null);
    }
    processor.checkAndClearProcessResult("0:null", "1:null");
    checkJoinedValues(getter, kv(0, null), kv(1, null), kv(2, "X2+YY2"), kv(3, "X3+YY3"));
    // push all four items to the primary stream. this should produce four items.
    for (int i = 0; i < expectedKeys.length; i++) {
        driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]);
    }
    processor.checkAndClearProcessResult("0:null", "1:null", "2:XX2+YY2", "3:XX3+YY3");
    checkJoinedValues(getter, kv(0, null), kv(1, null), kv(2, "XX2+YY2"), kv(3, "XX3+YY3"));
}