Building Stream Processing Apps with Spring Boot & Kafka Streams: State Stores, Windowing & Exactly-Once

1. Core Concepts and Topology Design

Kafka Streams revolves around two core abstractions: KStream (a record stream where each record is processed independently) and KTable (a changelog stream that retains only the latest value per key, similar to a database table). The processing topology is a directed acyclic graph (DAG) composed of source processors, stream processors, and sink processors. While the DSL (e.g., stream().filter().map().to()) is convenient for simple logic, complex topologies often require the lower-level Processor API for debuggability.

2. Spring Boot Integration: Customizing Auto-Configuration

Spring Boot simplifies Kafka Streams setup via @EnableKafkaStreams, but production environments demand explicit configuration. The following example shows a custom configuration bean that sets the application ID, bootstrap servers, enables Exactly-Once v2, defines a dedicated state directory on a separate data disk, configures a dead-letter topic handler for deserialization failures, and registers an uncaught exception handler to prevent silent thread deaths.

@Configuration
@EnableKafkaStreams
public class KafkaStreamsConfig {
    @Bean
    public StreamsBuilderFactoryBeanCustomizer streamsCustomizer() {
        return factory -> {
            // Custom config and uncaught exception handler to avoid silent thread crashes
            factory.setStreamsConfiguration(customStreamsConfig());
            factory.setUncaughtExceptionHandler((thread, exception) -> {
                log.error("Stream thread {} crashed", thread.getName(), exception);
                // Trigger alert or call KafkaStreams.close() to restart app
            });
        };
    }

    @Bean
    public KafkaStreamsConfiguration customStreamsConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-stream-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // Production: always use EOS v2; legacy exactly_once blows up broker connections
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);

        // State dir must not reside on system disk; mount a dedicated data volume
        props.put(StreamsConfig.STATE_DIR_CONFIG, "/data/kafka-streams");

        // Dead-letter handler for deserialization errors (requires spring-kafka dependency)
        props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
                SendToDeadLetterTopicExceptionHandler.class);

        return new KafkaStreamsConfiguration(props);
    }
}

Spring Boot manages the lifecycle of StreamsBuilder and KafkaStreams, initializing the topology on startup and gracefully committing offsets and flushing state on shutdown.

3. State Storage with RocksDB: Performance and Pitfalls

Kafka Streams uses RocksDB for local state storage, delivering high-speed reads and writes. However, improper tuning can exhaust memory and disk.

State Recovery and Standby Replicas

Each state store is backed by an internal changelog topic . On node restart, state is rebuilt by replaying the changelog. A critical production practice is to configure standby replicas :

props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);

Without standbys, a failed node forces the replacement to replay the entire changelog, causing minute-scale recovery. With standbys, other nodes maintain asynchronous read-only replicas, enabling millisecond-level failover.

4. Window Aggregation: Disable Grace Period for Stability

Kafka Streams supports three window types:

Tumbling windows – fixed-size, non-overlapping (e.g., hourly counts).

Hopping windows – fixed-size, overlapping (e.g., 5-minute window advancing every minute for moving averages).

Session windows – dynamic size based on activity gaps (e.g., user session duration with 30-minute inactivity gap).

Example DSL snippets:

// 1. Tumbling window: fixed size, no overlap
KTable<Windowed<String>, Long> hourlyClicks = clickStream
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
    .count(Materialized.as("hourly-clicks-store"));

// 2. Hopping window: fixed size, overlapping
KTable<Windowed<String>, Long> movingAvg = eventStream
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)).advanceBy(Duration.ofMinutes(1)))
    .count();

// 3. Session window: dynamic size based on inactivity gap
KTable<Windowed<String>, Long> sessionDuration = userActivityStream
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
    .count();

Key recommendation: Use WithNoGrace to disable the default 24-hour grace period. The grace period retains late-arriving records but can cause RocksDB to accumulate massive amounts of expired state, leading to disk exhaustion. If the business can tolerate minimal late-data loss, turning off grace period is essential.

5. Stream-Table Joins: Unidirectional Trigger and Serialization Traps

Joining a KStream with a KTable (e.g., enriching order events with user details) is common. Two major pitfalls exist:

Unidirectional triggering: The join fires only when a new record arrives on the KStream side; updates to the KTable do not proactively push results to the stream.

Serialization completeness: The Joined.with method requires explicit Serdes for both key and value on both sides. Passing null for the right-side value Serde causes runtime serialization failures.

// Build user detail KTable
KTable<String, UserDetail> userTable = builder.table(
    "user-details-topic",
    Consumed.with(Serdes.String(), userDetailSerde)
);

// Order stream left-joins user table
KStream<String, EnrichedOrder> enrichedOrders = orderStream
    .leftJoin(
        userTable,
        (order, user) -> new EnrichedOrder(order, user),
        // Must provide Serdes for both sides; never leave right value Serde as null
        Joined.with(Serdes.String(), orderSerde, userDetailSerde)
    );

6. Exactly-Once Semantics: Transactional Guarantees

Exactly-Once (EOS) is implemented via Kafka transactions. The legacy exactly_once mode created a transactional producer per task, exploding broker connections. Always use exactly_once_v2 , which shares a single transactional producer across all tasks, drastically reducing resource consumption.

The transaction bundles four steps atomically: read input records, update local state, write output records, commit offsets. On crash recovery, processing resumes from the last committed offset, and state is restored from the changelog, guaranteeing no duplicate processing.

7. Interactive Queries (IQ): Cache Consistency Trap

IQ exposes the local state store via REST, turning the Streams application into a distributed key-value store. Example controller:

@RestController
@RequestMapping("/api/state")
public class StateQueryController {
    private final KafkaStreams kafkaStreams;

    public StateQueryController(KafkaStreams kafkaStreams) {
        this.kafkaStreams = kafkaStreams;
    }

    @GetMapping("/user/{userId}/score")
    public ResponseEntity<Long> getUserScore(@PathVariable String userId) {
        ReadOnlyKeyValueStore<String, Long> store = kafkaStreams.store(
            StoreQueryParameters.fromNameAndType("user-score-store", QueryableStoreTypes.keyValueStore())
        );
        Long score = store.get(userId);
        return score != null ? ResponseEntity.ok(score) : ResponseEntity.notFound().build();
    }
}

Major pitfall: Kafka Streams enables an in-memory cache by default. Recently written data may reside only in the cache and not yet flushed to RocksDB, causing IQ reads to return stale or missing data. Solutions:

Disable cache (severe performance penalty, not recommended).

Accept eventual consistency (recommended; tolerate seconds of delay).

In distributed deployments, implement custom routing using StreamsMetadata to forward queries to the node owning the target key.

8. Topology Optimization: Immutable DTOs and Custom Serialization

Use Lombok's @Value to create immutable DTOs, preventing accidental state mutation in stream processing.

@Value
@Builder
public class OrderAggregate {
    String orderId;
    BigDecimal totalAmount;
    int itemCount;
}

For complex logic beyond DSL capabilities (scheduled tasks, multi-store interactions), switch to the Processor API. Regarding serialization: JSON is developer-friendly but Protobuf offers superior performance and smaller payloads; adopt Protobuf for high-volume production workloads.

9. Multi-Tenancy Isolation: Avoiding Noisy Neighbors

In SaaS scenarios, simple topic prefixing (e.g., tenantA.orders) works until a large tenant monopolizes resources. Physical isolation via dedicated application.id per tenant ensures separate consumer groups and state stores:

props.put(StreamsConfig.APPLICATION_ID_CONFIG, "stream-app-" + tenantId);

Resource quotas (CPU, memory) must be enforced at the Kubernetes level using Limits and Requests per pod.

10. Production Operations: Reset, Monitoring, and Exception Handling

Application Reset

To reprocess historical data or recover from logic errors, use the reset tool after stopping the application :

kafka-streams-application-reset.sh --application-id order-stream-app \
  --bootstrap-server localhost:9092 --input-topics orders --to-earliest

Monitoring Metrics

Expose JMX metrics via Micrometer to Prometheus. Watch task-closed-rate closely; spikes indicate rebalance storms caused by frequent node failures or slow processing.

Exception Handling

Production environments must handle malformed data and broker errors gracefully:

// 1. Deserialization errors: route to dead-letter topic instead of crashing
props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
        SendToDeadLetterTopicExceptionHandler.class);

// 2. Production errors (e.g., broker rejects write): continue processing (Kafka 2.8+)
props.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
        ContinueOnProductionExceptionHandler.class);

Conclusion

Kafka Streams is a pragmatic, lightweight stream processing framework within the Java ecosystem. It handles 80% of real-time computing needs without the operational overhead of Flink. However, it has sharp edges: slow rebalances, state store tuning complexity, and IQ cache latency. Treat it as a powerful tool, not a silver bullet, and apply the practices above to run it reliably in production.