KStream join Retry / Delayed KStream

Question

We are trying to implement use case which described below, we are having implementation issues which we looking to overcome,

Use Case,

We are trying to do KStream join between 2 Kafka Topics by matching KEY present in messages(JSON) of both streams. Also we should maintain the message sequence as it is arrived in KStream from source.

Scenario is, If Matching Key is yet to arrive in any one of the stream, We should Stop or Retry join until expected key arrives in other topic. We thought to put unmatched records back to KStream but in this case sequence not guaranteed.

Issue 1: How to stop or hold join until the expected key to be arrived in other topic. Eg, KTable has Key 100, But KStream yet to receive Key 100 then We should retry Join or hold KStream until Key 100 arrives.

Issue 2: Is there any way to put Delay or Interval in KStream (Delayed KStream) to receive messages with delayed time or interval.

Additionally we have to build Keyed KStream from Non Keyed Topic (Key will be set by extracting it from Message - JSON)

Java is preferable as We done POC to Join between KTable and KStream

KTable<String, String> leftStream = builder.table("stream1");
KStream<String, String> rightStream = builder.stream("stream2");
KStream<String, String> outstream = rightStream.leftJoin(leftStream, (orig_msg, description) -> {
         String new_msg = "";
            if (description != null) {
                  new_msg = orig_msg+"-->Matched--"+description;
            }else {
                new_msg = orig_msg+"-->UnMatched<--"+description;
            }
                return new_msg;
     });

Michael G. Noll Michael G. Noll · Accepted Answer · 2019-05-09T11:16:46

You are doing a KStream-to-KTable LEFT join in your example. The Kafka Streams join semantics specify that (a) only data arriving in the KStream will trigger a join output and (b) if, when a new KStream event arrives, there is no matching data in the KTable (right side of join), then there will still be a join output being produced immediate but with null for the table-side data (i.e., there will be no waiting for data to arrive on the KTable side).

Issue 1: How to stop or hold join until the expected key to be arrived in other topic. Eg, KTable has Key 100, But KStream yet to receive Key 100 then We should retry Join or hold KStream until Key 100 arrives.

First, you cannot stop or hold the join with built-in Kafka Streams functionality.

Second, the specific example that you provided will not happen in practice because (see above) an event arriving in the KTable will not produce a join output. Only when an event arrives in the KStream will (a) a lookup into the KTable be made and (b) a join output produced, regardless of the outcome of (a).

But what may happen in a KStream-KTable LEFT join is that the opposite example happens: KStream has key 100, but the KTable has yet to receive key 100. How to handle this? See below.

Issue 2: Is there any way to put Delay or Interval in KStream (Delayed KStream) to receive messages with delayed time or interval.

Yes, there are ways to do that. But not with the existing join operations in the Kafka Streams DSL.

Instead, you can use the Processor API of Kafka Streams to implement the join semantics that you need with a few lines of code, and then plug this functionality into the DSL for easy re-use.

There is an example application that demonstrates this, coincidentally for a use case just like yours above: see the CustomStreamTableJoin at https://github.com/confluentinc/kafka-streams-examples (direct link to CustomStreamTableJoin example for Confluent v5.2.1 / Apache Kafka 2.2).

KStream join Retry / Delayed KStream

2 Answers