A KStream represents an ongoing flow of Kafka records. Each record is an event, so several records with the same key can each matter. For example, a stream of authentication attempts can be monitored as attempts arrive rather than reduced immediately to one current value per user. section 12.5.3
Create a KStream from a topic with `builder.stream("topic")`. A KTable, created with `builder.table("topic")`, serves a different purpose: it holds the current value for each key. This distinction lets a pipeline react to events in a KStream while using a table when it needs current state. section 12.4.1
When a stream needs an aggregate, the steps matter:
- Group by the existing key with `groupByKey()`, or use `groupBy()` to select a new key. Selecting a new key creates a repartitioning topic. section 12.4.2
- Aggregate the grouped events into a KTable of current results. Use `toStream()` when subsequent processing needs those table updates as a KStream. section 12.4.2 section 12.4.1
Send notifications to a user's devices
Start with one KStream of notifications and another of device updates, both keyed by user ID. Group and aggregate the device updates into a KTable of active devices per user: add active devices and remove inactive ones. Join each notification with that current device list before producing device notifications. The streams carry incoming events; the table supplies the state the join needs. section 12.4.4
Flag repeated authentication attempts
Read `auth_requests` as a KStream, group by key, and count attempts in one-minute windows. Convert the windowed counts back to a stream, filter counts greater than three, remove the window from the output key, and write the results to `possible_frauds`. The filter acts on counts, not on individual input attempts. section 12.5.3
KStream<String, String> authRequests = builder.stream("auth_requests");
KTable<Windowed<String>, Long> windowedCounts = authRequests
.groupByKey() #1
.windowedBy(TimeWindows.ofSizeWithNoGrace(
Duration.ofMinutes(1))) #2
.count(); #3
KStream<String, Long> possibleFrauds = windowedCounts
.toStream()
.filter((windowedKey, count) -> count > 3) #4
.map((windowedKey, count) -> #5
KeyValue.pair(windowedKey.key(), count)); #5
possibleFrauds.to("possible_frauds"); #6
Follow the complete fraud-detection pipeline from a topic-backed KStream through grouping, windowed counting, `toStream()`, filtering, and output. section 12.5.3
