kstream_snippet.java

private static void kStreamPipeline(Properties props) {
    System.out.println("Starting...");

    Properties kafkaStreamsProps = new Properties();
    kafkaStreamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, props.getProperty("APPLICATION_ID"));
    kafkaStreamsProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, props.getProperty("BOOTSTRAP_SERVERS"));
    kafkaStreamsProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, props.getProperty("AUTO_OFFSET_RESET_CONFIG"));
    kafkaStreamsProps.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, props.getProperty("COMMIT_INTERVAL_MS_CONFIG"));

    StreamsBuilder builder = new StreamsBuilder();
    builder
            .stream(props.getProperty("INPUT_TOPIC"), Consumed.with(Serdes.Void(), CustomSerdes.Purchase()))
            .peek((unused, purchase) -> System.out.println(purchase.toString()))
            .flatMap((KeyValueMapper<Void, Purchase, Iterable<KeyValue<String, Total>>>) (unused, purchase) -> {
                List<KeyValue<String, Total>> result = new ArrayList<>();
                result.add(new KeyValue<>(purchase.getProductId(), new Total(
                        purchase.getTransactionTime(),
                        purchase.getProductId(),
                        1,
                        purchase.getQuantity(),
                        purchase.getTotalPurchase()

                )));
                return result;
            })
            .groupByKey(Grouped.with(Serdes.String(), CustomSerdes.Total()))
            .reduce((total1, total2) -> {
                total2.setTransactions(total1.getTransactions() + total2.getTransactions());
                total2.setQuantities(total1.getQuantities() + total2.getQuantities());
                total2.setSales(total1.getSales().add(total2.getSales()));
                return total2;
            })
            .toStream()
            .peek((productId, total) -> System.out.println(total.toString()))
            .to(props.getProperty("OUTPUT_TOPIC"), Produced.with(Serdes.String(), CustomSerdes.Total()));

    KafkaStreams streams = new KafkaStreams(builder.build(), kafkaStreamsProps);
    streams.start();
    System.out.println("Running...");
}