Event Time and Processing Time

Three clocks

Module 3 left GatewayMeter with a running dollar total per tenant, usd_total, that grows forever. An invoice does not want a running total; it wants the cost of each minute. "Each minute" forces a question the running total never had to answer: a minute by which clock?

Upstream names three notions of time, and the course keeps its definitions exact:

event time: the time when an event occurred, as recorded by the device producing (or storing) the event

ingestion time: a timestamp recorded by Flink at the moment it ingests the event

processing time: the time when a specific operator in your pipeline is processing the event

For GatewayMeter the three are three different numbers on the same event. Event time is Event.ts — stamped by the gateway when the thing happened, milliseconds since epoch, never the Kafka append time. Ingestion time is whenever the Kafka source subtask read the record. Processing time is whatever the wall clock said on the TaskManager running tenant-cost when the record reached it.

The Kafka connector adds a fourth candidate to recognise and reject. Upstream's words: "By default, the record will use the timestamp embedded in Kafka ConsumerRecord as the event time." That is the broker's or the producer's clock, not the gateway's at the moment the request completed. ts is assigned explicitly, below.

Why processing time cannot be repeated

Upstream's definition of processing-time windows is precise about what they contain:

An hourly processing time window will include all records that arrived at a specific operator between the times when the system clock indicated the full hour. For example, if an application begins running at 9:15am, the first hourly processing time window will include events processed between 9:15am and 10:00am, the next window will include events processed between 10:00am and 11:00am, and so on.

And about why that is a problem:

Processing time is the simplest notion of time and requires no coordination between streams and machines. It provides the best performance and the lowest latency. However, in distributed and asynchronous environments processing time does not provide determinism, because it is susceptible to the speed at which records arrive in the system (for example from the message queue), to the speed at which the records flow between operators inside the system, and to outages (scheduled, or otherwise).

Read the three causes against your own job. Records arriving slowly: the gateway's retry buffer, the failure this module injects. Records flowing between operators: keyBy(tenant_id) shuffles every RequestRecord across the network, and a buffer that flushes a millisecond later moves a record across a minute boundary. Outages: the records that queue up during a TaskManager restart all land in the minute in which it came back.

The consequence for a bill: a minute of cost computed on processing time is the cost of the requests the operator happened to see in that minute, not the requests the tenant made in it. Run the same job over the same events tomorrow and the numbers move.

What event time buys, and what it costs

Upstream, again exact:

In event time, the progress of time depends on the data, not on any wall clocks. Event time programs must specify how to generate Event Time Watermarks, which is the mechanism that signals progress in event time.

In a perfect world, event time processing would yield completely consistent and deterministic results, regardless of when events arrive, or their ordering. However, unless the events are known to arrive in-order (by timestamp), event time processing incurs some latency while waiting for out-of-order events. As it is only possible to wait for a finite period of time, this places a limit on how deterministic event time applications can be.

Two things are being said. The first is the promise: "an hourly event time window will contain all records that carry an event timestamp that falls into that hour, regardless of the order in which they arrive, or when they are processed." At the minute scale: the minute a request belongs to is the minute the gateway stamped, not the minute the operator saw it. That is the accounting rule a tenant can audit from their own logs.

The second is the price. An event-time window cannot close on the wall clock, because the wall clock says nothing about whether more 10:00 requests are sitting in a retry buffer. It closes when event time has passed its end, and event time advances by watermarks. Upstream's definition, kept for module 5: "A Watermark(t) declares that event time has reached time t in that stream, meaning that there should be no more elements from the stream with a timestamp t' <= t." How that claim is generated, how it propagates through a parallel job, why an operator's event time is the minimum of its inputs, and what an idle partition does to it — all of that is module 5. Here you need two facts: event time advances by watermarks, and the simplest strategy upstream documents will do until the next module makes you replace it.

Which timestamp is the request's timestamp?

A RequestRecord carries two event times, started_ts and completed_ts, and the window assigner needs one. TenantCost uses completed_ts, for three reasons. Cost does not exist until completion: completion_tokens arrive with RequestCompleted, and a record windowed on started_ts would book a request in a minute during which its price was unknown. A request that starts at 10:00:58 and completes at 10:01:20 belongs, for the tenant, to the minute in which the response was delivered. And module 6's ModelLatency, keyed by model, also windows on completed_ts — a latency is only measurable at completion — so one event-time field per record keeps the two aggregates comparable minute for minute. started_ts is not wasted: module 6's abandoned-request timer measures from it.

Assigning the timestamp in PyFlink

Upstream: "In order to work with event time, Flink needs to know the events timestamps, meaning each element in the stream needs to have its event timestamp assigned. This is usually done by accessing/extracting the timestamp from some field in the element by using a TimestampAssigner." A WatermarkStrategy bundles the TimestampAssigner with a WatermarkGenerator, and the API takes the bundle.

Upstream prefers setting the strategy directly on the source, where it can watermark per partition. PyFlink allows that with a built-in strategy; the Kafka connector page marks only a custom WatermarkStrategy class as "(Not supported in PyFlink)" as of 2026-09. This module keeps the simplest shape — WatermarkStrategy.no_watermarks() at the source and the timestamp assigned after parse_event, upstream's second, less preferable option — so that time is the only new idea. Module 5 moves a built-in strategy onto the source and shows what the move buys.

from pyflink.common import Duration, WatermarkStrategy
from pyflink.common.watermark_strategy import TimestampAssigner


class EventTsAssigner(TimestampAssigner):

    def extract_timestamp(self, value, record_timestamp) -> int:
        # value is an Event; ts is already milliseconds since epoch
        return value.ts


events = env.from_source(source, WatermarkStrategy.no_watermarks(), "gateway-events") \
    .map(parse_event) \
    .assign_timestamps_and_watermarks(
        WatermarkStrategy.for_monotonous_timestamps()
                         .with_timestamp_assigner(EventTsAssigner()))

record_timestamp is the timestamp the record already carries — the Kafka one — and EventTsAssigner ignores it. Upstream's warning applies verbatim: "Both timestamps and watermarks are specified as milliseconds since the Java epoch of 1970-01-01T00:00:00Z." Event.ts was defined that way in module 1 for this line.

for_monotonous_timestamps() is "the simplest special case" upstream lists: it assumes "timestamps seen by a given source task occur in ascending order", so "the current timestamp can always act as a watermark". Your data does not satisfy that assumption — the retry buffer below is the proof — and module 5 opens with what to use instead.

Downstream, the RequestRecord that RequestAssembler emits inherits the event time of the RequestCompleted that produced it, so completed_ts is the record's timestamp without a second assigner.

TenantCost becomes a window

Module 3's usd_total: ValueState[float] was a running total. Now TenantCost is an AggregateFunction, and a window assigner decides which minute each RequestRecord is added to. The function is written once; only the assigner differs between the two versions.

Upstream's definition: an AggregateFunction "has three types: an input type (IN), accumulator type (ACC), and an output type (OUT)", with "a method for adding one input element to an accumulator", one "for creating an initial accumulator, for merging two accumulators into one accumulator and for extracting an output".

# Module paths for the window and function classes come from the PyFlink API reference
# (https://nightlies.apache.org/flink/flink-docs-release-2.3/api/python/).
from typing import Tuple
from pyflink.common.typeinfo import Types
from pyflink.datastream.functions import AggregateFunction, ProcessWindowFunction


class TenantCost(AggregateFunction):
    """accumulator = (usd, requests)"""

    def create_accumulator(self) -> Tuple[float, int]:
        return 0.0, 0

    def add(self, rec, acc: Tuple[float, int]) -> Tuple[float, int]:
        in_usd, out_usd = PRICE_PER_1K[rec.model]
        usd = rec.prompt_tokens / 1000 * in_usd + rec.completion_tokens / 1000 * out_usd
        return acc[0] + usd, acc[1] + 1

    def get_result(self, acc: Tuple[float, int]) -> Tuple[float, int]:
        return acc

    def merge(self, a: Tuple[float, int], b: Tuple[float, int]) -> Tuple[float, int]:
        return a[0] + b[0], a[1] + b[1]


class EmitTenantCostMinute(ProcessWindowFunction):

    def process(self, tenant_id, context: ProcessWindowFunction.Context, results):
        usd, requests = next(iter(results))
        window = context.window()   # the TimeWindow; its end is TenantCostMinute.window_end
        yield TenantCostMinute(tenant_id, window_end_of(window), usd, requests)

The ProcessWindowFunction only stamps tenant_id and window_end on the output — upstream's "incremental window aggregation with AggregateFunction" pattern, where results holds exactly one element, the pre-aggregated accumulator. Upstream documents the window's bounds on TimeWindow (start inclusive, end exclusive, with methods for querying both; the Python example reads context.window().start); the lab's starting code supplies window_end_of for the PyFlink object.

First version, processing time:

from pyflink.common import Time, Types
from pyflink.datastream.window import TumblingProcessingTimeWindows

tenant_cost = records \
    .key_by(lambda r: r.tenant_id) \
    .window(TumblingProcessingTimeWindows.of(Time.minutes(1))) \
    .aggregate(TenantCost(),
               window_function=EmitTenantCostMinute(),
               accumulator_type=Types.TUPLE([Types.DOUBLE(), Types.LONG()]),
               output_type=TENANT_COST_MINUTE_TYPE) \
    .uid("tenant-cost")

Second version, event time. One token changes:

from pyflink.datastream.window import TumblingEventTimeWindows

    .window(TumblingEventTimeWindows.of(Time.minutes(1))) \

Three upstream properties hold for both assigners. Windows are aligned to the epoch — "The first window will be 55 minutes long and close at 1:00" for a job started at 12:05 — so a minute window is always hh:mm:00.000–hh:mm:59.999 on either clock; the optional offset argument exists for time zones. A window "is created as soon as the first element that should belong to this window arrives", so "if there are no events in a given time frame, no results will be reported": a tenant idle for a minute gets no TenantCostMinute, not a zero. And the element a window emits carries a timestamp of "end timestamp - 1" on both clocks, which is what makes window_end a safe sink key in module 8.

The failure: a burst that is 30 seconds late

The gateway buffers events when Kafka rejects a produce and retries them. Suppose a broker hiccup holds one gateway instance's buffer from 10:00:40 to 10:01:15, so RequestCompleted events for requests that completed 10:00:40–10:00:59 reach tenant-cost at 10:01:15–10:01:25. Tenant acme, every request priced at $0.10.

Processing time bills acme $0.20 for 10:00 and $0.40 for 10:01. Event time bills $0.40 and $0.20, and the tenant's own logs agree with the second. The totals are identical, which is why nobody notices: this is an accounting error, a minute billed to the wrong minute — not a chargeback, not lost revenue. It surfaces the day the tenant disputes a rate-limit overage at 10:01 that their logs say never happened. And what decided the processing-time result was not the data — the same six records, the same six timestamps — but a broker hiccup and the wall clock of whichever TaskManager owned key acme. Re-run the job over a recording of the topic and the boundary lands somewhere else.

Now the honest half of the event-time column. Assignment is settled by completed_ts: r-103 and r-104 belong to the 10:00 window, whatever order they arrive in. Whether that window is still there to receive them at 10:01:17 is a different question. It is evaluated when the operator's event time passes 10:00:59.999, and with the monotonous strategy above, event time is the largest timestamp seen so far. When r-105 went through the assigner at 10:01:09, event time became 10:01:09 — past the window's end — so by the time r-103 arrives the window may already have fired with $0.20 in it, and upstream's default is blunt: "By default, when using event time windows, late events are dropped." Wrong minute or no minute: which you get is decided by how far event time was allowed to run ahead of the retry buffer, and that decision is the whole of module 5. Lab 4.2 has you watch it happen with this strategy, so module 5's fix lands on something you have seen.

The glossary line applies: r-103 is out of order the moment it is delivered behind r-105; it is late only if a watermark at or past 10:00:44 has already been seen. Late is relative to the watermark, never to the wall clock.

Replaying yesterday

The second thing event time buys is one upstream sentence: a program "might progress through weeks of event time with only a few seconds of processing, by fast-forwarding through some historic data already buffered in a Kafka topic". Point the event-time job at yesterday's gateway-events from the earliest offset and every RequestRecord lands in the minute its completed_ts names; event time runs from 00:00 to 23:59 as fast as the job can read, and the TenantCostMinute records come out identical to the live run — same minutes, same dollars, because assignment depends only on the data. That is how you re-bill a day after fixing a price-table bug, and how you test a new TenantCost against a recorded hour before trusting it with money.

Run the processing-time job on the same replay and you get one or two enormous minutes: the whole day's cost credited to whatever wall-clock minute the replay ran in. Upstream's list of processing-time limitations is the summary — it "can not correctly process historic data, can not correctly handle out-of-order data, results will be non-deterministic" — and the middle item is the retry burst you just traced.

If a TaskManager dies right now

Which numbers survive, which are recomputed, and which could be emitted twice?

Checkpointing is still off, so the module-1 answer stands: the job restarts with empty state and its source re-reads from the configured starting offsets (module 1's table), or fails without a restart strategy. Time changes what "empty" costs.

Which numbers survive: any TenantCostMinute a window already emitted and the at-least-once sink already wrote. Both versions are equal here.

Which are recomputed: for a processing-time window, nothing can be. Its accumulator on the dead TaskManager is gone, and the window itself was an accident of that machine's clock — the subtask that takes over key acme opens a fresh window at whatever boundary its clock is in, and records delayed by the restart are folded into the minute in which the job came back. Cost consumed and never emitted is revenue lost; cost delayed is booked to the wrong minute. An event-time window's accumulator is equally gone, but the window is reconstructible: replay the topic from before 10:00:00 and the same records land in the same 10:00 window by completed_ts, on any machine, under any wall clock. Module 7 makes that replay automatic and exact; this module is why it is possible.

Which could be emitted twice: if the restarted source re-reads records whose minute was already emitted, an event-time window fires a second TenantCostMinute for the same (tenant_id, window_end) — a chargeback; a processing-time window folds the same records into today's minute — an accounting error. Neither is fixed here. Note only that the event-time duplicate is identical to the first record, which is what makes it removable in module 8.

Try it

Go to labs/04-event-time-and-processing-time/. The compose cluster, the module-3 GatewayMeter and a recorded hour of gateway-events are the starting point.

• 4.1 Add both TenantCost windows to the job side by side, writing to two print sinks, and run the live generator for five minutes. Confirm the per-minute sums agree while events arrive promptly.
• 4.2 Inject the 30-second retry delay for one gateway instance. Reproduce the table above from the two outputs, and record whether the event-time window still receives the burst under for_monotonous_timestamps() — and what the operator's event time was when r-103 arrived.
• 4.3 Replay the recorded hour from the earliest offset into both jobs. Diff the event-time output against the live run; explain the processing-time output.
• 4.4 Switch EventTsAssigner to started_ts, replay again, and find a request whose cost moved minutes. State in one sentence each why TenantCost and module 6's ModelLatency both use completed_ts.

Lab page: lab-04-event-time-and-processing-time.html — the exercises, every source file, and the run-it-locally steps.