I have a Kafka Streams topology that starts with a KStream from a topic of Event messages. Each Event has a Patient ID. This KStream then joins with a patient demographics KStream to enhance the resulting value's patient information.
Both topics have 30 partitions and both use the Patient ID as the key. Therefore it's co-partitioned and the data in both topics are sharded based on the same key.
The problem is that even though every patient is in TOPIC_PATIENT_INFO, quite often the demographic record isn't found in the left join. For example, in the ValueJoiner below, the demog object will be null.
KStream<Integer, GenericRecord> ksEvents = builder.stream(TOPIC_EVENTS);
KStream<Integer, GenericRecord> ksDemogs = builder.stream(TOPIC_PATIENT_INFO);
ksEvents
.peek((key, value) -> log.info("before join k:{}, v:{}", key, value))
.leftJoin(ksDemogs, getValueJoinerWithKey(), JoinWindows.of(Duration.ofDays(500000)))
.peek((key, value) -> log.info("after join k:{}, v:{}", key, value))
elsewhere in the code:
ValueJoinerWithKey<Integer, GenericRecord, GenericRecord, GenericRecord> getValueJoinerWithKey() {
return (key, event, demog) -> {
// demog *should* never be null because every patient ID is in the patient_info topic.
}
}
For the demographics, I could have used a GlobalKTable instead of a KStream yet this would remove the advantage of co-partitioning and require that all instances of the streaming app contain all the demographic records; I'd like to use sharding if possible.
FYI: the large JoinWindow is to guarantee that no demographic records are ignored based on having an older timestamp.