Seventy-Three Thousand Duplicates

A Flink job has a hundred thousand rows to write into an Iceberg table. Stop it with a savepoint a quarter of the way through, then resume it. So far, an ordinary recovery. Now resume it again from the same savepoint.

  stopped at ~25s with savepoint savepoint-946a8a-…;  27,000 rows committed
  resumed once:                        FINISHED;  100,000 rows, 100,000 distinct
  resumed AGAIN from the same savepoint: FINISHED;  173,000 rows, 100,000 distinct -> 73,000 duplicates
  distinct flink.job-id values in the snapshot log: 3

The first resume was exact — every row once, none lost across the stop. The second resume, from the same savepoint, replayed every row after it, and the table now holds seventy-three thousand duplicates. Nothing failed. Both jobs finished green. The sink de-duplicates commits by job id and checkpoint id, and a job resumed from a savepoint is a new job with a new id, so it de-duplicates against nothing.

Book 1’s chapter 17 described streaming without running it. Here, Flink 2.1.3 with the 1.11.0 runtime writes through Polaris to the bookshop’s event tables. A writer that never stops puts the previous three chapters’ maintenance, retention and recovery policies to work together. The runs below include deliberate failures, because a policy that works only while the writer is healthy leaves the hardest questions unanswered.

What a checkpoint leaves behind

An unbounded stream of order events at two hundred rows a second, into a day-partitioned table, with a five-second checkpoint interval. Sampled every thirty seconds.

  t= 30s:   6 snapshots    6 manifests  (45 KB)   18 data files   metadata.json  8 KB   checkpoints: completed 6
  t= 60s:  12 snapshots   12 manifests  (90 KB)   36 data files   metadata.json 14 KB   checkpoints: completed 12
  t= 90s:  18 snapshots   18 manifests (135 KB)   54 data files   metadata.json 21 KB   checkpoints: completed 18
  latest snapshot: append; summary: added-data-files 3, added-records 1000,
                   flink.job-id 37eb0ad8…, flink.max-committed-checkpoint-id 18, flink.operator-id 3d05135c…

The counts grow together: one snapshot per checkpoint, as Book 1 found, and one data file per partition touched — three here. The less obvious cost is one manifest per checkpoint. That answers the question chapter 9 left open: the Flink sink uses a fast append, and a fast append never merges. Spark’s writer folded six hundred commits into six manifests; Flink’s would have left six hundred. The metadata file grew by 0.7 KB per snapshot — exactly chapter 9’s slope.

Extend that rate to a day: 17,280 snapshots, 17,280 manifests, fifty thousand small files, and a metadata file past twelve megabytes served on every load. None of that is a defect. It is what a streaming writer is, and it means chapter 8’s compaction, chapter 9’s manifest rewrite and chapter 10’s hot-class retention are not maintenance a streaming table might need. They are the other half of the writer.

The snapshot summary carries the three keys that make everything below auditable: the job id, the operator id, and the highest checkpoint the job has committed. Every finding in this chapter was read from them.

Maintenance against a writer that does not pause

The three maintenance jobs, run from Spark while the stream above kept committing.

  rewrite_data_files (partial progress):  54 -> 3;   job RUNNING, checkpoints completed 18, failed 0
  expire_snapshots (no args, hot-class props): deleted 0 data files, 16 manifest lists;   job RUNNING
  rewrite_manifests:                       4 -> 1;   job RUNNING
  after maintenance + 12 s: 9 snapshots, 3 manifests, 12 data files, metadata.json 14 KB;
                            checkpoints completed 21, failed 0;  job exceptions: 0

All three committed, the writer never noticed, and twelve seconds later the table was smaller than it had been at any point during the stream. Chapter 8 established that appends and rewrites do not conflict. This is that fact under a writer that appends every five seconds, and it holds because each side touches files the other never will.

The first attempt at this section did not run on Polaris. It ran on the reference fixture, and its SQLite could not take the writer’s commit and Spark’s rewrite at the same moment. The rewrite committed. The Flink job’s own commit got a 500 and the job went to RESTARTING with one restored checkpoint. That is the correct response: a streaming writer that cannot commit does not lose the checkpoint’s rows. It restores and commits them again. The next Spark call got the 500 too, and the stage moved to Polaris. The catalog is the coordination point for a streaming table, and it has to take the writer’s commit rate plus every maintenance job’s commits on top. Chapter 3 said so in the abstract — this is what it looks like when it does not.

Polaris had a lesson of its own before it would take the writer, and it is chapter 3’s address problem in a new place. A Polaris catalog stores an endpoint for clients and another for itself, and it hands the client one to every engine as a table-level setting that overrides whatever the engine configured. The lab’s catalog stored the host’s address, so Flink inside Docker was told to write to localhost:8333, and every checkpoint failed for ninety seconds with a connection refused. Nothing in the job’s own configuration was wrong. The fix was a catalog whose stored endpoint resolves from both sides, and the finding is that a catalog decides where a streaming writer writes, and it can decide wrongly for the writer while being right for everyone else.

Failure one: the TaskManager is killed

A bounded sequence of two hundred thousand rows at two thousand a second, and thirty seconds in, docker kill on the only TaskManager, then eight seconds later, docker start.

  t=30s: 34,000 rows committed, checkpoints completed 2;  killing the TaskManager
  job FINISHED: 200,000 rows, 200,000 distinct ids;  checkpoints completed 19, failed 7, restored 1
  committed checkpoint ids, newest first: 9223372036854775807, 25, 24, 23, …

Every row once. Seven checkpoints failed while the TaskManager was gone, one was restored when it came back, and the job finished with the exact count. That is the checkpoint-to-commit coupling doing what it is for. The sink only commits files at a completed checkpoint, so the rows written between checkpoint two and the kill were never in the table. The restored job wrote them again from the source’s checkpointed position. The files it had written before the kill are on the store — uncommitted — and they are chapter 9’s orphans until remove_orphan_files finds them.

The odd number at the top of the list is the end-of-stream commit. A bounded job’s final flush commits under Long.MAX_VALUE rather than a checkpoint id, and a monitoring rule that expects checkpoint ids to be small integers will trip on it.

Failure two: the savepoint that was used twice

The duplicate rows in the opening came from a recovery mechanism working twice. A stop-with-savepoint drains nothing and records the source position and the sink’s pending files. Resumed once, the new job commits from that position and the count is exact. Resumed again, from the same savepoint, the second new job does the same thing — and the sink lets it. Its de-duplication is against its own job id’s committed checkpoints, not against the table.

  distinct flink.job-id values in the snapshot log: 3

A savepoint is single-use, and the table cannot tell you it was used twice — the snapshot log can. The audit is a GROUP BY summary['flink.job-id'] on snapshots: two job ids after a savepoint is a resume, three is a replay. The repair is chapter 10’s rollback to the last snapshot of the first resumed job, and then a new savepoint from a running job, never the old one. Both are in the runbook at the end.

A schema change on the wire

Two changes made from Spark to a table a Flink job was writing at that moment.

  ADD COLUMN extra INT, plus one Spark insert:   job RUNNING; snapshots 3 -> 7; schema now [id, ts, note, extra]
                                                 rows with extra NULL: 3,000   NOT NULL: 1
  DROP COLUMN note (a column the job writes):    job RUNNING after 20 s; exceptions 0; snapshots 7 -> 11; checkpoints completed 10

The job survived both, and that is the finding, because surviving is not the same as following. After the add, its rows carry a null in the new column, which is right. After the drop of a column it writes on every checkpoint, it kept writing that column, and its commits kept succeeding. A commit validates the files against the table’s state, not the files’ schema against the table’s current one. The rows it writes carry a field the table no longer has, addressed by a field id no reader will project. Whether a restarted job picks up the current schema was not measured here, and the chapter says so; what was measured is that a running one does not.

Schema changes on a streaming table are deployed with the job, not alongside it. The dangerous case is the one this run set up and did not complete: drop note, then a month later add a column called note. It gets a new field id, every file the running job wrote in between carries the old one, and the data is in the files and unreachable from the table.

Late data lands where its timestamp says

Six events through Kafka into a day-partitioned table: three stamped today, three stamped a month ago.

  after two checkpoints:  2026-08-01: 1 row in 1 file;  2026-08-02: 1 row in 1 file;  2026-08-03: 1 row in 1 file;  2026-09-06: 3 rows in 1 file
  compaction scoped to this month: rewrote 0 files;  the August partitions keep their one-row files: 3

Late data is not a streaming problem — it is a compaction-scope problem. The rows landed in the partitions their timestamps name, as one-row files, and chapter 8’s rule of scoping compaction to the partitions the writer is filling scoped straight past them. A month-scoped compaction rewrote nothing. The compaction where on a streaming table has to cover the lateness window the source can produce, or a second unscoped pass has to run on a slower schedule. Chapter 8’s fleet score, files beyond one per partition, is what catches the ones that were missed.

Equality deletes, produced for real

The table Book 1 could only describe: a keyed upsert stream, a thousand keys, five hundred rows a second, five-second checkpoints.

  t=30s:   6 snapshots,  6 data files;  delete files: position 6 (8,281 positions),  equality 6 (5,219 rows)
  t=60s:  12 snapshots, 12 data files;  delete files: position 12 (17,801),         equality 12 (10,699)
  Spark count(*) = 1,000 (1,000 keys):  GETs 104, 289 ms         Trino count(*) = 1,000

Every checkpoint adds a data file, a position-delete file and an equality-delete file. The position deletes are the keys that were upserted more than once within the checkpoint, resolved against the checkpoint’s own data file. The equality deletes are the keys that existed before the checkpoint, expressed as “delete any row with this key from any older file”. That is the form no writer can resolve at write time, because the older files are not being read.

The counts give the shape of the workload. Thirty thousand rows in sixty seconds over a thousand keys: 17,801 of them collided inside their own checkpoint and became position deletes, 10,699 replaced a key from an earlier checkpoint and became equality deletes, and a thousand are live. An upsert stream over a small key space is mostly deletes, and every one of them is a file a reader has to open and apply. Both engines that read equality deletes agree on a thousand rows, and the reader that does not, PyIceberg, was Book 1’s and stage 4’s finding and still holds.

The cost is on the read side: a hundred and four requests and nearly three hundred milliseconds to count a thousand rows, and both numbers grow with every checkpoint. Then the two rewrites.

  rewrite_position_delete_files:               13 -> 13;  equality deletes untouched
  rewrite_data_files (delete-file-threshold 1): 13 -> 1;  delete files now: none;  count(*): GETs 0, 46 ms

The position-delete rewrite did nothing useful — thirteen files in, thirteen out, and it cannot touch an equality delete at all. The data-file rewrite with a delete threshold of one removed every delete file of both kinds. Rewriting the data files applies the deletes and leaves nothing for them to apply to. An upsert table’s maintenance is the data-file rewrite, scheduled by checkpoint count rather than by file size. Its files are never large, and its delete files arrive at a fixed rate.

The sink that is not in this chapter

The Kafka Connect Iceberg sink was set up on the same platform in the Gate 1 lab and is not in these results. Its only prebuilt build for this Kafka is 1.9.2. It runs, it creates its table — and it has never committed. That is recorded as open in the lab and is stated here so that its absence is not read as a choice. Chapter 12 uses Flink for the change-data path for the same reason.

The maintenance schedule and the runbook

Append and upsert streams leave different work behind. This schedule brings the jobs from chapters 8 to 10 together by table class, so the checkpoint rate and lateness window determine when maintenance runs.

Job	Append stream (events)	Upsert stream (orders)	Why
rewrite_data_files	hourly, where covering the lateness window; partial progress on	every N checkpoints, delete-file-threshold 1	files per checkpoint; delete files per checkpoint
rewrite_manifests	hourly, after compaction	with compaction	one manifest per checkpoint, never merged
expire_snapshots	no arguments; hot-class properties	same	the metadata file grows 0.7 KB per checkpoint
remove_orphan_files	daily, older_than past the longest recovery	same	every restore leaves uncommitted files
unscoped compaction	weekly	monthly	the late partitions the where missed

The second is the runbook, one line per failure this chapter produced.

Event	What the table shows	What to do
TaskManager lost	failed checkpoints, then restored > 0; row count exact	nothing; schedule orphan cleanup
catalog returned 500 to the writer	job RESTARTING; the checkpoint re-commits	fix the catalog; the writer will not lose the checkpoint
job resumed from a savepoint	one new flink.job-id in snapshots	expected; delete the savepoint
the same savepoint resumed twice	a third flink.job-id; duplicates in the table	roll back to the first resumed job’s last snapshot; take a new savepoint
column dropped under a running job	the job keeps committing with the old field	restart the job with the schema change; never re-add the name
one-row files in old partitions	partitions shows files in dates the writer is not filling	widen the compaction where
read cost climbing on an upsert table	delete_files count equals the checkpoint count	the data-file rewrite with a delete threshold

What was not run

A restart of the schema-changed job, to see whether it picks up the current schema. The Kafka Connect sink, which is open. A TaskManager loss on an upsert stream, where the pending equality deletes are part of the state. Flink’s own write.upsert.enabled interaction with a copy-on-write compaction landing between two checkpoints, which chapter 8 says will conflict and this chapter did not force. And the sink’s exactly-once behaviour across a JobManager loss, which needs a high-availability configuration this lab does not have.

Exercises

1. Audit the job ids. On any Flink-written table, run SELECT summary['flink.job-id'], count(*), min(committed_at), max(committed_at) FROM t.snapshots GROUP BY 1 ORDER BY 3. Every row is a job. Two overlapping time ranges is two jobs writing at once; a job id whose first commit is minutes after another’s last is a resume. Match each to a savepoint or a deployment.

Show answer

Overlapping ranges mean duplicated writers, which is the double-resume case, and the duplicates start at the second job’s first commit. A gap between one job’s last commit and the next’s first is the outage the resume covered, and the source’s position at the savepoint is what made it lossless.

2. Find your lateness window. On a day-partitioned streaming table, run SELECT partition, min(committed_at) FROM t.partitions JOIN t.snapshots … or, more simply, compare each partition’s date with the snapshot timestamp of its newest file. The largest difference is how late your source has actually been.

Show answer

The files table’s snapshot_id joined to snapshots.committed_at, minus the partition date, gives the lateness of every file. The maximum is the window the compaction where has to cover; if it is longer than the retention window, the late rows are also landing in partitions whose history has already been expired, which is correct but surprising.

Final thoughts

A streaming table is a table with a writer that has opinions about time. It commits when its checkpoint says — not when the data is tidy. It leaves a manifest per commit and never merges them. It recovers from losing its worker by re-writing the rows and leaving the first attempt on the store. It will happily replay a savepoint into duplicates. And it will keep writing a column that no longer exists. Every one of those is correct behaviour for a streaming writer — and every one is a maintenance or a runbook entry for the platform that owns the table.

The next chapter is where the rows come from. Change data from PostgreSQL, through Debezium and Kafka, into an Iceberg table that has to converge on the source, and out again to consumers reading only what changed.

Next: Sixty Thousand Deletes, Nothing Deleted