KStream.merge combines multiple KStream objects into one stream. It is useful after branching, when independently processed streams—such as legacy and updated IoT sensor readings—must rejoin the topology. section 6.6.2
Use the operator with these ordering expectations:
- Do not rely on ordering between input streams; records from the legacy and updated streams may interleave unpredictably. section 6.6.2
- Preserve each input stream’s relative order while processing the merged result. section 6.6.2
- Obtain branched streams with Branched.withFunction when they must be merged later; Branched.withConsumer is suitable when the branch is written directly to a topic. section 6.6.2
Rejoining sensor streams
After splitting a combined sensor stream and converting proximity readings to meters, merge each branch with its corresponding existing stream:\n\nKStream<String, Sensor> temperatureSensorStream =\n builder.stream("temperature-sensors", sensorConsumed);\n\nKStream<String, Sensor> proximitySensorStream =\n builder.stream("proximity-sensors", sensorConsumed);\n\ntemperatureSensorStream\n .merge(sensorMap.get("sensor-temperature"))\n .to("temp-reading", Produced.with(stringSerde, sensorSerde));\n\nproximitySensorStream\n .merge(sensorMap.get("sensor-proximity"))\n .to("proximity-reading", Produced.with(stringSerde, sensorSerde)); section 6.6.2
Listing 6.19 Splitting the stream and gaining access to the newly created streams
KStream<String, Sensor> temperatureSensorStream = #1
builder.stream("temperature-sensors", sensorConsumed);
KStream<String, Sensor> proximitySensorStream = #2
builder.stream("proximity-sensors", sensorConsumed);
temperatureSensorStream.merge(sensorMap.get("sensor-temperature"))
.to("temp-reading", Produced.with(stringSerde, sensorSerde)); #3
proximitySensorStream.merge(sensorMap.get("sensor-proximity"))
.to("proximity-reading", Produced.with(stringSerde, sensorSerde)); #4
Trace the two merge calls that reunite temperature and proximity streams with their corresponding branched streams before producing topic records. section 6.6.2
