I Need to Join multiple KTables(to join kafka messages from them) by foreign keys. But this block of code can join them only in such blocking way. Is there are some way to join KTables asynchronously?
KTable<String, JoinedFoo> master = ...;
KTable<String, Foo> fooKTable = ...;
KTable<String, Bar> barKTable = ...;
KTable<String, JoinedFoo> afterFoo = joinFoo(master, fooKtable);
// needs 'afterFoo' as a param instead of 'master'
KTable<String, JoinedFoo> afterBar = joinBar(afterFoo, barKTable);
This is my methods with join logic:
private KTable<String, JoinedFoo> joinFoo(
KTable<String, JoinedFoo> master,
KTable<String, Foo> fooKTable) {
return master
.join(fooKTable,
joinedFooMessage ->
joinedFooMessage.getFooId(),
(joinedFooMessage, fooMessage) -> {
joinedFoo.setFoo(fooMessage);
return joinedFoo;
});
}
private KTable<String, JoinedFoo> joinBar(
KTable<String, JoinedFoo> master,
KTable<String, Bar> barKTable) {
return master
.join(barKTable,
joinedFooMessage ->
joinedFooMessage.getBarId(),
(joinedFooMessage, barMessage) -> {
joinedFoo.setBar(barMessage);
return joinedFoo;
});
}