[KAFKA-12943] Bug in Kafka Streams Documentation (Aggregating)

• Type: Bug
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Fix Version/s: 3.3.0
• Component/s: documentation, streams
• Assignee: Marco Lotz
• Reporter: Rajesh KSV
• Created: 12/Jun/21 02:21
• Updated: 05/Jul/23 14:51
• Resolved: 07/Jul/22 21:00

Description

In the doc, for aggregating function, the example is incorrect

https://kafka.apache.org/documentation/streams/developer-guide/dsl-api.html#aggregating

It says

KTable<byte[], Long> aggregatedStream = groupedStream.aggregate(
    () -> 0L, /* initializer */
    (aggKey, newValue, aggValue) -> aggValue + newValue.length(), /* adder */
    Materialized.as("aggregated-stream-store") /* state store name */
        .withValueSerde(Serdes.Long()); /* serde for aggregate value */

Generic types are missing. Instead, it should be 

KTable<byte[], Long> aggregatedStream = groupedStream.aggregate(
    () -> 0L, /* initializer */
    (aggKey, newValue, aggValue) -> aggValue + newValue.length(), /* adder */
    Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("aggregated-stream-store") /* state store name */
        .withValueSerde(Serdes.Long()); /* serde for aggregate value */ 

Otherwise, code won't work. I myself verified it. 

Reference

https://stackoverflow.com/questions/51040555/the-method-withvalueserde-in-the-type-materialized-is-not-applicable/51049472

Issue Links

• links to: GitHub Pull Request #12091