"Streaming SQL" — Apache Flink I

Skim Card

One thing to remember: a streaming query is a table that keeps changing; know whether it appends or updates, and choose a sink that can store that.

• EXPLAIN CHANGELOG_MODE before choosing a sink: windows, dedup and temporal joins insert; running totals and Top-N update.
• Join a stream to a versioned table as of event time; a join to today’s prices mispriced 2,338 of 58,119 lines.
• Deduplicate by arrival unless you can afford to wait: keep-first on event time silently dropped 84 late baskets in this replay.
• Windows are final and append-only, so late rows vanish; a running total keeps them and makes the sink upsert.
• An Iceberg upsert table paid 58,119 row writes for 775 live rows; mini-batch cut that to 21,594.
• Dedup state must outlive the re-send window: a replay started on 15 August counted a UK basket first sent on the 14th.
• Reconcile the stream with batch: 774 of 775 store-days matched exactly; the 775th is that re-send.

Use …	When …
Window TVF into an append-only sink	Dashboard tiles that can drop late rows, or a side path fixes them
Updating aggregate into an upsert sink (Iceberg v2 with a key, upsert-kafka)	Every record must count, late ones included
Temporal join on a versioned table	Attributes must be the ones in force when the event happened

Verified against: Apache Flink 2.2.1 (Java 17) with the Kafka SQL connector 5.0.0-2.2, Apache Iceberg 1.12.0 (iceberg-flink-runtime-2.2) with Hadoop client 3.4.3, Apache Polaris 1.8.0, Apicurio Registry 3.3.3, Apache Kafka 4.4.0, PyIceberg 0.12.0, DuckDB 2.0.0 · vendor docs · 30 October 2026

Jump to: Concepts · Hands-On · Landscape · Design Drill

Story Opening

The stock-out alerts need sales by store as they happen, not after the nightly run. Rohan wrote it in Flink SQL like a Postgres report: keep the first copy of each basket by its timestamp, join each line to the price table, sum by store and hour.

The hourly tile came out 84 baskets short of the till log, and some lines carried today’s price, not what the till charged in August.

“Nothing is wrong with the SQL,” Ingrid said. “It’s the wrong mental model. Your query never finishes: it’s a table that keeps changing, and each operator in it decides what to do with time.”

The Quick Map

Concepts, with use cases

Of Part 2’s five promises, this part strengthens timeliness: sales by store within seconds of the till, keeping the uniqueness Part 21 won in batch. It builds on Part 26’s watermarks and Part 27’s topics.

Flink’s moving parts

A Flink cluster is a JobManager, which accepts jobs, schedules them and hands out task slots, and one or more TaskManagers, which run them. The planner turns SQL into operators, chains neighbours into tasks (one thread each) and runs each task as parallel subtasks. A slot separates managed memory only (“no CPU isolation happens here”); with slot sharing, a job needs as many slots as its highest parallelism.

What breaks without the picture. Teams add TaskManagers while one hot key keeps a single subtask busy. Larkspur: one JobManager and one 2-slot TaskManager; this job’s eleven tasks run at parallelism 1 and need one slot. Elsewhere: an ad exchange runs bid aggregation at parallelism 256, one subtask per Kafka partition.

Dynamic tables and changelogs

Flink SQL treats a stream as a dynamic table; a continuous query over it “never terminates and produces dynamic results — another dynamic table”, equal at any moment to the batch query over the input so far. The result leaves as a changelog: +I (insert), -U and +U (the row before and after an update), -D (delete). An append-only query emits only +I. An updating query revises rows it already emitted, either as a retract stream (-U then +U) or, given a unique key, an upsert stream (just the new row, keyed).

What breaks if you ignore it. Inside Flink, nothing: the planner refuses an updating query into an insert-only sink (Step 4). Outside, a reader that treats a changelog topic as plain rows counts every intermediate total. Larkspur: the hourly window appends; the running total per store and day updates. Elsewhere: a sports-betting live table where every goal rewrites the standings.

Windows, briefly

Window table-valued functions (TUMBLE, HOP, CUMULATE, SESSION; Part 26) emit each window once, when the watermark passes its end: append-only, and a later row is dropped. Larkspur: units out and sales per store per UTC hour. Elsewhere: dropped calls per cell per minute at a mobile network.

Joins in a stream

• Regular join: both sides kept in state “forever”; a state time-to-live bounds it but “might affect the correctness of the query result”.
• Interval join: an equi-join with a time bound on both sides (a click and the order within 30 minutes); append-only inputs, state expires with time.
• Temporal join (FOR SYSTEM_TIME AS OF): each row of the stream is joined “with the build side’s version at the time specified by the time attribute” of a versioned table, one with a primary key and an event-time attribute. It is the streaming form of Part 4’s as-of join to an SCD Type 2 dimension.
• Lookup join: queries an external store (a database) per row at processing time, so the answer depends on when the job ran.

What breaks with the wrong one. A regular join to prices rewrites old results when a price changes; a lookup join replayed next month gives different answers. Larkspur: each basket line joins the shelf price in force when the till scanned it. Elsewhere: a broker converts each trade at the FX rate quoted at trade time, never the closing rate.

Deduplication and Top-N

ROW_NUMBER() OVER (PARTITION BY key ORDER BY time_attr) = 1 is deduplication: ASC keeps the first row, DESC the last, ordered by processing time (first to arrive) or event time (first to happen, so wait for the watermark). rn <= N with any other order is Top-N, which updates as values change.

What breaks without it. Every till re-send counts twice, the first root cause of the 0.7% UK gap. Larkspur: the first copy of each basket_id by arrival, as Part 21 accepts the first upload (keyed by till, copies share a partition and arrive in upload order); the top three stores per country. Elsewhere: a smart-meter platform drops readings a gateway re-sent after reconnecting.

Sinks that append, and sinks that upsert

Flink’s kafka connector appends; upsert-kafka writes each change keyed by the primary key and a delete as a tombstone (a null value), so a compacted topic converges to the table. Iceberg’s Flink sink appends, or on a format v2 table with a primary key and write.upsert.enabled upserts with delete files, committing after each completed checkpoint (Part 26): the checkpoint interval is the lake’s commit interval. SQL uses FlinkSink; the SinkV2 IcebergSink “is currently an experimental feature” and isn’t used here.

What breaks with the wrong sink. Read as plain rows, the top-stores topic shows every store that was ever in the top three; an upsert table nobody compacts makes every read merge more delete files. Larkspur: lines and hourly windows append into Iceberg; the daily total upserts; the top-stores board is an upsert topic. Elsewhere: an e-commerce site keeps live stock per SKU in a compacted topic that the product pages read.

Under the Hood

What each dedup does with a late basket

Flink 2.2.1’s keep-first-on-event-time operator holds the earliest copy per key and emits it by timer when the watermark passes; a row below the watermark is ignored, counted only in a late-records metric. The processing-time variant emits the first arrival at once and keeps a flag per key: nothing is late, nothing forgotten unless a state TTL says so.

How the temporal join picks a version

TemporalRowTimeJoinOperator buffers both sides until the combined watermark passes (“triggered by a watermark from the left and right sides”), joins each line to the latest version at or before its event time, then deletes versions older than the watermark except the newest. So a line arriving behind the watermark is unmatched if the price changed between its sale and the watermark (null in a LEFT JOIN, dropped in an inner one). That is from the source; no late line here straddled a change.

What an upsert costs in Iceberg

For each +I or +U, Flink’s Iceberg writer deletes the key, then writes the row (-U is skipped). A key already written in this commit gets a position delete; otherwise an equality delete, one per key per commit, even on an empty table. Readers merge them until compaction rewrites the files (Part 7). A total updated on every line writes a row per line.

Hands-On

The code is in the every-record-counts folder of the companion-workspaces repository (tag every-record-counts/part-28). make up-flink-lake starts core, Kafka, Apicurio and Flink 2.2.1, because no released Kafka connector supports 2.3 (re-checked 30 October); Connect, Spark and Trino stay stopped. Peak container memory was 3.84 GiB (TaskManager 1.00, JobManager 0.91).

make up-flink-lake
uv run python -m scripts.part28_flink setup            # job principal, price topic, empty tables
./gradlew -q :kafka-backbone:run --args="prices"       # the price versions (Java 25)
uv run python -m scripts.part28_flink run              # POS + prices -> three Iceberg tables
uv run python -m scripts.part28_flink explain
uv run python -m scripts.part28_flink dedup
uv run python -m scripts.part28_flink asof
uv run python -m scripts.part28_flink reconcile
uv run python -m scripts.part28_flink changelog
uv run python -m scripts.part28_flink errors
uv run python -m scripts.part28_flink run --mini-batch

Step 0: Iceberg in the Flink image

flink/Dockerfile adds the Iceberg Flink runtime and AWS bundle 1.12.0 and, because the runtime needs Hadoop even for a REST catalog (Troubleshooting), Hadoop’s client jars 3.4.3, all pinned by checksum.

Step 1: a principal for the job, and a price topic

The job writes as its own Polaris principal, flink, with storage keys vended per table (Part 9), and may only list namespaces, create, list, read and write tables in stream, and read that namespace’s properties (Flink looks them up before a read). No drop: setup drops tables as root.

The price feed is a lab replay: a Java producer writes the raw zone’s 7,075 price versions to larkspur.catalog.price-changes, keyed by SKU, kept forever rather than compacted, because an as-of replay needs every old price. In production it would be the pricing service’s events or CDC on its database (The Log Never Lies); a Debezium changelog declared with a primary key is a versioned table without the dedup view.

Step 2: the sources and the views

flink/sql/part28/tables.sql declares the POS topic as Part 27 read it (read_committed, which the connector doesn’t default to) with a 5-minute watermark, the price topic with WATERMARK FOR effective_from, and three views:

-- Dedup: the first copy of each basket to arrive. arrived_at is PROCTIME().
CREATE VIEW baskets AS
SELECT * FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY basket_id ORDER BY arrived_at ASC) AS rn FROM pos_basket_events)
WHERE rn = 1;

-- A versioned table from an append-only topic: key (sku, country), version time effective_from.
CREATE VIEW price_versions AS
SELECT sku, country, currency, price_minor, effective_from FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY sku, country ORDER BY effective_from DESC) AS rn FROM price_changes)
WHERE rn = 1;

-- One row per basket line, with the shelf price in force when the till sold it.
CREATE VIEW sales_lines AS
SELECT b.event_id, b.basket_id, b.store_id, …, b.event_ts, l.sku, l.qty,
       l.qty * l.unit_price_minor - l.discount_minor AS net_minor,
       p.price_minor AS list_price_minor, p.effective_from AS price_from
FROM baskets AS b
CROSS JOIN UNNEST(b.lines) AS l (line_no, sku, qty, unit_price_minor, discount_minor, promo_id)
LEFT JOIN price_versions FOR SYSTEM_TIME AS OF b.event_ts AS p
  ON p.sku = l.sku AND p.country = SUBSTRING(b.store_id, 1, 2);

price_versions is Flink’s documented versioned-view shape: key from PARTITION BY, version time from ORDER BY. sales_lines also derives the store-local business_date (Part 5).

Step 3: one job, three Iceberg tables

sales_to_lake.sql sets a 10-second checkpoint interval, creates the tables (the daily one format v2 with PRIMARY KEY (store_id, business_date) NOT ENFORCED and 'write.upsert.enabled' = 'true') and runs three INSERTs as one statement set:

EXECUTE STATEMENT SET
BEGIN
INSERT INTO lake.`stream`.store_sales_hourly
SELECT store_id, currency, window_start, window_end, COUNT(DISTINCT basket_id), COUNT(*), SUM(qty), SUM(net_minor)
FROM TABLE(TUMBLE(TABLE sales_lines, DESCRIPTOR(event_ts), INTERVAL '1' HOUR))
GROUP BY store_id, currency, window_start, window_end;
INSERT INTO lake.`stream`.store_sales_daily
SELECT store_id, business_date, MAX(currency), COUNT(DISTINCT basket_id), SUM(qty), SUM(net_minor)
FROM sales_lines
GROUP BY store_id, business_date;
END;

units_out is stock movement as the tills see it: units leaving the shelf, which Part 29 turns into stock levels. The sources read to the topic’s end (scan.bounded.mode), so the job finishes:

job 59721f17 FINISHED in 4.0 s; checkpoints completed 1
table                      snapshots  written files eq deletes pos deletes  key
stream.pos_sales_lines             1   58,119     3          0           0  -
stream.store_sales_hourly          1    7,295     1          0           0  -
stream.store_sales_daily           1   58,119     1        775      57,344  business_date,store_id

One commit per table, with flink.max-committed-checkpoint-id 9223372036854775807 (Long.MAX, Iceberg’s end of input). The daily table: 58,119 rows written, 775 equality deletes, 57,344 position deletes, 775 live rows. One cold run (8.7 s) crossed a periodic checkpoint and committed two tables twice; the others finished first.

Flink’s REST API shows the plan: eleven tasks at parallelism 1. The POS dedup (Deduplicate[4] -> Calc[5] -> (Correlate…)) serves all three INSERTs, but the TemporalJoin runs three times and the price dedup twice: a statement set shares the source, not necessarily the expensive operators.

Step 4: what each query emits, and what the dedup keeps

query                                         operator                             emits
dedup, first copy by arrival                  Deduplicate FirstRow on PROCTIME     I
dedup, first copy by event time               Deduplicate FirstRow on ROWTIME      I
versioned view (latest price per key)         Deduplicate LastRow on ROWTIME       I,UA,D
temporal join (price as of the sale)          TemporalJoin                         I
tumbling window, 1 hour                       GlobalWindowAggregate                I
running total per store and day               GroupAggregate                       I,UA
top 3 stores per country                      Rank                                 I,UA,D
running total into the Iceberg upsert table   GroupAggregate -> Iceberg sink       I,UB,UA

explain runs EXPLAIN CHANGELOG_MODE per query. The temporal join reads an updating view and still only inserts; the running total emits -U (UB) only for a consumer that asks, as the Iceberg sink does before ignoring it. A mismatched sink fails at planning time (errors, trimmed):

an updating query into an insert-only sink (kafka):
  org.apache.flink.table.api.TableException: Table sink '…store_totals_append' doesn't support consuming update changes which is produced by node GroupAggregate(…)

An upsert-kafka table without a primary key fails too (“‘upsert-kafka’ tables require to define a PRIMARY KEY constraint”).

Both dedups only insert, but keep different baskets:

first copy by arrival:    22,197 baskets
first copy by event time: 22,113 baskets

22,197 is every basket in the topic: 22,276 events minus Part 27’s 79 re-sends. The event-time dedup silently dropped 84 that arrived behind the watermark: 84 in every run of this replay, but the count depends on how the six partitions interleave.

Step 5: the price as of the sale

On 15 August, SKU-00676’s Singapore shelf price dropped at 13:00 local (05:00 UTC). What the job wrote:

SKU-00676 in Singapore stores on business date 2026-08-15, as the job wrote it (times UTC; SGT = UTC+8):
price version from   list price lines  first sale           last sale            till agrees
2026-07-04 16:00:00       1,609     7  2026-08-15 02:14:29  2026-08-15 04:54:25            7
2026-08-15 05:00:00       1,409    22  2026-08-15 05:10:14  2026-08-15 12:07:18           22
all lines: 58,119; priced 58,119; till price = as-of list price on 58,119
joined to each SKU's latest price instead (up to 2026-10-03): 2,338 lines disagree with the till

Seven morning sales took the July version, the 22 later ones the new one, and all 58,119 lines agree with their tills. The last line is a counterfactual on the host, from the same price file.

Step 6: does it match the batch answer?

reconcile reads the upsert table through Flink (equality deletes applied) and compares it with the lakehouse:

1. stream.store_sales_daily vs Part 21's accepted events of upload day 2026-08-15
   store-days 775, equal 774
   differs: GB-0044 2026-08-14  stream 4 baskets, Part 21 3; stream - Part 21 = 1,528 minor units
   GB-0044-20260814-00045 (GB-0044, 1,528): Part 21 says resend, first copy uploaded 2026-08-14
2. business date 2026-08-15: stream vs gold.daily_store_sales
   gold   600 stores, 22,260 baskets
   stream 600 stores, 21,981 baskets
   + accepted baskets of that business date uploaded 2026-08-14 (UTC upload day): 85 (GB 1, IN 20, SG 64)
   + accepted baskets of that business date uploaded 2026-08-16 (UTC upload day): 194 (GB 126, IN 61, SG 7)
   stores where stream + other upload days = gold (baskets and sales): 600 of 600
3. stream.store_sales_hourly vs the lines the job wrote
   hourly windows: 22,113 baskets, 57,912 lines, 75,827 units
   lines table:    22,197 baskets, 58,119 lines, 76,113 units
   missing from the windows: 84 baskets, 207 lines, in 84 store-hours
   store-hours where late lines cover the gap: 84 of 84
   baskets uploaded more than 5 minutes after the sale: 586 (0 of them within 2 hours)

1. Same input: 774 of 775 store-days match to the minor unit. The 775th holds a £15.28 UK basket first uploaded on 14 August and re-sent on the 15th: Part 21 looks back three upload days, while the stream’s dedup state began with this replay.
2. Same business date: 279 baskets apart, all explained. Upload days are UTC days (Part 21), business dates local: a Singapore or India morning on the 15th, or London’s first hour after midnight (BST), is still the 14th in UTC, so 85 baskets (64 SG, 20 IN, 1 GB) were uploaded outside the topic; 194 more came from tills that uploaded late, on the 16th. Added back, all 600 stores equal gold. Today’s real-time number is provisional until late tills upload.
3. The window lost only late baskets. It is 84 baskets short, as many as the event-time dedup dropped, and each short store-hour’s gap is no larger than its late lines. Uploads come within 2 minutes of the sale or after more than 2 hours; 502 of the 586 late baskets still made their windows, presumably because the replay’s watermark (the minimum over six partitions read in turns) trailed the clock. A live one would drop them all.

Step 7: the changelog on the wire

changelog.sql prints one store’s running total, then the top three stores per country, and writes that Top-N to the compacted topic larkspur.analytics.top-stores with upsert-kafka; a Java reader replays the topic (trimmed: SG-0042’s changes 6 to 42, six of the nine stores):

SG-0042, business date 2026-08-15: 43 changes (first 5 and the last)
op  store_id baskets sales_minor
+I  SG-0042        1       3,099
-U  SG-0042        1       3,099
+U  SG-0042        1       4,297
-U  SG-0042        1       4,297
+U  SG-0042        2       4,466
+U  SG-0042        7      62,435
top 3 stores per country: 8,237 changes (+I 4,123, -D 4,114)
larkspur.analytics.top-stores: 8,255 records = 4,132 upserts + 4,123 tombstones; 9 keys alive at the end
  GB GB-0009  sales_minor         571,059  baskets   89
  IN IN-0058  sales_minor      35,973,289  baskets   95
  SG SG-0094  sales_minor         867,364  baskets   83

SG-0042’s 22 lines become 43 changes: an insert, then a retraction and a new row per line. The Top-N answered each update of a top store with -D and +I, not -U/+U; 4,123 inserts minus 4,114 deletes leave nine rows. The printed Top-N and the topic come from two executions of the query. The topic keeps every change, tombstones included, until compaction drops superseded records, and its counts vary between runs (8,237 to 8,283 changes across mine) as six partitions interleave; replayed in order, it always ends with the same nine rows.

Step 8: fewer updates for the same answer

Mini-batch aggregation buffers input and updates each key once per batch (table.exec.mini-batch.enabled, with allow-latency 5 s and size 5,000; run --mini-batch passes them with -D):

stream.store_sales_daily           1   21,594     1        775      20,819  business_date,store_id

21,594 rows instead of 58,119, same 775 live rows, same reconcile report, for up to 5 s more latency; a fast replay is the best case. tests/test_part28.py repeats the sequence from empty tables; make ci passed with this profile up (133 passed; the 118 skipped need services this profile leaves stopped).

The Landscape

Current product names as of 30 October 2026. A cell that couldn’t be verified says so.

Tips & Gotchas

Gotcha — dedup state must outlive the re-send window. Without a TTL the dedup remembers every basket forever; table.exec.state.ttl bounds it, but for every stateful operator in the job (the STATE_TTL hint covers only regular joins and group aggregations). Forget too early and a re-send counts twice, like GB-0044’s: size it from Part 21’s measured window.

Gotcha — a quiet versioned side stalls the temporal join. Output waits for both watermarks, so a price topic that changes twice a day holds every sale until its next change. scan.watermark.idle-timeout (Part 26) releases it, at the risk that a late-published price misses sales. From the docs; the lab’s prices run to October, so it never stalled.

Gotcha — group by exactly the sink’s key. Grouped by one more column than the table’s key, Flink 2.2.1 added a SinkMaterializer: when the keys differ, a shuffle can reorder one key’s changes, so it keeps each key’s history to rebuild the right upsert (table.exec.sink.upsert-materialize, default AUTO; Flink 2.3 calls that state unbounded and reworks it, FLIP-558). Grouping by the key, with MAX(currency), removed it.

Gotcha — not every reader can read an upsert table. PyIceberg 0.12 refused stream.store_sales_daily (“does not yet support equality deletes”); Snowflake supports none, and the Iceberg v4 draft forbids new ones (Part 7). Compact them away, or on a v3 table let the SinkV2 IcebergSink (experimental) convert them to deletion vectors (convertEqualityDeletes(); docs only).

Tip — run EXPLAIN CHANGELOG_MODE in CI on every new INSERT. It names what each query emits before anyone picks a sink; tests/test_part28.py asserts Step 4’s table this way.

Troubleshooting

Design Drill

The question: “We sell concert and sports tickets online. Card-testing fraud spikes in the minutes after a big on-sale. Give the payment-risk model a feature it can use at checkout: the number of distinct cards used from the same device in the last 10 minutes. Twenty thousand checkouts a second at peak.”

Clarify first

• How fresh must the feature be at scoring time: the current event included, or a minute old?
• What identifies a device (fingerprint, app install ID), and can fraudsters rotate it?
• Where is the feature read: inside the payment call, from a low-latency store, or both?
• Will the model be trained on the same definition, as of each historical checkout?

The shape of a strong answer

• Checkout attempts on Kafka keyed by device ID; dedup on attempt ID by arrival first, so retries stay out of the window’s state and out of any per-event counts.
• An OVER window per device, COUNT(DISTINCT card_hash) OVER (PARTITION BY device_id ORDER BY event_ts RANGE BETWEEN INTERVAL '10' MINUTE PRECEDING AND CURRENT ROW): one exact value per event (Flink 2.2.1 plans an append-only OverAggregate; flink/sql/part28/drill_over.sql, tills per store standing in).
• Not a HOP window: sliding every minute, it is up to a minute stale (Flink keeps one-minute slices and merges ten per result). Tokenise card numbers before Kafka; the feature needs equality, not the number.
• Size the state: a RANGE window keeps every row in range, 20,000/s × 600 s = 12 million rows, about 1–2 GB across the cluster at 100–150 bytes a row.
• For the scoring service, keep each device’s card hashes with their last-seen times in a key-value store; at checkout it counts the set plus the current card, within 10 minutes.
• A watermark of seconds; the OVER window waits for it, so its value lags, which is why checkout reads the set.
• The same definition in batch over the lake for training, as of each checkout (Part 32), reconciled daily with the stream.

Trade-offs to say out loud

• Exact sets per device are small; HyperLogLog matters only for keys like cards per merchant.
• A shorter watermark lowers lag and drops more late events; for fraud, a missed late event costs less than a slow checkout.
• Set a TTL a little above 10 minutes plus the watermark; without one, state keeps every device ever seen.

Common weak answers

• A GROUP BY over the last 10 minutes recomputed by a cron job every minute.
• A tumbling 10-minute window: a burst across a boundary is split in two and never flagged.
• Counting raw events, so retries and duplicates look like more cards.

Further Reading

1. Apache Flink 2.2 documentation, “Dynamic Tables” and “Versioned Tables”. https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/dev/table/concepts/dynamic_tables/ and https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/dev/table/concepts/versioned_tables/
2. Apache Flink 2.2 documentation, “Joins” (regular, interval, temporal, lookup). https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/dev/table/sql/queries/joins/
3. Apache Flink 2.2 documentation, “Deduplication” and “Top-N”. https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/dev/table/sql/queries/deduplication/ and https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/dev/table/sql/queries/topn/
4. Apache Iceberg 1.12 documentation, “Flink Writes” (upsert, commits, SinkV2). https://iceberg.apache.org/docs/latest/flink-writes/
5. Apache Flink 2.2 documentation, “Flink Architecture”. https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/concepts/flink-architecture/

Story Closing

The tile now reads the running total, not the window: every basket counts when it arrives, at the price its till charged. Beside it, the reconciliation: 774 store-days equal, one explained.

“And GB-0044?” Farah asked.

“A re-send from the day before,” Ingrid said. “Production’s dedup state will be weeks old. The nightly control stays anyway.”

SQL had hidden the state the job kept. In Part 29, Rohan writes a stock-out detector in Java with keyed state and timers, then kills the TaskManager mid-run to see what a checkpoint really saves.

This is Part 28 of “Every Record Counts: Data Engineering for Backend Engineers” — Season 3: Real Time, Serving and AI.