Hello,
I'm trying to understand best practices related to joining streams using
the Kafka Streams API.
I can configure the topology such that two sources feed into a single
processor:
topologyBuilder
.addSource("A", stringDeserializer, itemDeserializer, "a-topic")
.addSource("B", stringDeserializer, itemDeserializer, "b-topic)
.addProcessor("hello-join", HelloJoin::new, "A", "B")...
And within my processor I can determine which topic a given message came
from:
public void process(String Key, String value) {
if (context.topic.equals("a-topic") {
...
} else {
...
}
This allows for a crude form of cross stream join with the following
issues/limitations:
i. A string compare on topic name to decide which stream a message came
from. Having actual access to the TopicPartition could lead to more
efficient validation. Priority low, as this is just a small performance
hit, but it is a per message performance hit so would be nice to eliminate.
ii. This requires "a-topic" and "b-topic" to have the same message
format, which for general join handling is a pretty big limitation. What
would be the recommended way to handle the case of different message
formats, e.g. needing different deserializers for different input topics?
E.g. how would I define my Processor if the topology was:
topologyBuilder
.addSource("A", stringDeserializer, itemADeserializer, "a-topic")
.addSource("B", stringDeserializer, itemBDeserializer, "b-topic)
.addProcessor("hello-join", HelloJoin::new, "A", "B")...
where itemADeserializer and itemBDeserializer return different classes?
Thanks,
Caleb
