Inner Join on KStreams

[memaskal]

Hello, I am trying to do an Inner Join (without windowing) in KStream but I can't make it work. Could you provide some example on how to use them properly? This is the example code I have tested:

function keyValueMapperEtl(message){
    const val = JSON.parse(message.value);
    return {
        key: message.key.toString(),
        value: val.number,
    };
}

const st1 = kafkaStreams
    .getKStream("testA_out")
    .map(keyValueMapperEtl);

const st2 = kafkaStreams
    .getKStream("testB_out")
    .map(keyValueMapperEtl);

const st3 = st1.innerJoin(st2);
st3.forEach((v) => console.log(v));  // key missmatch

st1.start();
st2.start();

[krystianity]

Hi @memaskal what do you mean with // key missmatch ? Can you post your output? And the output that you are expecting?

[memaskal]

Hello @krystianity, forgot to close the issue. I found out that streams don't perform a classic inner join (db style on the same key value). The KTables inner join would be more appropriate to my problem, but are not implemented yet. Thank you for the response, great job with this project !