Event Time, Windows and Late Data

Data Engineering & Modern Data Platforms · Lesson 45 · Real-Time Data Engineering and Kafka

Lesson Overview

The five-minute course activity tile already has a definition. Structured Streaming with Spark counts learning.progress by course_id in tumbling windows of five minutes, on event_time, with a watermark delay of ten minutes, and it emits each window once in append mode. Apache Flink Fundamentals repeated those numbers so a second engine would not invent a second tile. This lesson is the Kafka Streams version of the same contract, plus the vocabulary Streams uses when Spark says "watermark."

A mobile client buffers progress on a train and flushes twenty minutes later. Those events carry an event_time from the lesson player, not from the moment the broker appended the record. If the window is computed on broker time, the flush looks on time and the tile shows a spike in the wrong five minutes. Spark already refuses that definition. Streams will happily use the record timestamp, which the producer or the broker set, unless you install a TimestampExtractor that reads event_time. The extractor is the whole alignment. Without it, matching counts is impossible and both jobs look broken.

Streams does not call the lateness threshold a watermark. It tracks stream time, the highest timestamp a task has seen, and it keeps a window open for a grace period after the window ends. Grace of ten minutes on a five-minute tumble is the same product decision as Spark's ten-minute watermark: an event may still change its window until stream time moves far enough past the window end. The mechanisms differ, and the difference is what you operate. Spark in append mode emits the window once and then drops state. Streams updates a windowed store continuously unless you suppress until the window closes. Continuous updates are Spark's update mode. suppress plus grace is the append-mode shape. Pick the same shape as the tile or stop comparing the two outputs.

The running revenue total from Kafka Streams Fundamentals is not windowed. Do not put a grace period on it to look consistent. Lateness does not change a purchase line's cents. It changes which five minutes a progress event belongs to.

/learn/kafka explains record timestamps as metadata the log can store. This lesson uses that metadata only as a fallback you should not need, and it puts event time in the extractor. The morning course_day table remains the historical record, because anything later than grace is dropped from the tile on purpose.

Learning Objectives

By the end of this lesson you will be able to:

1. Distinguish event time, processing time, ingest time, and Kafka Streams stream time.
2. Implement a TimestampExtractor that reads event_time so a Streams tile can match the Spark tile.
3. Define a five-minute tumbling window with a ten-minute grace using TimeWindows.ofSizeAndGrace, and avoid the deprecated constructor whose default grace was a day.
4. Explain when a record is late, which metric moves when it is dropped, and why an idle partition does not close the window.
5. Use suppress until the window closes to emit each window once, and say how that maps to Spark append mode.
6. Contrast tumbling, hopping, and session windows as different products, not as interchangeable knobs.
7. State why per-task stream time can mark a slow partition's records as late when another partition in the task has run ahead.
8. Keep the morning batch as the system of record for counts that must include data later than grace.
9. Test closure with a later event, not with a sleep, the same way lesson 32 tested the watermark.
10. Decline a second production tile unless the counts have been compared window by window.

Prerequisites

• Structured Streaming with Spark: five-minute tumble, ten-minute watermark, append mode, event_time, and the checkpoint. This lesson reuses that contract and does not reopen output modes.
• Apache Flink Fundamentals: the same numbers on a third engine, so you have seen that "watermark" is not a Spark trademark. Streams will use different words for a related idea.
• Lesson 44: topology, application.id, state stores, changelogs, and TopologyTestDriver. Window stores are state stores with a retention.
• Data Quality Engineering: a dropped late event is a severity decision. The batch restatement is the correction path.
• /learn/kafka for the record timestamp (CreateTime versus LogAppendTime). You need to know that the log's timestamp is not automatically the player's clock.

Detailed Core Concepts

Four clocks

Event time is event_time in the progress payload, assigned by the learning client in UTC. It is the clock the tile is about: when the learner did the thing.

Ingest time is when the broker appended the record, or when the producer set CreateTime at send. A client that was offline has an ingest time near "now" and an event time twenty minutes ago. Those clocks answer different questions.

Processing time is when the Streams task handled the record. It moves when the task is slow. A backlog then looks like it happened during the catch-up.

Stream time is a Kafka Streams notion. For a task, it is the maximum record timestamp observed so far on the partitions that task processes. It never moves backward. It does not move because the wall clock moved. If records stop, stream time stays, and windows stay open. Spark's watermark has the same idle behavior: max(event_time) - delay does not advance when no new event arrives. Lesson 32 told you to test by sending a later event, not by sleeping. The same test is required here.

Stream time is not one number for the whole application. It is per task. A task may own several partitions. If partition 0 receives an event stamped 12:40 and partition 1's next event is stamped 12:10, and both partitions are on the same task, stream time can already be 12:40 when the 12:10 event is processed. With a ten-minute grace, 12:10 may already be late relative to 12:40. In Spark, the watermark is also driven by the maximum event time observed, with a documented policy when more than one input watermark exists. The operational symptom is the same: a fast partition can close windows out from under a slow one. The fix is not a bigger grace as a habit. The fix is to notice skew, to avoid putting a hot partition and a cold one in a pairing you did not look at, and to size grace for real client delay rather than for partition skew. Skew of an hour is an incident. Client delay of ten minutes is the product.

The extractor is the definition

TimestampExtractor is org.apache.kafka.streams.processor.TimestampExtractor. The library calls extract(record, partitionTime) and uses the returned milliseconds as the record's timestamp for windowing and for stream time. The default, FailOnInvalidTimestamp, uses the timestamp already on the ConsumerRecord. That value is CreateTime if the producer set it, or LogAppendTime if the topic is configured that way. Neither is event_time inside a JSON or Avro payload.

A custom extractor parses the payload and returns event_time. If the field is missing, throw. Returning partitionTime or wall-clock time hides the hole and puts the event in a window you will defend by mistake. LogAndSkipOnInvalidTimestamp exists in the library and will drop bad timestamps. On this tile, a missing event_time is a contract break with lesson 32. Fail the task, or send the record to a reject topic in the topology before the window, which you can do only after a parse step that does not depend on the extractor. The extractor itself should throw on a missing field so the failure is visible. A reject topic is a second path you add if operations cannot tolerate a stop. Prefer the stop until the producer is clean, then keep the stop.

Configure it with default.timestamp.extractor set to your class. One extractor for the application. A topology that windows progress and also windows purchase lines would want different fields. Those are two applications, which you already have: this tile, and the unwindowed revenue job. Do not share an application.id between them, and do not share an extractor that reads event_time from a purchase line that stores the timestamp under a different name.

Windows are products

Tumbling windows tile the line without overlap. TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(10)) is the tile: buckets [12:00, 12:05), [12:05, 12:10), and so on, and each bucket accepts events for ten minutes of stream time after it ends. The Spark expression window(event_time, "5 minutes") plus withWatermark(event_time, "10 minutes") is the same bucket and the same lateness budget. Compare counts on (course_id, window_start) after both sides have closed. Differences should be empty or explained by events one job has not closed yet. That sentence is the same bar lesson 33 set for Flink.

Hopping windows overlap. A five-minute window that advances every minute (ofSizeAndGrace(5 minutes, grace).advanceBy(1 minute)) answers "the last five minutes" every minute. It stores more state, it emits overlapping buckets, and it will not match the Spark tumble. Do not use it as a drop-in replacement for the tile. It is a different product, useful for a live "recent activity" strip if someone asks for one, with its own name.

Session windows group events that fall within an inactivity gap. SessionWindows.ofInactivityGapAndGrace(Duration.ofMinutes(10), Duration.ofMinutes(10)) is a learning session: events for one learner, a new session when the gap exceeds ten minutes. The key is learner_id, not course_id. The state is per learner, which is much larger than per course. Lesson 33 already warned that session state is a different budget from the tile. Do not implement sessions by keeping every event in a list. Use the session window. Do not compare session counts to the five-minute course tile.

Sliding windows (SlidingWindows.ofTimeDifferenceAndGrace) start from each record's timestamp and the difference you set. They can create many windows per key. They are the wrong tool for the course tile. Leave them until a requirement is actually "every pair of events this far apart," and then read the version's page before you size state.

The deprecated TimeWindows.of(Duration.ofMinutes(5)) used a default grace of 24 hours. Windows stayed open for a day, state held a day of course keys, and late data from yesterday rewrote totals you thought were done. ofSizeWithNoGrace sets grace to zero: any record older than stream time is late as soon as a newer timestamp has been seen. That is harsher than the Spark tile and will drop ordinary mobile delay. Use ofSizeAndGrace with ten minutes so the number is written in the code.

Grace, closure, and suppress

A record falls in a window by its timestamp. The window remains updateable until stream time passes the window end plus grace. After that, a record for that window is late. Streams drops it and increments the task's dropped-record metric. It does not append the event to a side output unless you add that branch yourself before the window, for example by comparing timestamps in a processor. The stock window operator drops. If you need the late events for the morning reconciliation, they are still on learning.progress as long as retention holds. You do not need the tile to keep them. The batch job reads the landed files with its restatement window, which is the design lesson 32 already stated.

Without suppress, every update to the window store can flow downstream. A five-minute count would be rewritten many times as events arrive, which is Spark update mode. The file sink in lesson 32 cannot do that. A Kafka topic can, if the consumer keeps the latest value per (course_id, window_start).

suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded())) holds those updates and emits one record when the window closes. That is the append-mode shape: one output per window, after grace. unbounded() means the suppress buffer will not drop data to shed load. If you cannot hold the open windows, the honest response is to scale the state or narrow the product, not to switch to a buffer that evicts. untilWindowCloses needs a grace that is defined, including zero. A ten-minute grace defines it. The old default of 24 hours also "defined" it and made closure a day late. Explicit grace is the point.

Suppress emits to the output topic when the window closes. Downstream consumers of that topic see the closed count, not the partials. The Spark file sink sees the closed count when the watermark passes. Align those two if you dual-run. Do not compare a suppressed Streams output to a Spark update-mode sink, or to a Streams output you forgot to suppress.

State retention

Window stores keep state until the window is closed and then for a retention period so late-but-within-grace updates still find the window. The retention must be at least the window size plus grace, or the store can discard a window that grace would still have accepted. Set retention explicitly when you materialize the window store, and make it longer than size plus grace by the outage margin you care about. A restore that takes thirty minutes should not lose the windows grace still considers open. Changelog topics for window stores are part of the same contract as lesson 44: replicated, not given a short retention.ms that fights compaction.

What this tile is allowed to be wrong about

Events later than grace are absent. The tile is allowed to disagree with tomorrow's course_day count. That disagreement is the lateness budget, not a defect, unless it exceeds what the product named. A second Streams tile in production beside Spark doubles the on-call surface. Build this topology to understand the alignment, compare a replayed hour, and keep one engine as the served tile. The comparison is the lesson. The decommission is a separate decision with the numbers written down.

Architecture and Design Diagrams

mermaidCopy
flowchart LR
    subgraph source [learning.progress]
        rec[Record bytes plus broker timestamp]
    end

    subgraph task [Streams task]
        ex[TimestampExtractor reads event_time]
        st[Stream time equals max timestamp seen]
        win[5-minute tumble grace 10 minutes]
        store[(Window store)]
        sup[Suppress until window closes]
    end

    subgraph compare [Same contract]
        spark[Spark append tile watermark 10 minutes]
        batch[Morning course_day restatement]
    end

    rec --> ex --> win
    ex --> st
    st --> win
    win <--> store
    win --> sup
    sup --> spark
    rec --> batch

The arrow from suppress to Spark is a comparison during a dual-run, not a runtime dependency. The batch reads the topic or the landed files, including events the tile dropped as late.

mermaidCopy
sequenceDiagram
    participant P as Partition
    participant T as Task
    participant W as Window 12:00-12:05
    participant O as Output

    P->>T: event_time 12:04 course A
    Note over T: Stream time becomes 12:04
    T->>W: Count is 1, window not closed
    P->>T: event_time 12:09 course A
    Note over T: Stream time becomes 12:09
    Note over W: End plus grace is 12:15, still open
    P->>T: event_time 12:16 course B
    Note over T: Stream time becomes 12:16
    T->>O: Suppress emits the 12:00 window once
    P->>T: event_time 12:03 course A
    Note over T: 12:03 is before stream time minus grace
    T->>T: Drop the late record

Wall-clock sleep between the 12:04 event and the 12:16 event does nothing if those records never arrive. Closure is a property of timestamps the task has seen.

Technical Explanation

Extractor

javaCopy
public final class ProgressEventTimeExtractor implements TimestampExtractor {
    @Override
    public long extract(ConsumerRecord<Object, Object> record, long partitionTime) {
        Object value = record.value();
        if (!(value instanceof ProgressEvent event) || event.eventTimeMillis() <= 0L) {
            throw new StreamsException(
                "learning.progress record missing event_time at offset " + record.offset());
        }
        return event.eventTimeMillis();
    }
}

The value deserializer has already turned bytes into ProgressEvent. If deserialization fails, the fail-close handler from lesson 44 runs before a bad timestamp can become a window. eventTimeMillis() is the UTC instant from the payload, the same field Spark parses as event_time. Returning record.timestamp() would reintroduce broker time. Returning partitionTime on failure would assign the event to whatever timestamp the partition had already seen, which is a quiet way to pass the test suite and fail the train ride.

Windowed count

javaCopy
KStream<String, ProgressEvent> progress = builder.stream(
    "learning.progress",
    Consumed.with(Serdes.String(), progressSerde)
        .withTimestampExtractor(new ProgressEventTimeExtractor()));

KTable<Windowed<String>, Long> counts = progress
    .groupBy((key, event) -> event.courseId(), Grouped.with(Serdes.String(), progressSerde))
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(10)))
    .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("progress-counts")
        .withRetention(Duration.ofMinutes(5 + 10 + 30)));

counts.suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
    .toStream()
    .to("learning.progress.course_window");

withTimestampExtractor on Consumed is the per-source form, which is better than the application default when you are explicit. groupBy on courseId repartitions, as lesson 44 warned. The progress topic's key may be event_id or a learner id. The tile's key is course_id. Pay the repartition. Do not key the progress topic by course_id if producers use event_id for idempotency.

Retention of 45 minutes is window size plus grace plus a 30-minute margin. Adjust the margin to your restore time. It is not the lateness budget. Grace is the lateness budget.

The output key is windowed: course_id plus window start and end. Downstream JSON should include those as fields so a consumer does not have to know Windowed serde internals. Map to a plain record before to if the lake sink is easier that way. Include window_start, window_end, course_id, and count.

Hopping and session, named so they are not the tile

A hopping "last five minutes" for a live strip:

javaCopy
TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(2)).advanceBy(Duration.ofMinutes(1))

Grace of two minutes here is a different product decision. Do not copy the tile's ten minutes without thinking. Overlapping windows with a long grace multiply state.

A session per learner:

javaCopy
SessionWindows.ofInactivityGapAndGrace(Duration.ofMinutes(10), Duration.ofMinutes(10))

Key the group by learner_id. Emit session start, end, and event count. This is not a course tile. It will not match Spark's window(event_time, "5 minutes"). It answers "did this learner go quiet."

How this lines up with Spark's watermark

Spark computes watermark = max(event_time) - 10 minutes and, in append mode, emits a five-minute window after the watermark passes the window end, then drops that state. An event with event_time older than the watermark is not applied.

Streams computes stream time as max(timestamp) per task and drops a record that is older than stream time minus grace when it can no longer fall in an open window. With grace equal to the Spark delay, a window closes when a later event pushes stream time past the window end plus grace. Both systems wait for a later timestamp. Both stay open if the source goes idle. Both can let a fast partition's timestamps outrun a slow partition.

They are not the same code. Spark 3.5's watermark can be tracked per stateful operator, and multiple watermarks have a policy. Streams' stream time is per task and advances as records are processed in that task. A dual-run on one replayed partition, one course_id, and one hour should match exactly once both have seen a timestamp past the last window's end plus ten minutes. A dual-run on a live topic with many partitions should match after you account for which offsets each engine has committed. If they do not match on a single-partition fixture, the extractor is wrong, the grace is wrong, or one side is not suppressing and you are comparing a partial update to a final count.

Spark's failOnDataLoss and checkpoint are still how the Spark job recovers. Streams recovers with the application.id offsets and the window changelog. Do not unify those cursors. Compare outputs, not offsets.

Late data and the morning table

The batch job's restatement window, from the Spark lesson, is allowed to include events the tile dropped. Publish the tile as "closed under a ten-minute grace." Publish course_day as "restated through the batch policy." A reconciliation that expects equality every hour will page forever. Reconcile a closed hour on a quiet day to catch a bug, and expect a small gap when clients were offline. Page when the gap exceeds a threshold you set with the product, not when it is non-zero.

Practical Implementation or Design Walkthrough

Problem

Build the Streams tile so a one-hour fixture matches the Spark tile's definition: five-minute tumble, ten-minute grace, one output row per (course_id, window_start) after closure. Do not serve both in production as part of this lesson.

Fixture

Three events, one partition, one course:

Processing order is e1, e2, e3, then e4.

e1 falls in [12:00, 12:05). e2 falls in [12:05, 12:10) and moves stream time to 12:06. The first window's end plus grace is 12:15, so it stays open. e3 moves stream time to 12:16, which closes the 12:00 window. Suppress emits count 1 for course-a at 12:00. e4 is event time 12:03, which would have belonged to the 12:00 window, and stream time is 12:16. The window is closed. e4 is dropped. The 12:05 window is still open because its end plus grace is 12:20.

Send e5 at 12:21 and expect the 12:05 window to emit with count 1. That is the test. Sleeping for ten minutes without e5 leaves both windows unemitted. If your test sleeps and then asserts, it will fail for the right reason.

Driver

Pipe records with explicit timestamps equal to event_time, and also set the payload field, so a bug that uses the record timestamp still passes only if those two match. Then change the piped timestamp to the "broker time" and keep the payload time, and assert that the extractor's result follows the payload. If the test still windows on the piped timestamp, the extractor is not installed.

javaCopy
input.pipeInput("e1", event("course-a", "2026-03-02T12:04:00Z"), millis("2026-03-02T12:04:00Z"));
input.pipeInput("e3", event("course-a", "2026-03-02T12:16:00Z"), millis("2026-03-02T12:16:00Z"));
// After a record at 12:16, the 12:00 window for the earlier event can close.

Read learning.progress.course_window and assert a single record for the closed window. Pipe the late 12:03 event last and assert the output count does not become 2.

Dual-run note

Export an hour of learning.progress to a fixture topic with one partition. Run the Spark query from lesson 32 with Trigger.AvailableNow and this topology with a reset application id. Join outputs on course_id and window_start. Investigate any mismatch before you discuss cluster sizing. The usual mismatches are: Spark used event_time and Streams used CreateTime; one side's delay was ten minutes and the other's grace was zero because ofSizeWithNoGrace was copied from a snippet; suppress was omitted and Streams emitted partials; the Spark query was in update mode.

Session add-on, not the acceptance test

If you have time, a second topology with a new application.id windows by learner_id with a ten-minute inactivity gap. Show two bursts twenty minutes apart becoming two sessions, and two bursts five minutes apart becoming one. Do not write those sessions into learning.progress.course_window.

Enterprise Real-World Scenarios

The train and the tile

A learner completes a segment at 12:04 UTC offline and the client uploads at 12:18. event_time is 12:04. Broker append time is 12:18. Grace is ten minutes. Whether the event makes the 12:00 window depends on stream time at the moment the task processes it. If the partition has already seen a 12:16 event, the 12:00 window is closed and this completion is late. The tile misses it. The morning course_day restatement includes it, because the file landing retained the record. Product has already accepted that the live tile is not the system of record. The incident is when someone pages engineering because the tile and the next morning differ by the offline completions. The response is to show grace and the restatement policy, not to set grace to 24 hours by reviving TimeWindows.of. A day of grace makes the tile a slow batch with RocksDB. If product wants a thirty-minute grace because train rides are longer than ten minutes, change both the Spark watermark and the Streams grace in the same review, or stop claiming the tiles match.

Partition skew closes someone else's window

course-a is in a live event and produces a record every second. course-b trickles. If they share a task, course-a's timestamps pull stream time forward and course-b's slightly delayed events start dropping even though course-b is not late by the wall clock. The Spark job can show the same pattern when one max event time dominates the watermark. The design response is to measure dropped-record rate by task and to look at partition assignment, not to disable grace. If one course's clock is wrong and stamps event time in the future, stream time jumps and everyone else looks late. Reject event times more than a few minutes ahead of the broker time in the producer, or in a flatMap that routes future-dated events to a reject topic before they enter the window. A future timestamp is a poison clock. Grace does not protect other keys from it. The extractor can refuse timestamps more than fifteen minutes ahead of record.timestamp() when the topic uses CreateTime close to ingest, and that threshold is a product number you write down. It is not a second watermark. It is a validation, in the spirit of lesson 17, applied before the window.

Best Practices

• Read event_time in a TimestampExtractor. Do not window learning.progress on broker append time and call it the tile.
• Use ofSizeAndGrace or ofSizeWithNoGrace. Do not use the deprecated TimeWindows.of that defaulted to 24 hours of grace.
• Set grace to the same duration as the Spark watermark delay if the products are meant to match. Ten minutes is the current contract.
• Suppress until close when you want one record per window. Leave updates flowing only if the consumer is built to overwrite.
• Test closure by sending a later event. Assert that a sleep without a later event does not emit.
• Keep window-store retention at or above window size plus grace, plus restore margin.
• Fail on a missing event_time. Do not substitute wall-clock time.
• Bound future event times before they advance stream time for the task.
• Leave late events on the source topic for the batch restatement. Do not expect the tile to equal course_day.
• Dual-run on a fixture before you discuss replacing Spark. One served tile is enough.
• Separate session windows onto their own application.id and their own output topic.
• Watch dropped records and per-task lag together. A grace change is a product change, not a midnight config tweak.

Common Mistakes and Anti-Patterns

Broker time as event time. The offline flush lands in the window when it was uploaded. The Spark tile, using event_time, disagrees. The extractor is the fix, not a bigger grace.

TimeWindows.of and a day of grace. State grows, windows rewrite yesterday, and the "match Spark" comparison is nonsense. Name the grace in ofSizeAndGrace.

Sleeping in the test. Stream time does not move with the wall clock. The assertion flakes or, worse, someone adds a punctuator that closes windows on processing time and silently changes the definition.

Comparing unsuppressed Streams updates to Spark append output. One side has partial counts. They both look wrong. Suppress, or compare Spark update mode to unsuppressed Streams, and do not use the file sink for that comparison.

Hopping windows as the tile. Overlap means the Spark tumble cannot match. Name the product.

Session state stored as a list of every event. The window operator already exists. A list will OOM on an active learner and will not close when stream time does.

Grace of zero copied from ofSizeWithNoGrace because it "sounded strict." Ordinary jitter and a one-second reorder drop events. The Spark job with ten minutes will not drop them. The tiles diverge on every slightly late record.

One global grace to paper over a future-dated poison clock. One bad producer moves stream time to tomorrow. Reject future timestamps at the door.

Raising grace in Streams only. The served Spark tile and the experimental Streams tile no longer share a contract, and someone will still subtract them in a spreadsheet. Change both or neither.

Troubleshooting and Design Review

Symptom: The output topic stays empty while input lag is zero.

Likely cause: no record has advanced stream time past the window end plus grace, or suppress is waiting for that moment. The same symptom as Spark append mode with a watermark that has not passed the window end.

Investigation: the maximum event_time processed, the window end, and grace. Confirm the extractor is the class you deployed, not the default.

Corrective action: send or wait for a later event in the fixture. In production, check that producers are still sending. Do not switch to processing time to force a close.

Symptom: Counts are lower than Spark for the same closed window.

Likely cause: Streams dropped records as late because grace is shorter, the extractor used a different clock, or a future-dated event on the same task closed windows early. Alternatively Spark has not actually closed the window you think it has.

Investigation: dropped-record metric on the task, a sample of event_time versus record.timestamp(), and whether one poison timestamp sits in the future.

Corrective action: align extractor and grace with the Spark job. Quarantine future timestamps. Reprocess the hour from retention if the output was wrong and the source still has the bytes. A new application.id avoids mixing old window state with the corrected definition.

Symptom: Counts are higher than Spark, and they change after you thought the window closed.

Likely cause: suppress is missing, so you are reading partial updates, or grace is 24 hours because a deprecated API slipped in.

Investigation: the topology description for suppress and the window definition's grace. Topology.describe() is the artifact to read, not the blog you copied.

Corrective action: add suppress, replace the window definition, and reset the application. Partial updates already written to the output topic need a consumer that keeps the latest key or a new output topic.

Symptom: One course's windows look on time. Another course on the same task loses events that are only a minute late.

Likely cause: shared stream time. The fast course advances the task's max timestamp.

Investigation: partition assignment and per-partition timestamp ranges.

Corrective action: confirm the loss is inside the product's tolerance. If not, isolate the hot course's partitions or reduce skew. Increasing grace for everyone increases state for everyone. Do that only as a product decision.

Symptom: After a restore, a window that was open is gone and its count is short.

Likely cause: store retention shorter than size plus grace, or changelog retention deleted segments the restore needed.

Investigation: the Materialized.withRetention value and the changelog topic's retention configuration.

Corrective action: lengthen retention, reset, and rebuild from learning.progress if the source retention still covers the hour. If the source is gone, the hour is whatever the batch landing captured. Say so.

A design review asks: Which field is event time, and which class extracts it? What are the window size and the grace in code? Does suppress emit once? What metric shows drops? What does the morning table include that this tile will not? What happens to a timestamp fifteen minutes in the future? Is this topology the served tile or a comparison? What is the application.id, and is it distinct from the revenue job?

Interview Questions and Answers

Q1. What is the difference between event time, processing time, and stream time in this tile?

Event time is event_time from the learning client. Processing time is when the Streams task runs the record. Stream time is the highest event timestamp that task has seen, and it is the clock grace uses. It does not advance when the task is idle, and it does not go backward. I window on event time by installing a TimestampExtractor. If I leave the default extractor in place, I window on the Kafka record timestamp, which is producer send time or broker append time, and I will not match the Spark tile.

Q2. How does Kafka Streams grace relate to Spark's watermark on this course's tile?

Spark's watermark is max event time minus a ten-minute delay, and append mode emits a five-minute window once the watermark passes the window end. Streams keeps a tumbling window updateable until stream time passes the window end plus grace. I set grace to ten minutes with TimeWindows.ofSizeAndGrace so the lateness budget matches. I suppress until the window closes so the output is one record, like append mode, not a stream of partial sums. The code is not the same. A single-partition fixture should match after both sides have seen a timestamp past the last window. I do not expect equality with course_day, which restates later data.

Q3. Why is TimeWindows.of(Duration.ofMinutes(5)) a defect here?

That older factory defaulted grace to 24 hours. Windows would stay open for a day, state would hold a day of updates, and the result would not be the Spark tile's ten-minute contract. I use ofSizeAndGrace and write both durations at the call site. ofSizeWithNoGrace is the zero-grace factory. It is honest and too strict for mobile clients that reconnect a few minutes late.

Q4. How do you close a window in a test?

I pipe a record whose event time is past the window end plus grace. I do not sleep. Stream time moves when a greater timestamp is processed. An idle source leaves the window open, which is also how Spark's watermark behaves. The test asserts one emitted count after the closing event, and it asserts that a subsequent record for the closed window does not change the count.

Q5. What happens to an event that shows up after grace?

The window operator drops it and the task's dropped-record metric increases. The event remains on learning.progress until retention deletes it. The morning batch can still count it from the landed files. I do not add the late event to a closed tile to force a match with the batch. If product needs a longer delay, I change Spark's watermark and Streams' grace together.

Q6. Why can one fast partition make another partition's events look late?

Stream time is per task, and a task may process several partitions. The maximum timestamp wins. A hot course that has already moved the max forward causes an older timestamp on a quiet partition in that task to fall outside grace. Spark's max-event-time watermark can do the same to a query. I measure drops per task before I lengthen grace. I also reject event times far in the future so one bad clock cannot move stream time to tomorrow.

Q7. When would you use a hopping window or a session window instead of this tumble?

A hop with a one-minute advance answers a rolling "last five minutes" and will not match the Spark tumble. I name it as a different output. A session window with an inactivity gap answers "this learner's burst of activity" and is keyed by learner, with a much larger state. I do not use either as a silent substitute for the course tile. Sliding windows create a window from record timestamps and a time difference, and I do not reach for them for a five-minute course count.

Q8. What does suppress(untilWindowCloses) change for the consumer?

Without it, each update to the count can be written downstream, and the consumer must keep the latest value per windowed key. With it, the topology emits the count once, when grace closes the window. That matches a consumer that expects append-only final results, including a lake sink that should not receive corrections. The buffer configuration I use is unbounded, so a load spike fails the job instead of dropping a window to save memory. I would rather page than publish a partial "final" count.

Q9. How should window-store retention relate to grace?

Retention has to cover the window size plus grace, or the store can discard state that grace still considers updateable. I add a margin for how long a restore might take. Retention is not the lateness policy. A longer retention does not make a late event count after grace. It lets the task restore windows that are still legitimately open. Changelog retention has to allow that restore to finish.

Q10. Why not replace the Spark tile with this topology as soon as the fixture matches?

A matching hour on one partition shows the definition is aligned. Production adds partition skew, commit lag, extractor deploy mistakes, and a second on-call path. I keep Spark as the served tile until a dual-run on production traffic shows an empty diff on closed windows for a period the product accepts, and until someone owns the Streams process. The morning table stays the historical count either way, because grace drops late events in both engines.

Q11. A future-dated event at tomorrow 12:00 arrives. What do you want the topology to do?

I want it rejected before it updates stream time. Once stream time jumps to tomorrow, today's windows close and ordinary events start dropping as late. The extractor or a step in front of the window should compare event_time to the record's ingest timestamp and send poison clocks to a reject topic. I do not "fix" the resulting drops by setting grace to 24 hours. That freezes the bad clock into the product.

Hands-On Exercises

Exercise: Match the five-minute tile on a fixture

Problem statement. Implement the Streams window so the fixture in this lesson emits the same closed counts Spark's definition would, and so a late event does not change a closed window.

Tasks.

1. Write ProgressEventTimeExtractor and a test that passes a record whose Kafka timestamp differs from event_time. Assert the extracted timestamp equals event_time.
2. Define a five-minute tumble with ten minutes of grace. Set store retention above size plus grace.
3. Suppress until close. Map the output to course_id, window_start, window_end, and count.
4. Pipe the fixture, including a closing event and a late event. Assert emission happens only after the closing event, and the late event does not increment the closed window.
5. Remove the closing event and show that a sleep does not emit. Put the closing event back.
6. In a paragraph, map grace and suppress to Spark's watermark and append mode, and state what course_day will still count.
7. Optional: a separate application.id for a ten-minute session gap on learner_id, with two bursts that merge and two that do not.

Expected deliverables. The extractor, the topology, the driver assertions, and the paragraph.

Evaluation rubric.

You pass when the late event is visible in the input and invisible in the closed count, and when you can point at the extractor as the reason broker time did not win.

Lesson Summary

The activity tile's clock is event_time, read by a TimestampExtractor, not the broker timestamp. Kafka Streams keeps windows open until per-task stream time passes the window end plus grace, and it does not move that clock when the source is idle. TimeWindows.ofSizeAndGrace(5 minutes, 10 minutes) is the Streams spelling of the Spark tile's tumble and watermark delay. suppress until close is the spelling of append mode's single emission. Hopping and session windows are other products and do not get written to the tile's topic. Late events increment a dropped-record metric and remain on the source for the morning restatement. A future timestamp is poison to stream time and should be rejected. One served tile is enough until a fixture and then a dual-run actually match.

CDC with Kafka and Debezium leaves windows behind and returns to enrollment, where the change is a delete, not a late event.

Further Reading

• Kafka Streams DSL windowing — tumbling, hopping, session, and sliding windows, grace, and suppress. Read the page for your release. The 24-hour default belongs to the deprecated factory.
• TimestampExtractor — the interface the tile uses to read event_time.
• KIP-633 grace and suppress behavior — why ofSizeAndGrace exists.
• Structured Streaming with Spark — watermark, append mode, and the ten-minute delay this lesson aligns to.
• Apache Flink Fundamentals — the same tile numbers on Flink's watermark syntax.
• /learn/kafka — CreateTime and LogAppendTime, the timestamps you are choosing not to window on.