How to Implement Session Windows for User Activity Tracking

Session windows group temporally related events into user sessions by adapting to activity patterns rather than fixed clock boundaries. When a new event arrives within the configured inactivity gap of an existing session, the window extends; when no event arrives for longer than the gap, the session closes. That makes session windows the natural fit for user-journey and web-session analytics — and the most state-intensive windowing pattern Kafka offers, because an open session has no upper bound on duration. This guide covers implementation in Kafka Streams and ksqlDB, plus the state sizing, RocksDB tuning, metrics, and late-data handling you need to run them in production.

Understanding Session Windows in Stream Processing

A session window is defined by a single parameter: the inactivity gap. Events for the same key that fall within the gap belong to the same session; a stretch of silence longer than the gap ends it. Because session size is determined by the data, not the clock, session windows behave differently from the fixed-grid windows.

In the broader context of Stream Processing, the window types contrast as follows:

  • Tumbling windows: fixed-size, non-overlapping buckets.
  • Hopping windows: fixed-size, overlapping buckets.
  • Session windows: variable-size buckets whose boundaries are derived from the data.

This dynamic nature creates two operational consequences. First, state must be retained for every active session, and an active session has no fixed upper bound on duration — a bot that polls every 29 minutes against a 30-minute gap can hold one session open for hours. Second, a late event whose timestamp falls between two existing sessions can cause them to merge into one larger session, which retracts the two old results and emits a new combined one.

The inactivity gap directly drives both state-store size and result latency. A shorter gap yields more granular sessions but more windows and more frequent updates; a longer gap reduces churn but risks merging genuinely distinct visits. Pick the gap from product behavior, not convenience: a 30-minute gap is the web-analytics convention, but a checkout funnel may want 5 minutes and a video-watch session an hour. For teams already using ksqlDB aggregations, see Windowed Aggregations and Joins in ksqlDB.

Implementing Session Windows with Kafka Streams DSL

The Kafka Streams DSL expresses session windowing declaratively through SessionWindows. Since KIP-633, the grace period is no longer a hidden 24-hour default: you choose explicitly between ofInactivityGapWithNoGrace and ofInactivityGapAndGrace.

Basic Session Window Aggregation

The following counts page views per user session with a 30-minute inactivity gap and no grace period:

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.SessionStore;
import java.time.Duration;

StreamsBuilder builder = new StreamsBuilder();

KStream<String, PageView> views = builder.stream(
    "page-views",
    Consumed.with(Serdes.String(), pageViewSerde)
);

// The output key is Windowed<String>, so it needs a session-windowed key serde.
Serde<Windowed<String>> windowedKeySerde =
    WindowedSerdes.sessionWindowedSerdeFrom(String.class);

KTable<Windowed<String>, Long> sessionCounts = views
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
    .count(Materialized.<String, Long, SessionStore<Bytes, byte[]>>as("session-counts-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long()));

sessionCounts.toStream()
    .to("session-counts-output", Produced.with(windowedKeySerde, Serdes.Long()));

The aggregation materializes into a SessionStore named session-counts-store. ofInactivityGapWithNoGrace sets the grace period to zero, so any out-of-order record arriving after a session has ended is dropped (and counted under the dropped-records metric — watch it). Note the output key type: a session-windowed aggregation emits Windowed<String> keys, so the topic is written with WindowedSerdes.sessionWindowedSerdeFrom(String.class), not a plain Serdes.String(). A downstream consumer that reads it with a plain string serde will silently get garbage keys.

Production Configuration with a Grace Period

Most production pipelines need a grace period so that modestly out-of-order events still land in the right session:

KTable<Windowed<String>, Long> sessionCounts = views
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapAndGrace(
        Duration.ofMinutes(30),   // inactivity gap
        Duration.ofMinutes(15)    // grace period (afterWindowEnd)
    ))
    .count(Materialized.<String, Long, SessionStore<Bytes, byte[]>>as("session-counts-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long())
        .withCachingDisabled());

The 15-minute grace period accepts events arriving up to 15 minutes after a session would otherwise have ended, allowing them to extend or merge sessions. Once the grace period elapses, the window is final. withCachingDisabled() turns off the record cache so every update is forwarded immediately — useful when you want per-update visibility, at the cost of higher downstream volume. (To instead emit only final results, see suppression below.)

You do not need to set cleanup.policy on the changelog manually: changelog topics backing windowed and session stores are created with cleanup.policy=compact,delete by default, and retention is derived from the window’s retention period.

Implementing Session Windows in ksqlDB

ksqlDB exposes session windows through a WINDOW SESSION clause. The syntax is compact, but the same merge and grace-period semantics apply underneath.

Creating a Session Windowed Aggregation

CREATE TABLE user_session_counts AS
SELECT
    user_id,
    COUNT(*) AS action_count,
    COLLECT_LIST(action_type) AS action_sequence,
    WINDOWSTART AS session_start,
    WINDOWEND AS session_end
FROM user_actions
WINDOW SESSION (30 MINUTES)
GROUP BY user_id
EMIT CHANGES;

COLLECT_LIST accumulates the ordered actions within each session, and the WINDOWSTART/WINDOWEND pseudo columns expose the session boundaries (as Unix-time BIGINT values). EMIT CHANGES produces a continuously updating result as sessions evolve. Be deliberate with COLLECT_LIST on session windows: an unbounded session multiplies its per-key state by every action it collects, so a runaway session and an unbounded list compound into the same memory problem.

Adding a Grace Period and Sink Settings

The GRACE PERIOD clause is part of the window expression, and the WITH clause controls the output topic:

CREATE TABLE user_session_counts
WITH (
    kafka_topic = 'user_session_counts_topic',
    value_format = 'JSON',
    partitions = 12,
    replicas = 3
) AS
SELECT
    user_id,
    COUNT(*) AS action_count,
    WINDOWSTART AS session_start,
    WINDOWEND AS session_end
FROM user_actions
WINDOW SESSION (30 MINUTES, GRACE PERIOD 10 MINUTES)
GROUP BY user_id
EMIT CHANGES;

GRACE PERIOD 10 MINUTES keeps each session open to late events for 10 minutes past the inactivity gap. Without an explicit grace period, ksqlDB applies a 24-hour default, which retains far more state than session analytics usually needs — set it deliberately. For related patterns, see Windowed Aggregations and Joins in ksqlDB.

State Management and Operational Tuning

Session aggregations are stateful: every active session lives in a RocksDB-backed SessionStore and in its changelog topic until it closes and its retention elapses.

Sizing State Stores

State size is driven by the number of concurrently active sessions and the bytes each one accumulates:

State Size ≈ Active Sessions × (Key Size + Accumulated Value Size)

The active-session count is bounded by how many distinct keys can be “in flight” within one inactivity gap plus grace period, not by raw event rate. If 3 million users are active in any 45-minute window (30-minute gap + 15-minute grace) and each session holds roughly 2 KB of accumulated state, the store approaches 6 GB per replica before compression — and grows with key cardinality, not throughput. Two decision rules follow: a count() keeps only a long (single-digit bytes per session), whereas a COLLECT_LIST-style aggregation can balloon per-session bytes without bound; and that 6 GB lives on each instance that owns a copy, so a standby replica doubles it. Always size against peak concurrent keys, then validate empirically — RocksDB compression (LZ4 below) typically recovers a meaningful fraction, but the uncompressed in-flight estimate is the number you must fit on disk.

Tuning RocksDB

RocksDB tuning in Kafka Streams is not done with rocksdb.* properties — those keys do not exist. The only RocksDB-related Streams config is rocksdb.config.setter, which names a class you implement against the org.apache.kafka.streams.state.RocksDBConfigSetter interface. All block-cache, write-buffer, and compression settings are applied programmatically inside that class:

import org.apache.kafka.streams.state.RocksDBConfigSetter;
import org.rocksdb.BlockBasedTableConfig;
import org.rocksdb.CompressionType;
import org.rocksdb.LRUCache;
import org.rocksdb.Options;
import java.util.Map;

public class SessionRocksDBConfig implements RocksDBConfigSetter {
    // Share one cache across all RocksDB instances so the budget is bounded.
    private static final org.rocksdb.Cache CACHE = new LRUCache(256 * 1024 * 1024L); // 256 MB

    @Override
    public void setConfig(String storeName, Options options, Map<String, Object> configs) {
        // Reuse the existing table config so defaults are preserved.
        BlockBasedTableConfig tableConfig =
            (BlockBasedTableConfig) options.tableFormatConfig();
        tableConfig.setBlockCache(CACHE);
        options.setTableFormatConfig(tableConfig);

        options.setCompressionType(CompressionType.LZ4_COMPRESSION);
        options.setWriteBufferSize(64 * 1024 * 1024L); // 64 MB
        options.setMaxOpenFiles(-1);                   // -1 = unlimited
    }

    @Override
    public void close(String storeName, Options options) {
        // Do not close CACHE here: it is shared across stores.
    }
}

Register it in your Streams configuration:

rocksdb.config.setter=com.example.SessionRocksDBConfig

Because a SessionStore is backed by multiple RocksDB instances, sharing a single LRUCache (as above) is what actually bounds total off-heap memory — a per-instance cache silently multiplies your budget by the number of segments. The shared cache has one reporting side effect worth knowing before you tune: when all instances share one cache, Kafka Streams records each block-cache metric from a single instance rather than summing across them, so a shared-cache hit ratio is representative, not additive.

Monitoring Session State

RocksDB state-store metrics live under the stream-state-metrics group, and the store-scope token for a session store is rocksdb-session-state:

kafka.streams:type=stream-state-metrics,thread-id=*,task-id=*,rocksdb-session-state-id=*

Useful members of this group include block-cache-data-hit-ratio, bytes-written-rate, and bytes-read-rate. Thread-level throughput and latency live under a separate group:

kafka.streams:type=stream-thread-metrics,thread-id=*

with members such as process-rate, process-latency-avg, and poll-rate.

One critical gotcha: RocksDB metrics are only recorded at DEBUG level. With the default metrics.recording.level=INFO, the stream-state-metrics RocksDB members are not exposed at all. Set:

metrics.recording.level=DEBUG

Expose JMX with the Prometheus JMX exporter (the Confluent images default to port 7071):

scrape_configs:
  - job_name: 'kafka-streams'
    static_configs:
      - targets: ['streams-host:7071']

Alert on block-cache hit ratio dropping below ~0.90 (cache too small for the working set), state-store disk usage exceeding 80% of capacity, and rising process-latency-avg.

Handling Late Events and Session Merging

A late event whose timestamp falls within the inactivity gap of two existing sessions causes those sessions to merge: the two prior results are retracted (emitted as tombstones / null-value updates) and replaced by a single combined session. Downstream consumers must be prepared for this retract-and-replace behavior — it is the single most common reason session-windowed pipelines produce “duplicate then corrected” counts.

Grace Period and Suppression

During the grace period, merges and extensions can generate a high volume of intermediate updates. To emit only the finalized result per session, use suppression:

import org.apache.kafka.streams.kstream.Suppressed;

KTable<Windowed<String>, Long> suppressedCounts = sessionCounts
    .suppress(Suppressed.untilWindowCloses(
        Suppressed.BufferConfig.unbounded()
    ));

untilWindowCloses holds every session’s result in an in-memory buffer and emits a single value only once the window closes (after the inactivity gap plus grace period). The trade-offs are real:

  • Latency. End-to-end latency now equals the inactivity gap plus the grace period — a 30-minute gap and 15-minute grace means a session is not emitted until ~45 minutes of event time after its last event. Suppression is incompatible with low-latency dashboards.
  • Memory. BufferConfig.unbounded() grows heap until close. Bound it with maxBytes(...) or maxRecords(...), then choose the overflow behavior: shutDownWhenFull() keeps the strict “emit once, at close” guarantee but halts the app if the buffer overflows, whereas emitEarlyWhenFull() evicts the oldest buffered records early — trading the single-emit guarantee for liveness, which reintroduces some intermediate updates downstream.
  • A non-zero grace is mandatory. untilWindowCloses only ever emits after the window closes, so it requires a grace period set via ofInactivityGapAndGrace. Pair the two, or the operator buffers forever and emits nothing.

Handling Session Merges Downstream

Because sessions are retracted and replaced, treat the windowed key (user + session start/end) as the upsert key in any database or cache sink, and apply tombstones to delete superseded sessions. A Kafka Connect sink consuming session-counts-output should be configured with the windowed key as the record key so that compaction and upserts collapse to the latest version automatically.

Do not try to “deduplicate” by re-aggregating the windowed table with LATEST_BY_OFFSET in ksqlDB — that function operates only on a stream source, not on a table, and a windowed CREATE TABLE produces a table. The correct path is to let the changelog/output topic’s last-write-wins semantics (or your sink’s upsert) handle versioning, keyed by the windowed key.

Testing and Debugging Session Window Logic

Session logic is timestamp-driven, so tests must control event time precisely.

Unit Testing with TopologyTestDriver

TopologyTestDriver runs the topology in-process with no broker. Note that the output key is a Windowed<String>, so the test must deserialize it with a session-windowed serde:

@Test
public void testSessionExtends() {
    StreamsBuilder builder = new StreamsBuilder();
    // ... build the topology shown earlier, writing to "session-counts-output" ...

    try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
        TestInputTopic<String, PageView> input = driver.createInputTopic(
            "page-views",
            Serdes.String().serializer(), pageViewSerde.serializer()
        );
        TestOutputTopic<Windowed<String>, Long> output = driver.createOutputTopic(
            "session-counts-output",
            WindowedSerdes.sessionWindowedSerdeFrom(String.class).deserializer(),
            Serdes.Long().deserializer()
        );

        // Two events 15 minutes apart, within a 30-minute gap -> one session of 2.
        input.pipeInput("user1", new PageView("page1"),
            Instant.parse("2024-01-01T10:00:00Z"));
        input.pipeInput("user1", new PageView("page2"),
            Instant.parse("2024-01-01T10:15:00Z"));

        List<KeyValue<Windowed<String>, Long>> results = output.readKeyValuesToList();
        // With caching disabled, the merge first retracts the first session (null)
        // and re-emits the merged one; assert on the last non-null count.
        assertEquals(2L, results.get(results.size() - 1).value.longValue());
    }
}

pipeInput(key, value, Instant) stamps each record with explicit event time, which is what drives session boundaries. To prove the negative case — that a third event arriving outside the gap starts a new session rather than extending the old one — pipe it at 10:50:00Z and assert two distinct windowed keys come back. For wall-clock punctuation (for example, to drive emission of suppressed results), advance time with driver.advanceWallClockTime(Duration). Remember that TopologyTestDriver processes records synchronously and behaves as if cache.max.bytes.buffering=0 and commit.interval.ms=0, so it surfaces every intermediate update — handy for asserting on retractions you would never see with caching on.

Inspecting State with Interactive Queries

When a running application produces unexpected sessions, read the SessionStore directly via interactive queries:

ReadOnlySessionStore<String, Long> store = streams.store(
    StoreQueryParameters.fromNameAndType(
        "session-counts-store",
        QueryableStoreTypes.sessionStore()
    )
);

try (KeyValueIterator<Windowed<String>, Long> iterator = store.fetch("user1")) {
    while (iterator.hasNext()) {
        KeyValue<Windowed<String>, Long> entry = iterator.next();
        System.out.printf("key=%s count=%d start=%s end=%s%n",
            entry.key.key(),
            entry.value,
            entry.key.window().startTime(),
            entry.key.window().endTime());
    }
}

fetch(key) returns every active session for that key, letting you confirm whether two visits became one merged session or remained distinct. For state-store fundamentals, see the Stream Processing guide.

Production Deployment Checklist

  1. Standby replicas. Set num.standby.replicas to at least 1. Session state is expensive to restore from a changelog; warm standbys cut failover recovery time — at the cost of doubling per-instance state footprint (see sizing above).

  2. Changelog retention. Changelog topics for session stores default to cleanup.policy=compact,delete with retention derived from the store’s retention period. If you override it, keep retention at least the inactivity gap plus grace period plus a safety margin (for a 30-minute gap and 15-minute grace, two hours is comfortable).

  3. Partition by key. Partition input topics by user_id so all of a user’s events land on one partition — session windows aggregate per key within a partition and cannot span partitions. A repartition (via groupByKey on a re-keyed stream) is the fix if your source is partitioned differently.

  4. Memory headroom. Account for both JVM heap and RocksDB off-heap memory (block cache, write buffers). Bound the cache with a shared LRUCache in your RocksDBConfigSetter, and watch GC pauses as on-heap suppression buffers grow.

  5. Grace alignment. Match the grace period to downstream tolerance for late data. If a dashboard refreshes every 5 minutes, a 10-minute grace ensures most sessions are final before they are read.

  6. Monitoring. Track session-store size, block-cache-data-hit-ratio, consumer lag on output topics, dropped-records (out-of-order events past grace), and the rate of retractions — a spike in retractions often signals clock skew or a burst of late data. Remember to run with metrics.recording.level=DEBUG for the RocksDB metrics to appear.

Session windows give you accurate, behavior-driven boundaries for user-activity analytics, but they trade fixed-grid simplicity for dynamic state and merge semantics. The four decisions that determine whether they scale: size against peak concurrent keys (not throughput), tune RocksDB through a real RocksDBConfigSetter with a shared cache, choose your grace period and emit strategy against downstream latency tolerance, and instrument retractions and dropped records so late data is visible rather than silent.

Back to Windowed Aggregations And Joins In ksqlDB