Kafka Streams with Spring Boot

@Configuration
@EnableKafkaStreams
@Slf4j
public class OrderStreamProcessor {

    @Bean
    public KStream<String, OrderEvent> orderPipeline(StreamsBuilder streamsBuilder) {

        // Spring Kafka 4.x name; earlier versions call it JsonSerde
        var orderSerde = new JacksonJsonSerde<>(OrderEvent.class);

        KStream<String, OrderEvent> orders = streamsBuilder
            .stream("orders.order.created",
                Consumed.with(Serdes.String(), orderSerde));

        // Filter + transform + route
        orders
            .filter((key, event) -> event.getAmount() > 1000)
            .mapValues(event -> EnrichedOrder.from(event, lookupService.getCustomer(event.getCustomerId())))
            .to("orders.high-value", Produced.with(Serdes.String(), new JacksonJsonSerde<>(EnrichedOrder.class)));

        // Aggregate: count orders per customer in 1-hour tumbling windows, no grace period
        orders
            .groupByKey()
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
            .count()
            .toStream()
            .map((windowedKey, count) ->
                KeyValue.pair(windowedKey.key(), new OrderCount(windowedKey.key(), count)))
            .to("analytics.order-counts", Produced.with(Serdes.String(), new JacksonJsonSerde<>(OrderCount.class)));

        return orders;
    }
}

spring:
  kafka:
    streams:
      application-id: order-stream-processor
      bootstrap-servers: localhost:9092
      state-dir: /var/lib/kafka-streams
      properties:
        "[default.key.serde]": org.apache.kafka.common.serialization.Serdes$StringSerde
        "[processing.guarantee]": exactly_once_v2
        "[num.stream.threads]": 4
        "[statestore.cache.max.bytes]": 10485760   # 10 MB; replaces the deprecated cache.max.bytes.buffering