@Test public void testJoin() throws Exception { final KStreamBuilder builder = new KStreamBuilder(); final int[] expectedKeys = new int[]{0, 1, 2, 3}; KTable<Integer, String> table1 = builder.table(intSerde, stringSerde, topic1, storeName1); KTable<Integer, String> table2 = builder.table(intSerde, stringSerde, topic2, storeName2); KTable<Integer, String> joined = table1.leftJoin(table2, MockValueJoiner.TOSTRING_JOINER); MockProcessorSupplier<Integer, String> processor; processor = new MockProcessorSupplier<>(); 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]); } driver.process(topic1, null, "SomeVal"); driver.flushState(); 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]); } driver.process(topic2, null, "AnotherVal"); driver.flushState(); 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 expectedKey : expectedKeys) { driver.process(topic1, expectedKey, "X" + expectedKey); } driver.flushState(); 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 expectedKey : expectedKeys) { driver.process(topic2, expectedKey, "YY" + expectedKey); } driver.flushState(); 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 expectedKey : expectedKeys) { driver.process(topic1, expectedKey, "X" + expectedKey); } driver.flushState(); 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); } driver.flushState(); 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 expectedKey : expectedKeys) { driver.process(topic1, expectedKey, "XX" + expectedKey); } driver.flushState(); 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")); }
@Override public Processor<K, Change<V1>> get() { return new KTableKTableLeftJoinProcessor(valueGetterSupplier2.get()); }
KTableKTableLeftJoin extends KTableKTableAbstractJoin<K, R, V1, V2> { @Override public Processor<K, Change<V1>> get() { return new KTableKTableLeftJoinProcessor(valueGetterSupplier2.get()); } }
KTableKTableLeftJoin extends KTableKTableAbstractJoin<K, R, V1, V2> { @Override public Processor<K, Change<V1>> get() { return new KTableKTableLeftJoinProcessor(valueGetterSupplier2.get()); } KTableKTableLeftJoin(KTableImpl<K, ?, V1> table1, KTableImpl<K, ?, V2> table2, ValueJoiner<? super V1, ? super V2, ? extends R> joiner); }
KTableKTableLeftJoin extends KTableKTableAbstractJoin<K, R, V1, V2> { @Override public Processor<K, Change<V1>> get() { return new KTableKTableLeftJoinProcessor(valueGetterSupplier2.get()); } KTableKTableLeftJoin(KTableImpl<K, ?, V1> table1, KTableImpl<K, ?, V2> table2, ValueJoiner<? super V1, ? super V2, ? extends R> joiner); @Override Processor<K, Change<V1>> get(); @Override KTableValueGetterSupplier<K, R> view(); }
KTableKTableLeftJoin extends KTableKTableAbstractJoin<K, R, V1, V2> { @Override public Processor<K, Change<V1>> get() { return new KTableKTableLeftJoinProcessor(valueGetterSupplier2.get()); } KTableKTableLeftJoin(KTableImpl<K, ?, V1> table1, KTableImpl<K, ?, V2> table2, ValueJoiner<? super V1, ? super V2, ? extends R> joiner); @Override Processor<K, Change<V1>> get(); @Override KTableValueGetterSupplier<K, R> view();  }