Windows and Timers

A window is state, a trigger and a timer

Module 4 turned TenantCost into a one-minute tumbling event-time window and module 5 explained who decides that the minute is over. This page opens the window up. There is nothing inside it you have not met: keyed state holding the elements or their aggregate, a trigger that says when the function runs, and an event-time timer the trigger registered for the window's end. When the watermark passes that timer, the trigger fires, the function runs, the result goes downstream.

Upstream's lifecycle, kept exact: "a window is created as soon as the first element that should belong to this window arrives, and the window is completely removed when the time (event or processing time) passes its end timestamp plus the user-specified allowed lateness". Two consequences. A minute in which acme sent nothing produces no TenantCostMinute — no element, no window. And a window does not vanish when it fires; it lives until the watermark passes end plus lateness, and until then a late element can change an answer that was already sent. That is this module's failure, and the first duplicate of the course.

Assigners

A window assigner "is responsible for assigning each incoming element to one or more windows". The four built-in ones, with the GatewayMeter job that each fits:

Time-based windows have "a start timestamp (inclusive) and an end timestamp (exclusive)", aligned to the epoch: [10:00, 10:01), not "the minute since the job started".

Sliding windows are the expensive one. Upstream's state-size rule: "Flink creates one copy of each element per window to which it belongs. Given this, tumbling windows keep one copy of each element ... In contrast, sliding windows create several of each element". Its example is a 24-hour window sliding every 15 minutes: "each event will be copied into 4 * 24 = 96 windows". ModelLatency at 5 minutes every 1 minute puts every RequestRecord into five windows, tolerable only because the window function keeps an aggregate, not the records.

Session windows close "when a gap of inactivity occurred", which sounds like abandoned-request detection. It is not: what we need is a fixed deadline after RequestStarted, which is a timer. Global windows need a custom trigger, since their default NeverTrigger "does never fire"; upstream suggests a ProcessFunction instead. GatewayMeter uses neither.

Window functions

Three kinds. A ReduceFunction combines two elements into one of the same type. An AggregateFunction has an input type, an accumulator type and an output type, with create_accumulator, add, get_result and merge. Both are applied incrementally, so the window's state is one accumulator. A ProcessWindowFunction gets an iterable of all the elements plus a Context with window(), current_watermark(), window_state() and global_state(); the price is that "Flink has to buffer all elements for a window internally before invoking the function".

The combination is the shape every GatewayMeter window uses: the AggregateFunction does the arithmetic as elements arrive, and the ProcessWindowFunction receives the single aggregated result with the window metadata — how TenantCostMinute gets its window_end. TenantCost from module 4 now reads:

late_tag = OutputTag("late-request-records", Types.PICKLED_BYTE_ARRAY())

class TenantCost(AggregateFunction):
    def create_accumulator(self):
        return 0.0, 0                                     # usd, requests

    def add(self, rec, acc):
        usd = (rec.prompt_tokens * PRICE_PER_1K[rec.model][0]
               + rec.completion_tokens * PRICE_PER_1K[rec.model][1]) / 1000
        return acc[0] + usd, acc[1] + 1

    def get_result(self, acc):
        return acc

    def merge(self, a, b):
        return a[0] + b[0], a[1] + b[1]

class EmitTenantCostMinute(ProcessWindowFunction):
    def process(self, key, context, results):
        usd, requests = next(iter(results))               # exactly one: the accumulator
        yield TenantCostMinute(key, context.window().end, usd, requests)

tenant_cost = records \
    .key_by(lambda r: r.tenant_id) \
    .window(TumblingEventTimeWindows.of(Time.minutes(1))) \
    .allowed_lateness(120_000)  # milliseconds \
    .side_output_late_data(late_tag) \
    .aggregate(TenantCost(),
               window_function=EmitTenantCostMinute(),
               accumulator_type=Types.TUPLE([Types.DOUBLE(), Types.LONG()]),
               output_type=Types.PICKLED_BYTE_ARRAY()) \
    .uid("tenant-cost")

late_records = tenant_cost.get_side_output(late_tag)

Two lines are new since module 5; the next sections explain them. Note also that the output record's own timestamp is "the maximum allowed timestamp of the processed window, which is end timestamp - 1", which lets a downstream window of the same size line up with this one.

Triggers

"A Trigger determines when a window (as formed by the window assigner) is ready to be processed by the window function." Every assigner has a default: "all the event-time window assigners have an EventTimeTrigger as default trigger. This trigger simply fires once the watermark passes the end of a window." Precisely, an arriving watermark "triggers computation of all windows where the maximum timestamp (which is end-timestamp - 1) is smaller than the new watermark". So [10:00, 10:01) fires when the watermark reaches 10:01:00.

A trigger has five methods — onElement, onEventTime, onProcessingTime, onMerge, clear — and the first three return a TriggerResult: CONTINUE, FIRE, PURGE or FIRE_AND_PURGE. "By default, the pre-implemented triggers simply FIRE without purging the window state." That matters below: a fire leaves the accumulator in place, so a second fire is a second, larger answer, not a delta.

Replacing the trigger replaces the timing entirely: "if you specify a CountTrigger for TumblingEventTimeWindows you will no longer get window firings based on the progress of time but only by count." An early trigger — fire every N elements, then at the watermark — would give the billing consumer a running total at the cost of several results per window: the same duplicate problem, on purpose. GatewayMeter keeps the default. The PyFlink shape includes .trigger(...) and upstream names four built-ins (EventTimeTrigger, ProcessingTimeTrigger, CountTrigger, PurgingTrigger), but shows no Python trigger implementation, only the Java Trigger class; treat custom triggers as a Java shape until you check the PyFlink API reference (as of 2026-09).

Allowed lateness and the side output

Module 5's definition, in upstream's words: "Lateness is defined relative to the watermarks. A Watermark(t) asserts that the stream is complete up through time t; any event following this watermark whose timestamp is ≤ t is late." By default, "late elements are dropped when the watermark is past the end of the window", and "its default value is 0". A dropped RequestRecord is a request no invoice ever sees. Revenue lost.

Allowed lateness "specifies by how much time elements can be late before they are dropped". Elements arriving after the watermark passed the window's end but before it passes end plus lateness "are still added to the window. Depending on the trigger used, a late but not dropped element may cause the window to fire again. This is the case for the EventTimeTrigger." Upstream names the two firings: "These firings are called late firings, as they are triggered by late events and in contrast to the main firing which is the first firing of the window."

The consequence, kept exact: "The elements emitted by a late firing should be treated as updated results of a previous computation, i.e., your data stream will contain multiple results for the same computation. Depending on your application, you need to take these duplicated results into account or deduplicate them."

The side output catches what lateness does not. side_output_late_data(late_tag) on the windowed stream, then get_side_output(late_tag) on the result, yields the elements discarded as late, in their input type — raw RequestRecords, not cost. "When the allowed lateness is greater than zero, only those events that are so late that they would be dropped are sent to the side output." In PyFlink, get_side_output "must be called"; otherwise the side output "will be output into the main stream", and a RequestRecord in tenant-cost fails the consumer's parser at best.

The failure: a second answer for the same minute

Module 5 set for_bounded_out_of_orderness(20 s), so the watermark trails the largest timestamp seen by 20 seconds. Module 4's gateway retry buffer delivers some events 30 seconds after they happened — beyond that bound. Here is one such event meeting acme's [10:00, 10:01) window. The times on the record boxes are event time (completed_ts); the axis along the bottom is the operator's watermark when each thing happens.

At 10:01:15 the tenant-cost topic holds two records for (acme, 10:01:00): $4.10 then $4.31. This is the first duplicate of the course, and it is by design — the window did exactly what allowed lateness asks. Flink does not retract the first record; it has no way to. The stream "will contain multiple results for the same computation", and the topic is append-only.

What happens next is the sink's and the consumer's problem, not the window's. A billing system that upserts by (tenant_id, window_end) overwrites $4.10 with $4.31 and bills the right amount. One that appends rows and sums them bills acme $8.41 for a $4.31 minute — a chargeback. Module 8 is about this topic: the two shapes of sink that make the duplicate harmless, and why it comes back by a second route when a TaskManager dies.

The record at 10:03:40 is the other case. Its window is gone; it reaches late_records, which the lab prints and a production job would write as an adjustment for the next invoice run. Not lost, but not in tenant-cost either.

The cost is state. "Flink keeps the state of windows until their allowed lateness expires." With zero lateness one minute of acme's windows is alive; with two minutes, three are. Because TenantCost is an AggregateFunction, each is one (usd, requests) pair. A bare ProcessWindowFunction would hold every RequestRecord of the last three minutes in keyed state, per tenant.

Evictors, briefly

An evictor removes elements "after the trigger fires and before and/or after the window function is applied" — CountEvictor, DeltaEvictor, TimeEvictor. Upstream's danger note: "Specifying an evictor prevents any pre-aggregation, as all the elements of a window have to be passed to the evictor before applying the computation." And, upstream's words: "Evictor is still not supported in Python DataStream API." GatewayMeter has no evictor; the Java shape, once, for recognition:

// Java only — no PyFlink equivalent as of 2026-09
stream.keyBy(...).window(...).evictor(CountEvictor.of(1000)).process(...);

ModelLatency: p50 and p99 over a sliding window

ModelLatencyWindow(model, window_end, p50_ms, p99_ms, n) needs percentiles, and a percentile is not a sum. The accumulator holds a bounded reservoir of latency samples plus the true count. It is an approximation: past K records the reservoir is a uniform sample, and the p99 of a thousand samples is a noisy estimate of the p99 of a hundred thousand. merge concatenates and truncates, cruder still; say so in the dashboard's legend.

class LatencySketch(AggregateFunction):
    K = 1000

    def create_accumulator(self):
        return [], 0                                      # samples, n

    def add(self, rec, acc):
        samples, n = acc
        n += 1
        if len(samples) < self.K:
            samples.append(rec.latency_ms)
        else:
            j = random.randrange(n)                       # reservoir sampling
            if j < self.K:
                samples[j] = rec.latency_ms
        return samples, n

    def get_result(self, acc):
        samples, n = acc
        s = sorted(samples)
        return s[int(0.50 * (len(s) - 1))], s[int(0.99 * (len(s) - 1))], n

    def merge(self, a, b):
        return (a[0] + b[0])[: self.K], a[1] + b[1]      # approximate

class EmitLatencyWindow(ProcessWindowFunction):
    def process(self, key, context, results):
        p50, p99, n = next(iter(results))
        yield ModelLatencyWindow(key, context.window().end, p50, p99, n)

model_latency = records \
    .key_by(lambda r: r.model) \
    .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1))) \
    .aggregate(LatencySketch(),
               window_function=EmitLatencyWindow(),
               accumulator_type=Types.PICKLED_BYTE_ARRAY(),
               output_type=Types.PICKLED_BYTE_ARRAY()) \
    .uid("model-latency")

Each record is added to five reservoirs. Five accumulators of at most K samples per model is the state; five copies of every record would have been the state without the aggregate.

Timers in a KeyedProcessFunction

A ProcessFunction gives you "events (stream elements)", "state (fault-tolerant, consistent, only on keyed stream)" and "timers (event time and processing time, only on keyed stream)". Upstream: "With event-time timers, the onTimer(...) method is called when the current watermark is advanced up to or beyond the timestamp of the timer, while with processing-time timers, onTimer(...) is called when wall clock time reaches the specified time. During that call, all states are again scoped to the key with which the timer was created". A window's trigger is a client of this same TimerService; now you use it directly.

Four facts govern timers. "The TimerService deduplicates timers per key and timestamp, i.e., there is at most one timer per key and timestamp." Coalescing follows: round the target down to whole seconds and there is at most one timer per key per second. "Flink synchronizes invocations of onTimer() and processElement()", so no locking around state. "Timers are fault tolerant and checkpointed along with the state of the application." And delete_event_time_timer(ts) needs the exact timestamp, so you keep it in state, or recompute it; "Stopping a timer has no effect if no such timer with the given timestamp is registered."

Module 3 left open behind for requests that never complete and promised a cleanup timer. This is its precise form: on RequestStarted, register an event-time timer at started_ts + 5 min; on RequestCompleted, emit, delete the timer, clear; in on_timer, if the request is still open, emit it as abandoned and clear. With the timer in place, lengthen module 3's 10-minute TTL on open to a backstop of hours. The TTL counts processing time and the timer counts event time, so during a watermark stall a short TTL expires the entry first; the completion then arrives to find no start, is dropped, and is never billed.

ABANDON_MS = 5 * 60 * 1000

class RequestAssembler(KeyedProcessFunction):          # keyed by request_id

    def open(self, runtime_context: RuntimeContext):
        # state name "open" per the spine; the attribute cannot be self.open
        self.open_state = runtime_context.get_state(
            ValueStateDescriptor("open", Types.PICKLED_BYTE_ARRAY()))

    def process_element(self, event, ctx):
        cur = self.open_state.value()
        if event.event_type == "RequestStarted":
            cur = OpenRequest(event.request_id, event.tenant_id, event.model,
                              event.ts, event.payload["prompt_tokens"])
            self.open_state.update(cur)
            ctx.timer_service().register_event_time_timer(cur.started_ts + ABANDON_MS)
        elif event.event_type == "RequestCompleted" and cur is not None:
            yield RequestRecord(cur.request_id, cur.tenant_id, cur.model,
                                cur.started_ts, event.ts, event.payload["latency_ms"],
                                cur.prompt_tokens, event.payload["completion_tokens"],
                                event.payload["status"])
            ctx.timer_service().delete_event_time_timer(cur.started_ts + ABANDON_MS)
            self.open_state.clear()

    def on_timer(self, timestamp, ctx):
        cur = self.open_state.value()
        if cur is not None and timestamp == cur.started_ts + ABANDON_MS:
            yield RequestRecord(cur.request_id, cur.tenant_id, cur.model,
                                cur.started_ts, None, None,
                                cur.prompt_tokens, 0, "abandoned")
            self.open_state.clear()

The timer fires when the watermark passes started_ts + 5 min — event time, so a stalled partition (module 5) stalls abandonment too, and a replayed backlog abandons requests exactly where it would have live. A duplicated RequestStarted re-registers the same timestamp for the same key; one timer per key and timestamp means one on_timer call. The timestamp == check is upstream's pattern for ignoring a stale timer. Abandoned records still carry prompt_tokens, so TenantCost bills the prompt, and they feed alerts.

The error-rate alert is the same machinery as a window built by hand, the way upstream's PseudoWindow does it: keyed by model, a MapState from minute end to (errors, requests), an event-time timer at each minute end registered when that minute's first event arrives; on_timer reads the entry, emits to alerts if errors / requests > 0.05, and removes it. Removing the entry in on_timer "is the equivalent of setting the allowedLateness to zero"; a late ModelError is dropped. Alerts tolerate that; invoices do not — which is why TenantCost stayed a window.

If a TaskManager dies right now

Window contents, accumulators, reservoirs and pending timers are all keyed state. Checkpointing is still off at this stage of the course, so today nothing survives: every open window's partial sum, every abandoned-request timer and the open state behind it are gone; the job restarts with empty state (or fails, with no restart strategy), and the minutes in flight are never billed.

Module 7 turns checkpointing on, and then "in case of a failure recovery or when starting an application from a savepoint, the timers are restored." Window state comes back with them. An event-time timer for an abandoned request that was due during the outage fires when the watermark catches up after recovery, not on the wall clock; a checkpointed processing-time timer "that [was] supposed to fire before [its] restoration, will fire immediately." Emitted twice: any window that fired between the last checkpoint and the crash fires again after recovery — the same (tenant, window_end) duplicate you saw today, by a different route. The sink that absorbs one absorbs the other. Module 8.

Try it

Lab 6 is in labs/06-windows-and-timers/:

• 6.1 Send a timestamp-controlled trace and advance event time until the first one-minute TenantCostMinute window fires. Record its (tenant_id, window_end) and value.
• 6.2 Deliver one more completion for that minute after its first firing but within the two-minute allowed lateness. Compare the corrected value with the first emission; this is a late correction, not a recovery duplicate.
• 6.3 Advance past the cleanup boundary, then send another event for that minute. Find its LATE <request_id> side-output line and explain the state cost of a longer allowed-lateness interval.
• 6.4 Leave a request open and advance the watermark beyond its five-minute event-time timer. Confirm one status="abandoned" record and contrast this rule with module 3's processing-time TTL.

Lab page: lab-06-windows-and-timers.html — the exercises, every source file, and the run-it-locally steps.