KTableKTableOuterJoin extends KTableKTableAbstractJoin<K, R, V1, V2> { @Override public Processor<K, Change<V1>> get() { return new KTableKTableOuterJoinProcessor(valueGetterSupplier2.get()); } KTableKTableOuterJoin(KTableImpl<K, ?, V1> table1, KTableImpl<K, ?, V2> table2, ValueJoiner<V1, V2, R> joiner); @Override Processor<K, Change<V1>> get(); @Override KTableValueGetterSupplier<K, R> view();  }
@Test public void testJoin() throws Exception { 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.outerJoin(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()); for (int i = 0; i < 2; i++) { driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]); } processor.checkAndClearProcessResult("0:X0+null", "1:X1+null"); checkJoinedValues(getter, kv(0, "X0+null"), kv(1, "X1+null"), kv(2, null), kv(3, null)); 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)); 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:X2+null", "3:X3+null"); checkJoinedValues(getter, kv(0, "X0+Y0"), kv(1, "X1+Y1"), kv(2, "X2+null"), kv(3, "X3+null")); 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")); 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")); for (int i = 0; i < 2; i++) { driver.process(topic2, expectedKeys[i], null); } processor.checkAndClearProcessResult("0:X0+null", "1:X1+null"); checkJoinedValues(getter, kv(0, "X0+null"), kv(1, "X1+null"), kv(2, "X2+YY2"), kv(3, "X3+YY3")); for (int i = 0; i < expectedKeys.length; i++) { driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]); } processor.checkAndClearProcessResult("0:XX0+null", "1:XX1+null", "2:XX2+YY2", "3:XX3+YY3"); checkJoinedValues(getter, kv(0, "XX0+null"), kv(1, "XX1+null"), kv(2, "XX2+YY2"), kv(3, "XX3+YY3")); for (int i = 1; i < 3; i++) { driver.process(topic1, expectedKeys[i], null); } processor.checkAndClearProcessResult("1:null", "2:null+YY2"); checkJoinedValues(getter, kv(0, "XX0+null"), kv(1, null), kv(2, "null+YY2"), kv(3, "XX3+YY3")); }