The join() methods in Spring for Apache Kafka Streams are inner joins, used to join messages with the same key published to two Apache Kafka topics.
For example, if you have two topics, customers and orders, to join messages with the same key in these two topics, you can create two new KStream objects and use the join() method to join these messages within a certain time interval, as follows:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
@Bean public KStream<String, String> customers(StreamsBuilder builder) { KStream<String, String> customers = builder.stream("customers"); KStream<String, String> orders = builder.stream("orders"); customers .join( orders, (ValueJoiner<String, String, String>) (customer, order) -> String.format("Customer %s ordered %s", customer, order), JoinWindows.ofTimeDifferenceAndGrace(Duration.ofSeconds(25), Duration.ofSeconds(150)), StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String())) .to("enriched-orders"); return customers; } |
- The first parameter is the name of the topic we want to join.
- The second parameter is the output we want when we receive the values of the message keys in the two topics.
- The third parameter is the time interval during which these messages are public. You can use two static methods of the
JoinWindowsclass,ofTimeDifferenceAndGrace()andofTimeDifferenceWithNoGrace(), to configure this join windows. TheofTimeDifferenceAndGrace()method defines the time interval during which messages created and received within this timeframe, plus a delay, will be joined. TheofTimeDifferenceWithNoGrace()method will only join messages created and received within this timeframe. - The fourth parameter defines the data type for serialization and deserialization of those messages.
Run this example and publish a message with key 001 and value “Khanh” to the topic customers, and another message with key 001 and value “Table” to the topic orders, and you will see a message published in the topic enriched-orders with the following content:
You can join:
- KStream with KStream
- KStream with KTable
- KStream with GlobalKTable
- KTable with KTable
If you join with KTable or GlobalKTable, you don’t need to specify a time period.
You can watch the video here.
