The Machine

Stream processing, before any Flink

If Flink is new to you, start here. The rest of this page, and of the course, assumes this picture.

The problem. An LLM gateway serves requests for many tenants across many models, and every request emits a few events along the way: started, tokens generated, completed, failed. The events never stop coming. Somebody has to turn them into answers: what tenant acme spent between 10:00 and 10:01, whether a model's p99 latency just doubled, which tenant's error rate crossed 5% in the last minute.

There are two ways to get those answers. Batch is the traditional one: store the events, and once the day is over, run a job across the complete set. It is simple because the input is finished before the job starts, and it is a day late; an alert that fires tomorrow is not an alert. Stream processing reads each event as it arrives, remembers what it must, and keeps its answers current for as long as events arrive — for a gateway, forever.

Upstream's definitions of the two are the ones to keep, exact:

Batch processing is the paradigm at work when you process a bounded data stream. In this mode of operation you can choose to ingest the entire dataset before producing any results, which means that it is possible, for example, to sort the data, compute global statistics, or produce a final report that summarizes all of the input.

Stream processing, on the other hand, involves unbounded data streams. Conceptually, at least, the input may never end, and so you are forced to continuously process the data as it arrives.

Why a loop over Kafka is not enough. For a demo, stream processing is a consumer loop: poll a record, update a dict, print. It stops being enough once the answer has to be right, because four problems appear that a batch job never faces, and each one is a part of this course:

Upstream's own introduction names the same core, exact: This training focuses on four critical concepts: continuous processing of streaming data, event time, stateful stream processing, and state snapshots.

What Flink is. Apache Flink is an engine that solves those problems once, so that each job does not have to. You write a program that describes a dataflow: a source that events come from (here, Kafka), operators that transform them (parse, group by key, count per minute), and a sink that results go to. You submit it to a Flink cluster. The cluster runs it in parallel across machines, keeps each operator's memory as state beside the code that uses it, follows event time from the timestamps in the data, and periodically snapshots all of that state together with how far into the input it has read. Upstream, exact: Flink is able to provide fault-tolerant, exactly-once semantics through a combination of state snapshots and stream replay. In this course's wording that is exactly-once state semantics; what it takes for a bill to be exactly-once end to end is module 8.

Two things Flink is not. It is not where the events live: Kafka stores gateway-events, and while a Flink job is down the events wait in Kafka, for as long as the topic retains them. And it is not a workflow engine: Temporal, the sibling course, makes long-running, step-by-step processes durable one execution at a time, while Flink computes continuously over high-volume streams. Appendix A1 compares the two.

The words this page uses. Each gets its full treatment later; this is enough to read module 1.

This module starts at the bottom: the processes that run a job, and what happens when one of them dies before any of those guarantees is switched on.

Kill a TaskManager. The job stops.

Here is the experiment this module is built around. A Flink job reads events from Kafka and prints a count per model every ten seconds, spread across four worker containers. You docker kill one of them. The count stops. In the Web UI the job reads FAILED, and nothing restarts it.

That is the honest starting point, and it is the opposite of the sibling course's first hour, where a killed Worker cost nothing because the Service held the program's history. Flink holds nothing for you until you ask it to. This module is the machine as it is before you ask: which processes exist, where your code runs, what a job is once it is running, and what the JobManager does when a worker disappears. Module 7 turns checkpointing on; doing module 1 without it is how you learn what the checkpoint is for.

The program this course builds

Every module works on one program, GatewayMeter: the real-time analytics job behind an LLM gateway, the proxy that fronts many model providers for many tenants. Every request the gateway serves emits events; the job turns them into per-tenant cost, per-model latency, error rates and agent-step accounting, continuously, with the correctness guarantees a bill requires.

The input is one Kafka topic, gateway-events, with 8 partitions. The gateway keys records by request_id, so one request's events share a partition, but there is no order across partitions. Every record is one envelope:

Event(event_type, request_id, tenant_id, model, ts, payload)

ts is event time — milliseconds since epoch, stamped by the gateway when the thing happened, never the Kafka append time. payload carries the per-type fields of the six event types:

parse_event(json) -> Event is a plain map. By module 12 the outputs are three Kafka topics: tenant-cost (one TenantCostMinute per tenant per minute), model-latency (p50/p99 per model) and alerts. tenant-cost is what the billing system reads, and that is why this course has a dollar sign in it: a minute of cost billed to the wrong minute is an accounting error; a minute billed twice is a chargeback; a minute never billed is revenue lost. Today's failure is the crude ancestor of all three — a count that is either never produced or produced twice.

Module 1 builds the smallest stage: read the topic, parse, count events per model in a 10-second processing-time window, print. Keys are module 2, state is module 3, time is module 4.

Everything is a dataflow

Upstream's words, kept exact: In Flink, applications are composed of streaming dataflows that may be transformed by user-defined operators. These dataflows form directed graphs that start with one or more sources, and end in one or more sinks. Usually one transformation in your program is one operator in the graph; sometimes it is several.

Boundedness, from the top of this page, has a precise definition for jobs too: Boundedness is a property of a data source that tells us whether all the input coming from that source is known before execution or whether new data will show up, potentially indefinitely. A job, in turn, is bounded if all its sources are bounded, and unbounded otherwise. gateway-events is unbounded; the Kafka source never stops until Flink job fails or is cancelled, and GatewayMeter runs in STREAMING execution mode for the whole course.

The classic first program is word count, named here once and not again: stage 1 has exactly its shape — source, map, key, window, count, print — except that the words are model names and the count will shortly be money.

Who runs your code?

Upstream: The Flink runtime consists of two types of processes: a JobManager and one or more TaskManagers. The client is a third party, not part of the runtime and program execution, but is used to prepare and send a dataflow to the JobManager; afterwards it may disconnect (detached) or stay for progress reports (attached).

JobManager. It decides when to schedule the next task (or set of tasks), reacts to finished tasks or execution failures, coordinates checkpoints, and coordinates recovery on failures. Three components in one process. The ResourceManager manages task slots, the unit of resource scheduling; in a standalone setup — which a docker compose cluster is — it can only distribute the slots of available TaskManagers and cannot start new TaskManagers on its own. Keep that sentence for the failure walkthrough. The Dispatcher is the REST endpoint you submit to; it starts a JobMaster per job and runs the Web UI. The JobMaster manages one JobGraph; several jobs means several JobMasters. There is always at least one JobManager; HA setups have a leader and standbys (module 11).

TaskManagers. Also called workers: they execute the tasks of a dataflow, and buffer and exchange the data streams. Each is a JVM process, and its capacity is counted in task slots — the smallest unit of resource scheduling in a TaskManager; the slot count indicates the number of concurrent processing tasks. A slot is a fixed share of the TaskManager's managed memory, and only that: no CPU isolation happens here; currently slots only separate the managed memory of tasks. Slot ≠ CPU core.

The lab cluster is the upstream session-mode compose file with one change: one jobmanager, four taskmanager containers, each with taskmanager.numberOfTaskSlots: 2 (the option defaults to 1). Eight slots. The architecture page's warning about session clusters is the experiment: if one TaskManager crashes, then all jobs that have tasks running on this TaskManager will fail.

Job graph, execution graph

The DataStream calls you make run nothing. Upstream: All Flink programs are executed lazily: When the program's main method is executed, the data loading and transformations do not happen directly. Rather, each operation is created and added to a dataflow graph. The operations are actually executed when the execution is explicitly triggered by an execute() call on the execution environment. That graph is the JobGraph — the glossary's logical graph, operators for nodes and streams for edges. env.execute() packages it and sends it to the JobManager; without the call, nothing runs.

The JobManager transforms the JobGraph into an ExecutionGraph. The ExecutionGraph is a parallel version of the JobGraph: For each JobVertex, it contains an ExecutionVertex per parallel subtask. An operator with a parallelism of 100 will have one JobVertex and 100 ExecutionVertices. Four words to keep apart: an operator is a node of the logical graph; a subtask is one parallel instance of it; parallelism is an operator's number of subtasks; a task is what a slot runs — exactly one parallel instance of an Operator or Operator Chain. Consecutive operators with no repartitioning between them are chained into one task on one thread, so records pass between them without serialization or the network.

An edge between operators is one-to-one — subtask 1 downstream sees exactly what subtask 1 upstream produced, in order — or redistributing, as under a key_by, where every sender fans out to every receiver and the ordering among the elements is only preserved within each pair of sending and receiving subtasks.

Slots are shared: By default, Flink allows subtasks to share slots even if they are subtasks of different tasks, so long as they are from the same job. The result is that one slot may hold an entire pipeline of the job, so a Flink cluster needs exactly as many task slots as the highest parallelism used in the job. Stage 1 on the lab cluster, at parallelism 4:

Where the chain boundaries fall in a PyFlink job — whose Python functions run in a separate Python process beside the JVM — is something to read off the Web UI in exercise 1.1 rather than assume from the Java picture.

GatewayMeter, stage 1

The whole job. Builder calls follow the upstream Python tabs; build_job(env) is the entry point every later module extends.

# gatewaymeter/job.py — stage 1 (module 1)
# Module paths for the connector, serialization and window classes come from the
# PyFlink API reference (https://nightlies.apache.org/flink/flink-docs-release-2.3/api/python/).
import json
from typing import NamedTuple

from pyflink.common import Configuration, Duration, WatermarkStrategy
from pyflink.common.serialization import SimpleStringSchema
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer
from pyflink.datastream.window import TumblingProcessingTimeWindows

BROKERS = "kafka:9092"
TOPIC = "gateway-events"          # 8 partitions, keyed by request_id at the gateway


class Event(NamedTuple):          # travels pickled; downstream code reads fields by name
    event_type: str
    request_id: str
    tenant_id: str
    model: str
    ts: int                       # event time, ms since epoch, stamped by the gateway
    payload: dict                 # per-type fields, parsed once here


EVENT_TYPE = Types.PICKLED_BYTE_ARRAY()   # the NamedTuple arrives intact, so event.tenant_id works downstream


def parse_event(raw: str) -> Event:
    d = json.loads(raw)
    return Event(d["event_type"], d["request_id"], d["tenant_id"],
                 d["model"], int(d["ts"]), d.get("payload", {}))


def build_job(env: StreamExecutionEnvironment):
    source = KafkaSource.builder() \
        .set_bootstrap_servers(BROKERS) \
        .set_topics(TOPIC) \
        .set_group_id("gatewaymeter") \
        .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
        .set_value_only_deserializer(SimpleStringSchema()) \
        .build()

    events = env.from_source(source, WatermarkStrategy.no_watermarks(), "gateway-events") \
        .map(parse_event, output_type=EVENT_TYPE)

    counts = events \
        .map(lambda e: (e.model, 1), output_type=Types.TUPLE([Types.STRING(), Types.INT()])) \
        .key_by(lambda t: t[0]) \
        .window(TumblingProcessingTimeWindows.of(Duration.of_seconds(10))) \
        .reduce(lambda a, b: (a[0], a[1] + b[1]))

    counts.print()                # to the TaskManager logs, prefixed by subtask: "3> ('gpt-x', 412)"
    return counts


if __name__ == "__main__":
    config = Configuration()      # exercise 1.3 adds the restart strategy here
    env = StreamExecutionEnvironment.get_execution_environment(config)
    env.set_parallelism(4)
    build_job(env)
    env.execute("GatewayMeter")   # the client submits the JobGraph here; nothing ran before this line

Three notes, then leave the rest to later modules.

WatermarkStrategy.no_watermarks() is deliberate. The window is processing-time: a bucket closes when the TaskManager's wall clock crosses a ten-second boundary, and ts rides along unread. Module 4 is what that costs. The Kafka builder's Python surface is narrower than Java's — Currently, PyFlink only supports set_value_only_deserializer to customize deserialization of the value of a Kafka record, and custom offsets initializers and a custom WatermarkStrategy on the Kafka source are marked not supported in PyFlink — so parsing is a map after the source rather than a deserializer inside it.

key_by(lambda t: t[0]) is there only because "per model" needs it: a keyed window keeps one count per key, and the key decides which subtask holds it. Choosing keys, and what breaks when you drop one, is module 2; what the window accumulates between boundaries is state, and keeping state bounded is module 3.

print() writes to the TaskManager logs, each line prefixed by the subtask that produced it — in the lab, docker compose logs taskmanager.

The first durability fact

Upstream, kept exact: By default, checkpointing is disabled. To enable checkpointing, call enableCheckpointing(n) on the StreamExecutionEnvironment, where n is the checkpoint interval in milliseconds. In the configuration file the switch is execution.checkpointing.interval: To enable checkpointing, you need to set this value larger than 0. Stage 1 does neither.

The second fact rides on the first. The restart strategy is chosen by restart-strategy.type, and when you set nothing: If checkpointing is not enabled, the no restart strategy is used. If checkpointing is activated and the restart strategy has not been configured, the exponential-delay restart strategy and the default values of exponential-delay related config options will be used. The types are fixed-delay, exponential-delay, failure-rate and none; their per-type defaults live in the generated tables on the upstream page and are not repeated here (as of 2026-09; see https://nightlies.apache.org/flink/flink-docs-release-2.3/docs/ops/state/task_failure_recovery/). The Python shape upstream gives, which exercise 1.3 uses:

config = Configuration()
config.set_string('restart-strategy.type', 'fixed-delay')
config.set_string('restart-strategy.fixed-delay.attempts', '3')      # number of restart attempts
config.set_string('restart-strategy.fixed-delay.delay', '10000 ms')  # delay
env = StreamExecutionEnvironment.get_execution_environment(config)

Exponential delay, which upstream strongly recommends, is Still not supported in Python API for programmatic configuration; it goes in the cluster's configuration file. Module 7 chooses fixed-delay instead (three attempts, ten seconds apart), which PyFlink can set in code.

So a stage-1 job that loses a task does not restart, and a job you make restart has no snapshot to restore. The Kafka source re-initialises from its KafkaOffsetsInitializer, and committed offsets are not what recovers it: Kafka source does NOT rely on committed offsets for fault tolerance. Committing offset is only for exposing the progress of consumer and consuming group for monitoring. Without checkpointing it commits only through the consumer's own enable.auto.commit / auto.commit.interval.ms. Three behaviours after a restart; stage 1 uses the first:

In every row the window that was open at the kill is gone: its partial count lived in the dead TaskManager's memory and nowhere else. Module 7 collapses this table to one line — offsets and window contents restore together from checkpoint k — and module 8 is about the row that still bites afterwards: output already written before the failure.

The failure: docker kill one TaskManager

Stage 1 on the lab cluster, no restart strategy. Times are upstream defaults and marked where they are configuration.

With fixed-delay, 3 attempts, 10 s: after failing the job enters restarting, waits ten seconds, and once the job has been completely restarted, it will reach the created state, then running. The JobMaster redeploys all subtasks into the six remaining slots (four are needed), the Kafka enumerator hands the eight partitions to four new source subtasks, the new window subtasks start with empty counts, and the source starts where earliest() says. From outside: a ten-second gap, then every count you have ever seen scrolls past a second time. docker compose up -d taskmanager re-registers the slots; nothing running moves onto them. A second kill before that still leaves four slots; a third leaves two, and the job waits for slots it cannot get.

Two things to read off the real logs rather than believe: how fast the JobManager noticed (the connection reset may beat the heartbeat timeout), and where the four pipelines landed after the restart.

If a TaskManager dies right now

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

Survive: none. Stage 1 has no checkpoint, so no window count and no source offset exists anywhere except in the memory of the process that died — and the other three TaskManagers' tasks are cancelled with the job.

Recomputed: whatever the restarted source re-reads. With earliest(), every event in the topic; with latest(), nothing; with committed offsets, whatever followed the last auto-commit. Recomputation is possible at all only because Kafka is replayable; a socket source would offer nothing to recompute from.

Emitted twice: with earliest(), every count printed before the kill is printed again — on tenant-cost, the chargeback case. With latest(), the events between kill and restart are never counted, and the window open at the kill is lost with its contents — the revenue-lost case. Neither gives the third option, counted once, because the job has no memory of what it already emitted. Module 7 gives it that memory inside Flink; module 8 extends it to the sink.

Try it

Go to labs/01-the-machine/. Everything runs against labs/docker-compose.yml: one JobManager, four TaskManagers with two slots each, Kafka, Web UI on http://localhost:8081.

• 1.1 Submit GatewayMeter a second time while the first run is still active. Read slots-available in the Web UI, then try a third submission and explain the result in slots rather than job counts.
• 1.2 Change WINDOW_SECONDS to 60 and resubmit. Stop the traffic after a count appears; explain why the final partial processing-time window does not fire just because input stopped.
• 1.3 Remove env.add_python_file(...) (or the compose PYTHONPATH) and resubmit. Find the ModuleNotFoundError in the Python worker and explain why a successful submission did not mean the TaskManagers could import common.
• 1.4 With the job running, kill the Kafka container, then restore it. Record how the source responds and contrast that with the TaskManager failure you just observed in the main lab walkthrough.

The TaskManager kill and its NoRestartBackoffTimeStrategy evidence are in the lab walkthrough before these four numbered exercises. The fixed-delay restart and count-reconciliation experiments above are extensions for a later verified lab pass, not steps Lab 1 currently asks you to run.

Lab page: lab-01-the-machine.html — the exercises, every source file, and the run-it-locally steps.