How to compute the sum of a field with Kafka Streams

An aggregation in Kafka Streams is a stateful operation used to perform a "clustering" or "grouping" of values with the same key. An aggregation in Kafka Streams may return a different type than the input value. In our example here we're going to use the reduce method to sum the total amount of tickets sold by title.

 builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), ticketSaleSerde))
        .map((k, v) -> KeyValue.pair(v.title(), v.ticketTotalValue()))
        .groupByKey(Grouped.with(Serdes.String(), Serdes.Integer()))
        .reduce(Integer::sum)
        .toStream()
        .mapValues(v -> String.format("%d total sales",v))
        .to(OUTPUT_TOPIC, Produced.with(Serdes.String(), Serdes.String()));

Let's review the key points in this example

    map((key, value) -> new KeyValue<>(v.title(), v.ticketTotalValue()))

Aggregations must group records by key. Since the stream source topic doesn't define any, the code has a map operation which creates new key-value pairs setting the key of the stream to the TicketSale.title field.

        groupByKey(Grouped.with(Serdes.String(), Serdes.Integer()))

Since you've changed the key, under the covers Kafka Streams performs a repartition immediately before it performs the grouping.
Repartitioning is simply producing records to an internal topic and consuming them back into the application. By producing the records the updated keys land on the correct partition. Additionally, since the key-value types have changed you need to provide updated Serde objects, via the Grouped configuration object to Kafka Streams for the (de)serialization process for the repartitioning.

 .reduce(Integer::sum)

The Reduce operator is a special type of aggregation. A reduce returns the same type as the original input, in this case a sum of the current value with the previously computed value. The reduce method takes an instance of a Reducer. Since a Reducer is a single method interface you can use method handle instead of a concrete object. In this case it's Integer.sum method that takes two integers and adds them together.

                .toStream()
                .mapValues(v -> String.format("%d total sales",v))
                .to(OUTPUT_TOPIC, Produced.with(Serdes.String(), Serdes.String()));

Aggregations in Kafka Streams return a KTableinstance, so it's converted to a KStream then mapValues appends a string to the count to give it some context on the meaning of the number.