Kafka KStream to KStream left join not matching as expected

Viewed 34

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.

0 Answers
Related