[KAFKA-4720] Add KStream.peek(ForeachAction<K,V>)

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.10.1.1
• Fix Version/s: 0.11.0.0
• Component/s: streams
• Labels: needs-kip
• Assignee: Steven Schlansker
• Reporter: Steven Schlansker
• Created: 01/Feb/17 19:00
• Updated: 28/Feb/17 03:03
• Resolved: 16/Feb/17 05:19

Description

Java's Stream provides a handy peek method that observes elements in the stream without transforming or filtering them. While you can emulate this functionality with either a filter or map, peek provides potentially useful semantic information (doesn't modify the stream) and is much more concise.

Example usage: using Dropwizard Metrics to provide event counters

KStream<Integer, String> s = ...;
s.map(this::mungeData)
 .peek((i, s) -> metrics.noteMungedEvent(i, s))
 .filter(this::hadProcessingError)
 .print();

Issue Links

• relates to: KAFKA-4772 Exploit #peek to implement #print() and other methods
• links to: GitHub Pull Request #2493
• mentioned in: Page Loading...