Stateful Kafka Streams with Spring Boot: Windows, Joins, Recovery, and PostgreSQL Views

Stateful Kafka Streams applications answer questions that cannot be solved by processing each record independently. They count events over time, correlate streams, enrich records from tables, build sessions, and maintain continuously updated views.

The state is not an incidental cache. It is part of the processing result. That means the design must explain:

which timestamp defines event time;
how records are partitioned;
where state is stored;
how state is restored after a failure;
when a window is considered complete;
what exactly-once processing does and does not cover;
how external systems such as PostgreSQL receive updates safely.

This guide builds that design with Spring Boot 4, Kafka Streams, local persistent state stores, changelog topics, and a separate PostgreSQL projection.

TL;DR Keep operational stream state inside Kafka Streams and let Kafka changelog topics restore it. Use event-time windows with an explicit grace period and preserve window boundaries in output records. Treat PostgreSQL as an eventually consistent projection unless it participates through a separate idempotent sink.

Choose Kafka Streams for the Right Workload

Kafka Streams is a client library embedded in an ordinary Java application. It is a strong fit when:

input and output are Kafka topics;
state can be partitioned by record key;
processing can be expressed as filters, maps, joins, aggregations, and windows;
local state plus Kafka changelogs provide an acceptable recovery model;
the application should scale by adding instances with the same application.id.

It is not automatically a full complex-event-processing engine. Kafka Streams can express many event-correlation patterns, but it does not provide a dedicated declarative pattern language for arbitrary sequences such as:

Event A
followed by Event B
unless Event C occurs
within 20 minutes

Those patterns can be implemented with the Processor API and state stores, but the application must own the matching, timeout, and cleanup logic.

Understand the Three Core Abstractions

KStream

A KStream models a sequence of independent facts.

customer-42 clicked product-7
customer-42 clicked product-9
customer-42 purchased product-9

A second record with the same key does not replace the first record.

KTable

A KTable models the latest value for each key.

customer-42 -> current customer profile
product-9   -> current product price

Its source topic should normally use compaction so Kafka can retain the latest value per key while allowing older versions to be removed according to compaction policy.

GlobalKTable

A GlobalKTable restores every source partition on every application instance. It is useful for relatively small reference datasets that every task needs locally.

It is not free:

every instance stores the entire table;
network and disk use grow with the number of instances;
restoration time grows with the table;
each instance must process every update.

Use an ordinary partitioned KTable when co-partitioning is practical. Use a GlobalKTable only when its replication cost is acceptable.

Project Setup with Spring Boot 4

Spring Boot auto-configures Kafka Streams when:

kafka-streams is on the classpath;
Kafka Streams is enabled;
bootstrap servers and an application ID are available.

A minimal Maven configuration is:

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>4.1.0</version>
    <relativePath />
</parent>

<properties>
    <java.version>25</java.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-kafka</artifactId>
    </dependency>

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

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

There is no separate spring-kafka-streams dependency. Spring for Apache Kafka provides the integration, while the Apache kafka-streams JAR must be present explicitly because it is optional.

Enable Streams lifecycle management:

@SpringBootApplication
@EnableKafkaStreams
public class ActivityAnalyticsApplication {

    public static void main(String[] args) {
        SpringApplication.run(
                ActivityAnalyticsApplication.class,
                args
        );
    }
}

Configure Identity, State, and Recovery

spring:
  application:
    name: activity-analytics

  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}

    streams:
      application-id: activity-analytics-v1
      state-dir: ${KAFKA_STREAMS_STATE_DIR:/var/lib/kafka-streams}
      properties:
        processing.guarantee: exactly_once_v2
        num.stream.threads: 2
        num.standby.replicas: 1
        replication.factor: 3
        commit.interval.ms: 1000

`application-id`

The application ID identifies one logical Streams application. Kafka uses it for:

consumer-group membership;
internal topic names;
local state-directory names;
client ID prefixes.

Changing it creates a new logical application that starts with separate offsets and separate internal topics. Do not change it merely to deploy a new application build.

`state-dir`

State is local to an application instance. In production, use storage with predictable capacity and performance.

Do not assume that a persistent volume eliminates the need for changelog recovery. Tasks can move to another instance during a rebalance, and that instance must restore the assigned stores.

Internal-topic replication

A replication factor of three requires at least three brokers. Local development with one broker must override it:

spring:
  kafka:
    streams:
      properties:
        replication.factor: 1
        num.standby.replicas: 0

Do not carry single-broker development settings into production.

Define Explicit Serdes

Avoid relying on one default JSON value Serde for an application that processes several record types. Explicit Serdes make each topology boundary easier to review.

public record CustomerActivity(
        String customerId,
        String action,
        Instant occurredAt
) {}

public record ActivityCount(
        String customerId,
        long windowStart,
        long windowEnd,
        long count
) {}

@Configuration
public class ActivitySerdes {

    @Bean
    JsonSerde<CustomerActivity> customerActivitySerde() {
        return new JsonSerde<>(CustomerActivity.class);
    }

    @Bean
    JsonSerde<ActivityCount> activityCountSerde() {
        return new JsonSerde<>(ActivityCount.class);
    }
}

In a larger system, use a schema-managed format such as Avro or Protobuf when independent producers and consumers need enforceable compatibility.

Event Time Must Be Deliberate

Windowed processing uses record timestamps. Depending on producer configuration and broker policy, the Kafka record timestamp may represent producer creation time or broker append time.

If the business event contains the authoritative timestamp, use a timestamp extractor:

public class CustomerActivityTimestampExtractor
        implements TimestampExtractor {

    @Override
    public long extract(
            ConsumerRecord<Object, Object> record,
            long partitionTime
    ) {
        if (record.value() instanceof CustomerActivity activity
                && activity.occurredAt() != null) {
            return activity.occurredAt().toEpochMilli();
        }

        if (partitionTime >= 0) {
            return partitionTime;
        }

        throw new IllegalArgumentException(
                "CustomerActivity has no valid event timestamp"
        );
    }
}

Do not silently substitute the current wall-clock time for malformed records. That moves bad records into the wrong window and hides data-quality problems.

Build a Persistent Windowed Count

The key must represent the entity whose state is aggregated. Do not call groupByKey() until the key contract is known.

@Configuration
public class ActivityTopology {

    @Bean
    KStream<String, CustomerActivity> activityStream(
            StreamsBuilder builder,
            JsonSerde<CustomerActivity> customerActivitySerde,
            JsonSerde<ActivityCount> activityCountSerde
    ) {
        KStream<String, CustomerActivity> source = builder.stream(
                "customer.activities",
                Consumed
                        .with(
                                Serdes.String(),
                                customerActivitySerde
                        )
                        .withTimestampExtractor(
                                new CustomerActivityTimestampExtractor()
                        )
        );

        KStream<String, CustomerActivity> keyed = source
                .filter((key, value) ->
                        value != null
                                && value.customerId() != null
                                && value.occurredAt() != null
                )
                .selectKey((key, value) -> value.customerId());

        TimeWindows windows = TimeWindows
                .ofSizeAndGrace(
                        Duration.ofMinutes(5),
                        Duration.ofMinutes(1)
                );

        KTable<Windowed<String>, Long> counts = keyed
                .groupByKey(
                        Grouped.with(
                                Serdes.String(),
                                customerActivitySerde
                        )
                )
                .windowedBy(windows)
                .count(
                        Materialized
                                .<String, Long, WindowStore<Bytes, byte[]>>
                                        as("customer-activity-counts")
                                .withKeySerde(Serdes.String())
                                .withValueSerde(Serdes.Long())
                );

        counts
                .toStream()
                .map((windowedKey, count) ->
                        KeyValue.pair(
                                windowedKey.key()
                                        + ":"
                                        + windowedKey.window().start(),
                                new ActivityCount(
                                        windowedKey.key(),
                                        windowedKey.window().start(),
                                        windowedKey.window().end(),
                                        count
                                )
                        )
                )
                .to(
                        "customer.activity-counts",
                        Produced.with(
                                Serdes.String(),
                                activityCountSerde
                        )
                );

        return source;
    }
}

The output preserves windowStart and windowEnd. Dropping the window and publishing only:

customer-42 -> 17

makes separate windows indistinguishable.

Persistent versus in-memory stores

The default persistent Kafka Streams store uses RocksDB when a stateful DSL operator materializes persistent state. An explicitly configured in-memory store does not survive process restart locally and must rebuild from its changelog.

In-memory stores can be useful for small state with acceptable restoration time, but they should not be described as RocksDB-backed persistence.

Window Size and Grace Period Are Different

A five-minute window answers:

Which event timestamps belong to this interval?

A one-minute grace period answers:

How long after the window end will the application still accept a record that belongs to it?

Kafka Streams evaluates lateness using stream time, which advances from observed record timestamps. It is not the same as waiting one wall-clock minute after every window.

The trade-off is explicit:

a short grace period reduces state retention and output delay;
a long grace period accepts more delayed records;
no grace rejects records that arrive after stream time passes the window end;
late records need an observability or recovery policy.

Monitor dropped-late-record metrics rather than treating late data as invisible.

Tumbling, Hopping, Sliding, and Session Windows

Tumbling windows

Fixed-size, non-overlapping windows:

TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofMinutes(1)
)

Hopping windows

Fixed-size windows that advance by a smaller interval:

TimeWindows
        .ofSizeAndGrace(
                Duration.ofMinutes(10),
                Duration.ofMinutes(2)
        )
        .advanceBy(Duration.ofMinutes(1))

One event can contribute to several hopping windows, increasing storage and output volume.

Sliding windows

Sliding windows are defined relative to record timestamps rather than aligned clock boundaries. They are useful for questions such as:

How many matching events occurred within ten minutes of this event?

Do not use hopping windows as a rough substitute without understanding the different semantics.

Session windows

Sessions group bursts of activity separated by inactivity.

SessionWindows.ofInactivityGapAndGrace(
        Duration.ofMinutes(30),
        Duration.ofMinutes(5)
)

Session windows can merge when a late event bridges two existing sessions. The aggregator therefore requires a merger.

A Correct Session Aggregation

public record UserEvent(
        String userId,
        String eventType,
        Instant occurredAt
) {}

public record SessionAccumulator(
        long firstEventAt,
        long lastEventAt,
        long eventCount
) {
    static SessionAccumulator first(long timestamp) {
        return new SessionAccumulator(
                timestamp,
                timestamp,
                1
        );
    }

    SessionAccumulator add(long timestamp) {
        return new SessionAccumulator(
                Math.min(firstEventAt, timestamp),
                Math.max(lastEventAt, timestamp),
                eventCount + 1
        );
    }

    SessionAccumulator merge(SessionAccumulator other) {
        return new SessionAccumulator(
                Math.min(firstEventAt, other.firstEventAt),
                Math.max(lastEventAt, other.lastEventAt),
                eventCount + other.eventCount
        );
    }
}

public record UserSession(
        String userId,
        long sessionStart,
        long sessionEnd,
        long eventCount,
        long activeDurationMillis
) {}

@Bean
KStream<String, UserEvent> sessionTopology(
        StreamsBuilder builder
) {
    JsonSerde<UserEvent> eventSerde =
            new JsonSerde<>(UserEvent.class);

    JsonSerde<SessionAccumulator> accumulatorSerde =
            new JsonSerde<>(SessionAccumulator.class);

    JsonSerde<UserSession> sessionSerde =
            new JsonSerde<>(UserSession.class);

    KStream<String, UserEvent> source = builder.stream(
            "user.events",
            Consumed.with(
                    Serdes.String(),
                    eventSerde
            )
    );

    KTable<Windowed<String>, SessionAccumulator> sessions = source
            .filter((key, event) ->
                    event != null
                            && event.userId() != null
                            && event.occurredAt() != null
            )
            .selectKey((key, event) -> event.userId())
            .groupByKey(
                    Grouped.with(
                            Serdes.String(),
                            eventSerde
                    )
            )
            .windowedBy(
                    SessionWindows.ofInactivityGapAndGrace(
                            Duration.ofMinutes(30),
                            Duration.ofMinutes(5)
                    )
            )
            .aggregate(
                    () -> new SessionAccumulator(
                            Long.MAX_VALUE,
                            Long.MIN_VALUE,
                            0
                    ),
                    (userId, event, aggregate) ->
                            aggregate.eventCount() == 0
                                    ? SessionAccumulator.first(
                                            event.occurredAt()
                                                    .toEpochMilli()
                                    )
                                    : aggregate.add(
                                            event.occurredAt()
                                                    .toEpochMilli()
                                    ),
                    (userId, left, right) ->
                            left.merge(right),
                    Materialized
                            .<String, SessionAccumulator,
                                    SessionStore<Bytes, byte[]>>
                                    as("user-sessions")
                            .withKeySerde(Serdes.String())
                            .withValueSerde(accumulatorSerde)
            );

    sessions
            .toStream()
            .map((windowedKey, aggregate) -> {
                long start = windowedKey.window().start();
                String outputKey =
                        windowedKey.key() + ":" + start;

                if (aggregate == null) {
                    // Preserve tombstones when two sessions merge so
                    // downstream materialized views can delete stale rows.
                    return KeyValue.<String, UserSession>pair(
                            outputKey,
                            null
                    );
                }

                long end = windowedKey.window().end();

                UserSession session = new UserSession(
                        windowedKey.key(),
                        start,
                        end,
                        aggregate.eventCount(),
                        Math.max(0, end - start)
                );

                return KeyValue.pair(outputKey, session);
            })
            .to(
                    "user.sessions",
                    Produced.with(
                            Serdes.String(),
                            sessionSerde
                    )
            );

    return source;
}

This topology emits updates while a session evolves. Session merges can also emit tombstones for superseded windows, so downstream materialized views must process those deletions instead of discarding null values. If consumers require only a final result, suppression can delay output until the window closes, but buffering must be bounded or carefully capacity-planned. An unbounded suppression buffer can exhaust local resources.

Join Only After Verifying Partitioning

A KStream-KTable join is local only when both sides are correctly co-partitioned.

Suppose order events arrive keyed by order ID, while customer profiles are keyed by customer ID. The order stream must be rekeyed before the join:

KStream<String, OrderEvent> ordersByCustomer = orders
        .selectKey((orderId, order) ->
                order.customerId()
        );

KTable<String, CustomerProfile> profiles = builder.table(
        "customer.profiles",
        Consumed.with(
                Serdes.String(),
                customerProfileSerde
        )
);

KStream<String, EnrichedOrder> enriched = ordersByCustomer.join(
        profiles,
        (order, profile) ->
                EnrichedOrder.from(order, profile)
);

Rekeying can create an internal repartition topic. That is expected, but it affects throughput, storage, and latency.

For co-partitioned joins, verify:

both topics use the same key type and serialization;
the same logical key produces identical bytes;
partition counts are compatible;
the partitioning strategy is compatible;
null keys are handled before the join.

Exactly-Once Semantics Have a Boundary

With:

processing.guarantee: exactly_once_v2

Kafka Streams can atomically coordinate:

consumed input offsets;
updates to Kafka Streams state and changelog topics;
records written to Kafka output topics.

That is a strong guarantee inside the Kafka transaction boundary.

It does not automatically include:

a PostgreSQL transaction;
an HTTP request;
an email provider;
an object-store write;
a custom database-backed state store;
arbitrary code executed outside the topology's Kafka transaction.

Do not claim end-to-end exactly-once behavior when a result is later written to PostgreSQL by a separate consumer. That projection is eventually consistent and should be idempotent.

Keep PostgreSQL Out of the Hot State-Store Path

Implementing Kafka Streams KeyValueStore directly on top of JPA or a remote PostgreSQL database is usually a poor default.

It changes a local state lookup into a network call and introduces:

database latency in every processor operation;
connection-pool constraints;
an additional failure domain;
difficult flush and transaction semantics;
mismatch with task migration and restoration;
no automatic inclusion in Kafka Streams exactly-once transactions.

Prefer this separation:

Kafka input topics
    -> Kafka Streams topology
    -> compacted result topic
    -> PostgreSQL projection consumer

Kafka Streams remains responsible for processing state. PostgreSQL becomes a query model for services that need SQL or centralized access.

Build an Idempotent PostgreSQL Projection

Publish absolute aggregate values rather than non-idempotent increments.

CREATE TABLE customer_activity_view (
    customer_id VARCHAR(200) NOT NULL,
    window_start TIMESTAMPTZ NOT NULL,
    window_end TIMESTAMPTZ NOT NULL,
    activity_count BIGINT NOT NULL,
    source_updated_at TIMESTAMPTZ NOT NULL,
    PRIMARY KEY (
        customer_id,
        window_start,
        window_end
    )
);

A projection consumer can use an upsert:

@Repository
public class ActivityViewRepository {

    private final JdbcTemplate jdbcTemplate;

    public ActivityViewRepository(
            JdbcTemplate jdbcTemplate
    ) {
        this.jdbcTemplate = jdbcTemplate;
    }

    public void upsert(ActivityCount value) {
        jdbcTemplate.update(
                """
                INSERT INTO customer_activity_view (
                    customer_id,
                    window_start,
                    window_end,
                    activity_count,
                    source_updated_at
                )
                VALUES (?, ?, ?, ?, NOW())
                ON CONFLICT (
                    customer_id,
                    window_start,
                    window_end
                )
                DO UPDATE SET
                    activity_count = EXCLUDED.activity_count,
                    source_updated_at = NOW()
                """,
                value.customerId(),
                Instant.ofEpochMilli(value.windowStart()),
                Instant.ofEpochMilli(value.windowEnd()),
                value.count()
        );
    }
}

@Service
public class ActivityProjectionListener {

    private final ActivityViewRepository repository;

    public ActivityProjectionListener(
            ActivityViewRepository repository
    ) {
        this.repository = repository;
    }

    @KafkaListener(
            topics = "customer.activity-counts",
            groupId = "activity-postgres-view-v1"
    )
    @Transactional
    public void updateView(ActivityCount count) {
        repository.upsert(count);
    }
}

If the database commits but the Kafka offset does not, the record is delivered again. The upsert writes the same absolute aggregate value, so the duplicate is harmless.

For more complex projections, keep a processed-event or source-version column and reject stale updates explicitly.

Query Local State Carefully

Kafka Streams interactive queries expose local stores. In a distributed application, one instance owns only part of the total state.

A complete query layer needs:

a named queryable state store;
application.server configured uniquely for each instance;
metadata lookup to find which instance owns a key;
an RPC or HTTP layer to forward remote queries;
handling for rebalances and temporarily unavailable stores.

Do not expose a REST endpoint that queries only the local store and describe it as the complete application state.

PostgreSQL may be the simpler query surface when:

many unrelated services need the full result set;
SQL filtering and joins are required;
slightly delayed projection updates are acceptable;
operating a distributed interactive-query RPC layer is not desirable.

Scaling and Recovery

Kafka Streams creates tasks from input partitions. The upper bound for useful parallelism is determined by the topology and partition counts, not by an arbitrary number of application instances.

Stream threads

Increasing num.stream.threads can add concurrency inside one process until partition and CPU limits are reached. More threads also increase memory, file handles, and local store activity.

Standby replicas

A standby replica is a warm copy of local state maintained on another Streams instance.

num.standby.replicas: 1

One standby requires at least two suitable instances and roughly doubles client-side state storage for the replicated stores. It reduces recovery time but does not remove rebalances.

State restoration

Recovery time depends on:

state size;
changelog retention and availability;
broker and network throughput;
local disk performance;
standby freshness;
number of tasks restored concurrently.

Track restoration progress and define a maximum acceptable recovery time.

Production State-Store Practices

Use fast local disks with sufficient capacity.
Monitor RocksDB and total state-directory size.
Keep changelog topics replicated.
Avoid ephemeral directories that disappear during ordinary container restarts unless restoration time is acceptable.
Do not share one state directory between concurrent application processes.
Use a stable application ID.
Plan upgrades and topology changes carefully.
Name stateful processors and stores when future topology evolution matters.
Test restoration from an empty local directory.

Deleting local state in production is not a routine fix. It forces restoration and may create a long processing pause.

Error Handling

A stateful topology needs policies for three failure classes.

Deserialization failures

Bad bytes cannot enter the typed topology. Decide whether to:

fail the application;
log and skip;
publish the original record to a dead-letter topic;
quarantine the source producer.

Skipping silently can make aggregate results incomplete.

Processing failures

An exception in a mapper, joiner, or aggregator can stop a Streams thread or be handled according to the configured processing exception policy. Continuing after an exception means the record may be omitted from state and output, so use it only with a clear data-quality policy.

Production failures

Output serialization or broker production failures need a separate policy. Exactly-once processing does not make an invalid output record serializable.

Monitor all three categories separately.

Testing the Topology

Use TopologyTestDriver for deterministic topology tests without starting a broker.

Test at least:

key selection;
window boundaries;
grace-period behavior;
out-of-order events;
null and malformed records;
session merging;
join behavior when table data is missing;
aggregate updates;
output key and Serde compatibility.

Use real Kafka integration tests for:

rebalances;
transactions;
state restoration;
standby promotion;
broker failures;
application restarts;
PostgreSQL projection redelivery.

A unit test cannot prove transaction and recovery behavior across actual processes.

Troubleshooting

No output appears

Check:

input topic names;
application.id;
source record keys;
explicit Serdes;
deserialization-handler metrics;
whether a window is still open;
whether output is suppressed;
Streams thread state;
ACLs for input, output, and internal topics.

A join returns no records

Check:

whether both sides use the same logical key;
whether the stream was rekeyed;
partition counts and partitioners;
whether the KTable has received its current value;
whether an inner join should be a left join;
null keys and tombstones.

State restoration takes too long

Check:

state size;
changelog topic replication;
broker throughput;
local disk speed;
standby configuration;
restoration metrics;
task assignment churn.

Window results keep changing

That is normal while accepted records continue to arrive. Confirm:

window size;
grace period;
source timestamps;
stream-time progression;
whether final-only suppression is required;
whether downstream consumers can process updates.

PostgreSQL contains duplicates or regressions

Check:

whether the sink writes absolute values or increments;
the projection's primary key;
upsert logic;
stale update detection;
listener exception handling;
source-key ordering;
whether different records represent the same business update.

Conclusion

Stateful stream processing is reliable when its boundaries are explicit.

For a Spring Boot and Kafka Streams application:

choose a stable application ID;
key records by the entity whose state is maintained;
use explicit Serdes;
define event time and grace periods deliberately;
preserve window identity in output;
keep operational state local and recoverable through Kafka changelogs;
use standby replicas when recovery objectives justify their storage cost;
understand that Kafka exactly-once semantics stop at the Kafka transaction boundary;
publish queryable results to a compacted topic;
update PostgreSQL through an idempotent projection;
test late data, rebalances, restoration, and duplicate delivery.

Kafka Streams does not remove distributed-systems trade-offs. It gives the application a coherent model for partitioned state, recovery, and transactional Kafka output—provided the design does not hide external side effects inside that model.

Choose Kafka Streams for the Right Workload

Understand the Three Core Abstractions

KStream

KTable

GlobalKTable

Project Setup with Spring Boot 4

Configure Identity, State, and Recovery

application-id

state-dir

Internal-topic replication

Define Explicit Serdes

Event Time Must Be Deliberate

Build a Persistent Windowed Count

Persistent versus in-memory stores

Window Size and Grace Period Are Different

Tumbling, Hopping, Sliding, and Session Windows

Tumbling windows

Hopping windows

Sliding windows

Session windows

A Correct Session Aggregation

Join Only After Verifying Partitioning

Exactly-Once Semantics Have a Boundary

Keep PostgreSQL Out of the Hot State-Store Path

Build an Idempotent PostgreSQL Projection

Query Local State Carefully

Scaling and Recovery

Stream threads

Standby replicas

State restoration

Production State-Store Practices

Error Handling

Deserialization failures

Processing failures

Production failures

Testing the Topology

Troubleshooting

No output appears

A join returns no records

State restoration takes too long

Window results keep changing

PostgreSQL contains duplicates or regressions

Conclusion

Official References

`application-id`

`state-dir`