streams-api.java

final KStreamBuilder builder = new KStreamBuilder();

// map the json transaction stream to (id, Transaction) key-value pairs
final KStream<String, Transaction> trnxStream = builder.stream(stringSerde, jsonSerde, TOPIC_TRNX)
    .map((key, value) -> {
            Transaction tx = Transaction.newFromJson(value);
            return new KeyValue<>(tx.id, tx);
        });

// aggregate the transaction stream: S1
trnxStream.map((id, trnx) -> new KeyValue<>(trnx.tokenId, trnx.amount))
    .groupByKey(stringSerde, longSerde)
    .reduce((v1, v2) -> v1 + v2, "trnx-amount-by-token")
    .toStream()
    .to(stringSerde, longSerde, TOPIC_TRNX_AMOUNT_BY_TOKEN);