Kafka Streams with Spring Boot

@Configuration
@EnableKafkaStreams
@Slf4j
public class OrderStreamProcessor {

    @Autowired
    public KStream<String, OrderEvent> buildOrderPipeline(StreamsBuilder streamsBuilder) {

        // Deserialize
        var orderSerde = new JsonSerde<>(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 JsonSerde<>(EnrichedOrder.class)));

        // Aggregate: count orders per customer (tumbling window, 1 hour)
        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 JsonSerde<>(OrderCount.class)));

        return orders;
    }
}

spring:
  kafka:
    streams:
      application-id: order-stream-processor
      bootstrap-servers: localhost:9092
      default-key-serde: org.apache.kafka.common.serialization.Serdes$StringSerde
      properties:
        processing.guarantee: exactly_once_v2
        num.stream.threads: 4
        cache.max.bytes.buffering: 10485760  # 10MB state cache
        state.dir: /tmp/kafka-streams