KStream.peek performs a stateless action on every record while leaving its incoming key and value unchanged. Use it to print, log, or collect information at an arbitrary point in a topology, then continue chaining stream operations. section 6.5
Use peek when you need visibility without terminating the stream:
- Place it before and after a transformation such as mapValues to compare records as they move through the topology. section 6.5
- Pass a ForeachAction, commonly a lambda receiving the record key and value, to perform the observation. section 6.5
- Prefer peek when subsequent operations must remain chainable; print is purpose-built for output but is terminal and returns void. section 6.5 section 6.7
- Convert a KTable to a KStream with toStream before using peek to inspect windowed aggregation results. section 9.1.6
Inspecting a transformation
ForeachAction<String, String> sysout = (key, value) -> System.out.println("key " + key + " value " + value); builder.stream("src-topic", Consumed.with(stringSerde, stringSerde)).peek(sysout).mapValues(value -> value.toUpperCase()).peek(sysout).to("out-topic", Produced.with(stringSerde, stringSerde)); The first peek observes the original records; the second observes the uppercased values, while the stream continues to out-topic. section 6.5
Listing 6.14 Printing records flowing through the Yelling app
ForeachAction<String, String> sysout =
(key, value) ->
System.out.println("key " + key
+ " value " + value);
builder.stream("src-topic",
Consumed.with(stringSerde, stringSerde))
.peek(sysout) #1
.mapValues(value -> value.toUpperCase())
.peek(sysout) #2
.to( "out-topic",
Produced.with(stringSerde, stringSerde));
Browse the complete chained example showing one peek before mapValues and another after it. section 6.5
