Kafka

60-Day Kafka 4 Learning Plan · Week 6 — Day 41 of 60

Editorial Team

60-Day Kafka 4 Learning Plan · Week 6 — Kafka Connect & ksqlDB Sources: Kafka: The Definitive Guide Ch.11 · docs.ksqldb.io — time and windows

Goal

Go beyond basic SELECT … EMIT CHANGES and write windowed aggregations in ksqlDB SQL. Learn tumbling, hopping, and session windows, correctly configure event-time extraction, handle late-arriving events with GRACE PERIOD, and expose window boundaries with WINDOWSTART / WINDOWEND.

1. Why windowed aggregations?

Streams are unbounded — you can’t GROUP BY an infinite sequence without a time boundary. Windows slice the stream into finite buckets so aggregations are meaningful.

“How many orders per user in the last 5 minutes?” — that’s a windowed aggregation.

2. TUMBLING — revenue per user per 5 minutes

Fixed, non-overlapping windows. Every 5 minutes a new window opens; the previous one closes and its result is emitted.

-- 5-minute tumbling window: order count + revenue per user
CREATE TABLE orders_per_5min AS
  SELECT user_id,
         COUNT(*)      AS order_count,
         SUM(amount)   AS revenue,
         WINDOWSTART   AS win_start,   -- epoch ms of window open
         WINDOWEND     AS win_end      -- epoch ms of window close
  FROM   orders_stream
  WINDOW TUMBLING (SIZE 5 MINUTES, GRACE PERIOD 30 SECONDS)
  GROUP BY user_id
EMIT CHANGES;

Pull query — read a specific window

-- Get the current 5-min window totals for user u-042
SELECT user_id, order_count, revenue,
       TIMESTAMPTOSTRING(win_start, 'HH:mm:ss') AS from_time,
       TIMESTAMPTOSTRING(win_end,   'HH:mm:ss') AS to_time
FROM   orders_per_5min
WHERE  user_id = 'u-042';

Timeline

time:  0     5     10    15    20 min
       │──W1──│──W2──│──W3──│──W4──│
events:  ●  ●    ●       ●  ●    ●

Each event belongs to exactly one window.

3. HOPPING — rolling 10-min average every 5 min

Windows overlap — each event appears in SIZE / ADVANCE BY windows. Use for smoothed metrics where you want continuous updates.

-- Rolling 10-minute average order value, updated every 5 minutes
CREATE TABLE rolling_avg_order AS
  SELECT user_id,
         AVG(amount)   AS avg_order_value,
         COUNT(*)      AS order_count,
         WINDOWSTART   AS win_start,
         WINDOWEND     AS win_end
  FROM   orders_stream
  WINDOW HOPPING (SIZE 10 MINUTES, ADVANCE BY 5 MINUTES, GRACE PERIOD 1 MINUTE)
  GROUP BY user_id
EMIT CHANGES;

Timeline

time:  0     5     10    15    20 min
W1:    │─────10m──────│
W2:          │──────10m──────│
W3:                │──────10m──────│

An event at t=7 appears in W1 and W2.

Key rule: SIZE must be ≥ ADVANCE BY. When SIZE == ADVANCE BY, hopping behaves like tumbling.

4. SESSION — user activity windows

Windows are activity-driven — they open on the first event and extend as long as events keep arriving within the inactivity gap. When no event arrives for longer than the gap, the window closes. Each key (user) has its own independent window timeline.

-- Session window: group orders by user shopping session (30-min gap)
CREATE TABLE user_sessions AS
  SELECT user_id,
         COUNT(*)    AS orders_in_session,
         SUM(amount) AS session_total,
         MIN(amount) AS min_order,
         MAX(amount) AS max_order,
         WINDOWSTART AS session_start,
         WINDOWEND   AS session_end
  FROM   orders_stream
  WINDOW SESSION (30 MINUTES)
  GROUP BY user_id
EMIT CHANGES;

Timeline

user A:  ●──●──●      (30-min gap)      ●──●
         │─Session 1─│                  │Session 2│

user B:  ●   ●──────────────────────────────●──●
         │─────────── Session 1 (long) ──────────│

Session windows can merge — if two windows overlap after a late event arrives, they combine into one.

5. GRACE period — handle late-arriving events

Network delays and out-of-order processing mean events can arrive after their window has already closed. GRACE PERIOD keeps the window alive in the state store to absorb late records.

-- GRACE on tumbling: window closes at T+5min, accepts events until T+5min+30s
WINDOW TUMBLING (SIZE 5 MINUTES, GRACE PERIOD 30 SECONDS)

-- GRACE on hopping
WINDOW HOPPING (SIZE 10 MINUTES, ADVANCE BY 5 MINUTES, GRACE PERIOD 1 MINUTE)

-- Session windows use the inactivity gap as their natural grace period
WINDOW SESSION (30 MINUTES)

What happens without GRACE?

Late events are dropped silently — they arrive after the window is finalized and cannot update the result. Always set a GRACE PERIOD in production, sized to your worst-case network latency.

6. Correcting the time semantics — event-time is set on the STREAM, not via a SET statement

An earlier draft of this material suggested toggling event-time vs ingestion-time with a SET 'ksql.streams.timestamp' = ... statement pointed at either a column name or the literal string 'ROWTIME'. That’s not how ksqlDB’s timestamp extraction actually works, and the two example lines contradicted each other (one implies ROWTIME means ingestion time, but ROWTIME is just the pseudo-column name that always holds whatever timestamp ksqlDB is using — setting a property to the string 'ROWTIME' doesn’t select “ingestion time” as a mode).

The correct mechanism — event-time extraction is configured in the WITH clause when you CREATE STREAM, pointing TIMESTAMP at a column that holds your actual business event time:

CREATE STREAM orders_stream (
  order_id   VARCHAR,
  user_id    VARCHAR,
  amount     DOUBLE,
  status     VARCHAR,
  event_ts   BIGINT          -- epoch millis of when the order actually happened
)
WITH (
  KAFKA_TOPIC = 'labs.events',
  VALUE_FORMAT = 'JSON',
  TIMESTAMP    = 'event_ts'  -- ★ this is the real mechanism — extracts event time from this column
);

Without an explicit TIMESTAMP column, ROWTIME defaults to the Kafka record’s own timestamp — which, exactly as covered in Day 31 §8 for raw Kafka Streams, could be CreateTime (set by the producer, closer to true event time) or LogAppendTime (set by the broker on write) depending on your topic’s message.timestamp.type configuration. This is the same event-time-vs-processing-time distinction from Day 31 §8, just configured through ksqlDB’s WITH clause instead of a Java TimestampExtractor implementation.

Why this matters for windowing specifically: all the window types in §2–§4 bucket events based on whatever ROWTIME resolves to. If TIMESTAMP isn’t pointed at your actual business event-time column, a WINDOW TUMBLING (SIZE 5 MINUTES) is bucketing by broker receipt time, not by when orders actually happened — silently producing the same “windowed by the wrong notion of time” bug flagged in Day 31 §8, just via ksqlDB’s WITH clause instead of a Streams extractor. Set TIMESTAMP explicitly whenever the topic carries a real business timestamp field.

7. Window retention — the ksqlDB equivalent of Day 31 §10

Exactly as covered for raw Kafka Streams windowed aggregations, ksqlDB distinguishes grace period (how long late events are still accepted) from retention (how long the computed window results remain queryable) — the same two-different-numbers distinction from Day 31 §10, expressed via SQL instead of Materialized.withRetention(...).

CREATE TABLE orders_per_5min AS
  SELECT user_id, COUNT(*) AS order_count, SUM(amount) AS revenue
  FROM   orders_stream
  WINDOW TUMBLING (SIZE 5 MINUTES, GRACE PERIOD 30 SECONDS, RETENTION 6 HOURS)
  GROUP BY user_id
EMIT CHANGES;

Same trap as Day 31 §10: setting only GRACE PERIOD and leaving RETENTION at its default can make old window results disappear from pull-query lookups (§2’s example) sooner than you might expect, even though late-event handling itself is working correctly. If a dashboard or REST endpoint needs to look back further than a few minutes into past windows, set RETENTION explicitly — it must be ≥ the grace period, and controls the backing changelog topic’s effective queryable history.

8. Testing windowed ksqlDB queries

Unlike raw Kafka Streams DSL code (TopologyTestDriver, Days 29–34), ksqlDB SQL statements aren’t unit-testable in the same way by default — but ksqlDB ships a dedicated tool for exactly this.

# ksql-test-runner: feed a JSON fixture of input records and expected output,
# validate a .sql file's statements produce the expected windowed results
ksql-test-runner \
  -s revenue_windowing.sql \
  -i input_orders.json \
  -o expected_output.json
// input_orders.json — records with explicit timestamps to exercise window boundaries
[
  {"topic": "labs.events", "timestamp": 0,      "value": {"user_id": "u1", "amount": 50}},
  {"topic": "labs.events", "timestamp": 299000, "value": {"user_id": "u1", "amount": 30}},
  {"topic": "labs.events", "timestamp": 301000, "value": {"user_id": "u1", "amount": 20}}
]

Same testing principle as Day 31 §11’s advanceWallClockTime: the fixture must include records that deliberately straddle a window boundary (here, just before and just after the 5-minute mark) to actually exercise the window-close and late-arrival logic — a fixture with all timestamps clustered together never tests the behavior that matters most. This is the SQL-pipeline equivalent of the same testing discipline, just driven by a JSON fixture file instead of Java test code.

9. Monitoring windowed query resource usage

Building directly on §8 from Day 40 (ksqlDB is Kafka Streams underneath): every WINDOW-based CREATE TABLE here creates its own named windowed state store and changelog topic (Day 29 §8, Day 31 §10), multiplying the monitoring surface for a pipeline with several windowed aggregations running concurrently.

# Each windowed TABLE gets its own changelog topic —
# naming follows {KSQL_KSQL_SERVICE_ID}{query-id}-changelog
kafka-topics.sh --bootstrap-server localhost:9092 --list | grep changelog

# Confirm actual disk usage per windowed aggregation
kafka-log-dirs.sh --bootstrap-server localhost:9092 --describe | grep changelog

Why this deserves explicit attention in a SQL-driven pipeline: it’s easy to accumulate several WINDOW TUMBLING/HOPPING/SESSION tables across iterative development (§10 from Day 40’s “audit persistent queries” advice applies directly) without a strong mental model of each one’s disk footprint, since none of it is visible in the SQL text itself. A HOPPING window in particular (Day 31 §2’s “counted size/advance times” cost) multiplies storage for the same input volume compared to an equivalent TUMBLING window — worth factoring into the choice between them, not just their query semantics.

10. Full example: real-time order analytics pipeline

-- 1. Base stream (from Day 40) — with correct event-time extraction (§6)
CREATE STREAM orders_stream (
  order_id VARCHAR, user_id VARCHAR,
  amount DOUBLE,    status  VARCHAR,
  event_ts BIGINT
) WITH (KAFKA_TOPIC='labs.events', VALUE_FORMAT='JSON', TIMESTAMP='event_ts');

-- 2. Filtered confirmed-only stream
CREATE STREAM confirmed AS
  SELECT * FROM orders_stream
  WHERE status = 'CONFIRMED'
EMIT CHANGES;

-- 3. Tumbling 5-min revenue table, with explicit retention (§7)
CREATE TABLE revenue_5min AS
  SELECT user_id,
         COUNT(*)    AS orders,
         SUM(amount) AS revenue,
         WINDOWSTART AS win_start,
         WINDOWEND   AS win_end
  FROM   confirmed
  WINDOW TUMBLING (SIZE 5 MINUTES, GRACE PERIOD 30 SECONDS, RETENTION 6 HOURS)
  GROUP BY user_id
EMIT CHANGES;

-- 4. Session-based cart analytics
CREATE TABLE cart_sessions AS
  SELECT user_id,
         COUNT(*)    AS items_viewed,
         SUM(amount) AS cart_value,
         WINDOWSTART AS session_start
  FROM   orders_stream
  WINDOW SESSION (30 MINUTES)
  GROUP BY user_id
EMIT CHANGES;

-- 5. Write 5-min revenue to a new Kafka topic
CREATE TABLE revenue_5min_topic
  WITH (KAFKA_TOPIC='labs.order-stats', VALUE_FORMAT='JSON')
AS SELECT * FROM revenue_5min
EMIT CHANGES;

11. Common pitfalls

  • Relying on ROWTIME‘s default without checking what it actually resolves to — silently windows by broker receipt time instead of business event time unless TIMESTAMP is set explicitly in the stream’s WITH clause (§6)
  • Setting GRACE PERIOD but never RETENTION — old window results can vanish from pull-query lookups sooner than expected, even though late-event handling works correctly (§7)
  • Testing windowed SQL with all-clustered timestamps — never actually exercises window-close or late-arrival code paths; fixtures need records deliberately straddling window boundaries (§8)
  • Accumulating several windowed TABLEs across iterative development without auditing them — each is a real changelog topic and RocksDB store, same resource-accumulation risk as Day 40 §10’s persistent-query cleanup advice, compounded by windowing’s extra storage overhead
  • Choosing HOPPING for a use case TUMBLING would serve — HOPPING’s per-event multiplication cost (Day 31 §2) is easy to reach for by default without weighing it against a simpler, cheaper TUMBLING window

Window type comparison

Key Takeaways

  • Windows slice unbounded streams into finite time buckets for aggregation
  • TUMBLING — fixed non-overlapping; each event in exactly one window
  • HOPPING — overlapping; use for rolling averages (SIZE > ADVANCE BY), at higher storage cost than tumbling
  • SESSION — activity-driven; window closes after inactivity gap per key
  • Event-time extraction is configured via TIMESTAMP in CREATE STREAM ... WITH (...), not a SET statement — without it, windowing silently uses broker receipt time by default
  • GRACE PERIOD absorbs late-arriving events — always set it in production
  • RETENTION is a separate, larger number from GRACE PERIOD — controls how long window results stay queryable, not just how long late events are accepted
  • WINDOWSTART / WINDOWEND expose bucket boundaries in output records
  • ksql-test-runner is the SQL-pipeline equivalent of TopologyTestDriver + advanceWallClockTime — test with fixtures that straddle window boundaries
  • Every windowed TABLE is a real changelog topic and state store — audit accumulated windowed queries the same way as any persistent query

Support me through GitHub Sponsors.

Thank you for Reading !! See you in the next post.

Next

➡️ Day 42: Week 6 capstone — DB → Kafka 4 → Redis → WebSocket

Resources

, ,

Related Posts

Kafka

60-Day Kafka 4 Learning Plan · Week 9 — Day 60 of 60

Editorial Team
Kafka

60-Day Kafka 4 Learning Plan · Week 9 — Day 59 of 60

Editorial Team