Kafka Streams with Spring Boot: Real-Time Aggregations

Kafka Streams lets you transform and aggregate event data inside your application without a separate cluster. Spring Boot wires it up with a few properties and a @Bean that defines your topology.

Dependencies and Configuration

Add the Kafka Streams starter alongside spring-kafka:

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-streams</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>

spring:
  kafka:
    bootstrap-servers: localhost:9092
    streams:
      application-id: order-analytics
      properties:
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.value.serde: org.springframework.kafka.support.serializer.JsonSerde

Windowed Order Counts

Count orders per minute keyed by region:

@Configuration
public class OrderStreamsConfig {
 
  @Bean
  public KStream<String, OrderEvent> orderStream(StreamsBuilder builder) {
    KStream<String, OrderEvent> orders = builder.stream("orders");
 
    orders
        .groupBy((key, event) -> event.region(), Grouped.with(Serdes.String(), new JsonSerde<>(OrderEvent.class)))
        .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
        .count()
        .toStream()
        .map((windowedKey, count) ->
            KeyValue.pair(windowedKey.key(), new RegionCount(windowedKey.key(), count)))
        .to("order-counts-by-region");
 
    return orders;
  }
}

Interactive Queries

Expose running aggregates via a REST endpoint using KafkaStreams:

@RestController
@RequestMapping("/analytics")
public class OrderAnalyticsController {
 
  private final KafkaStreams kafkaStreams;
 
  public OrderAnalyticsController(KafkaStreams kafkaStreams) {
    this.kafkaStreams = kafkaStreams;
  }
 
  @GetMapping("/regions/{region}/count")
  public long countForRegion(@PathVariable String region) {
    ReadOnlyWindowStore<String, Long> store = kafkaStreams.store(
        StoreQueryParameters.fromNameAndType("order-counts-by-region-store", QueryableStoreTypes.windowStore()));
 
    Instant now = Instant.now();
    try (WindowStoreIterator<Long> iter = store.fetch(region, now.minus(Duration.ofMinutes(1)), now)) {
      return iter.hasNext() ? iter.next().value : 0L;
    }
  }
}

Operational Tips

Set processing.guarantee to exactly_once_v2 when you need end-to-end semantics across producers and streams.
Size state stores with rocksdb.total.offheap.budget.mb if you run large windowed aggregations.
Use a dedicated application-id per environment so dev and prod never share consumer group state.