Exactly Once

Which exactly-once?

Module 7 ended with a claim you could test: kill a TaskManager between two checkpoints and the per-tenant totals inside GatewayMeter are neither lost nor double-counted. A billing engineer will ask one word about that claim — inside what? The invoice is built from the tenant-cost topic, and module 7 said nothing about what reached it.

"Exactly-once" names three different guarantees: exactly-once state semantics, the at-least-once mode you get when alignment is turned off, and end-to-end exactly-once, the only one a consumer of tenant-cost can observe. Which one is meant is the second cold question of this course.

Scope 1: exactly-once state semantics

This is module 7's guarantee, and upstream draws its boundary precisely (upstream's words, kept exact): "Given that Flink recovers from faults by rewinding and replaying the source data streams, when the ideal situation is described as exactly once this does not mean that every event will be processed exactly once. Instead, it means that every event will affect the state being managed by Flink exactly once."

Events after checkpoint k are processed again on recovery; TenantCost's accumulator does not count them twice because it was reset to its value at k. Anything that left the job between k and the failure — a record written to Kafka, a print, an HTTP call inside a map — is outside "the state being managed by Flink" and outside the guarantee.

Two conditions attach even here. The source must cooperate: "Flink can guarantee exactly-once state updates to user-defined state only when the source participates in the snapshotting mechanism." Kafka does (upstream's source table reads exactly once); a socket source does not. And the job must be in the default CheckpointingMode.EXACTLY_ONCE from module 7.

Scope 2: at-least-once mode

Barrier alignment costs latency, so Flink lets you skip it: env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.AT_LEAST_ONCE). Upstream says what you get (kept exact): "When the alignment is skipped, an operator keeps processing all inputs, even after some checkpoint barriers for checkpoint n arrived. That way, the operator also processes elements that belong to checkpoint n+1 before the state snapshot for checkpoint n was taken. On a restore, these records will occur as duplicates, because they are both included in the state snapshot of checkpoint n, and will be replayed as part of the data after checkpoint n."

So at-least-once mode weakens scope 1 itself: a RequestRecord can be added to usd twice inside Flink. Alignment only happens where an operator has several senders — after a keyBy, at a join — so a pipeline of map and filter is unaffected, but GatewayMeter has a shuffle before assemble and another before tenant-cost. Do not choose this mode for a job whose state is money; what it saves is, in upstream's words, "on the order of a few milliseconds".

Scope 3: end-to-end exactly-once

Upstream's two requirements (kept exact): "To achieve exactly once end-to-end, so that every event from the sources affects the sinks exactly once, the following must be true: 1. your sources must be replayable, and 2. your sinks must be transactional (or idempotent)". The guarantees page adds the mechanism — "the data sink needs to take part in the checkpointing mechanism" — and its sink table lists both routes: the Kafka producer, "exactly once with transactional producers (v 0.11+)", and the Cassandra sink, "exactly once only for idempotent updates". The rest of this page chooses one for tenant-cost.

The source: whose offsets are they?

On recovery "the sources are set to start reading the stream from position S_k", and the position lives in the checkpoint as the Kafka source's operator state, not in the consumer group. The connector doc is explicit (kept exact): "Kafka source commits the current consuming offset when checkpoints are completed, for ensuring the consistency between Flink's checkpoint state and committed offsets on Kafka brokers." And: "Note that 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."

That is what makes gateway-events replayable for Flink: the rewind is to a position Flink wrote, in a checkpoint Flink completed. The group's committed offset (commit.offsets.on.checkpoint, the KafkaSourceReader.committedOffsets metric) is a dashboard number; a failed commit "does not affect the integrity of Flink's checkpointed partition offsets". Without checkpointing there is no S_k to rewind to.

The sink: two-phase commit, as Flink runs it

A Flink sink is a SinkWriter with three methods (kept exact): "write(InputT element, Context context): Adds an element to the writer", "flush(boolean endOfInput): Called on checkpoint or end of the input, setting this flag causes the writer to flush all pending data for at-least-once. To archive exactly-once semantic, the writer should implement the SupportsCommitter interface", and "writeWatermark". Flushing at the barrier is all that DeliveryGuarantee.AT_LEAST_ONCE is: nothing is lost, and everything since the last checkpoint is written again on replay.

SupportsCommitter "is used to indicate that the sink supports exactly-once semantics using a two-phase commit protocol. The Sink consists of a CommittingSinkWriter that performs the precommits and a Committer that actually commits the data. To facilitate the separation, the CommittingSinkWriter creates committables on checkpoint or end of input and the sends it to the Committer." As a timeline:

1. Between barriers, the writer produces every record into an open Kafka transaction. Nothing is visible to a read_committed consumer.
2. Barrier n arrives at the writer. It pre-commits — flushes the transaction, hands a committable naming it to the committer, and opens a fresh transaction for what follows. The committable is part of checkpoint n.
3. The checkpoint coordinator declares n complete when every operator has acknowledged. Only then does the committer commit the transaction. Data written before barrier n becomes visible.

The gap between steps 2 and 3 is where a failure is interesting. If the TaskManager dies before n completes, checkpoint n is discarded, the job recovers from n−1, and the open transaction is never committed — the broker eventually aborts it. If n did complete and the death came before the commit landed, the committable is in checkpoint n, so recovery from n re-issues the commit. No record is committed twice either way; but the second case means a transaction opened before the failure must still be committable after the restart. Hence the connector's guidance (kept exact) "it is highly recommended to tweak Kafka transaction timeout (see Kafka producer transaction.timeout.ms)>> maximum checkpoint duration + maximum restart duration or data loss may happen when Kafka expires an uncommitted transaction."

Read the inequality with a dollar sign: a transaction Kafka expired is a TenantCostMinute that Flink's state believes was emitted and that no consumer will ever read. A minute never billed is revenue lost.

KafkaSink specifics

The three modes, in upstream's words (kept exact):

• "DeliveryGuarantee.NONE does not provide any guarantees: messages may be lost in case of issues on the Kafka broker and messages may be duplicated in case of a Flink failure."
• "DeliveryGuarantee.AT_LEAST_ONCE: The sink will wait for all outstanding records in the Kafka buffers to be acknowledged by the Kafka producer on a checkpoint. No messages will be lost in case of any issue with the Kafka brokers but messages may be duplicated when Flink restarts because Flink reprocesses old input records."
• "DeliveryGuarantee.EXACTLY_ONCE: In this mode, the KafkaSink will write all messages in a Kafka transaction that will be committed to Kafka on a checkpoint. Thus, if the consumer reads only committed data (see Kafka consumer config isolation.level), no duplicates will be seen in case of a Flink restart. However, this delays record visibility effectively until a checkpoint is written, so adjust the checkpoint duration accordingly."

Four consequences. The default is NONE — an unconfigured KafkaSink is not even at-least-once. AT_LEAST_ONCE and EXACTLY_ONCE both require checkpointing. EXACTLY_ONCE also requires setTransactionalIdPrefix(String), and upstream asks that you "use unique transactionalIdPrefix across your applications running on the same Kafka cluster such that multiple running jobs do not interfere in their transactions". And the consumer's isolation.level must be read_committed; at read_uncommitted it reads the aborted transaction's records as data, and every duplicate the mode was meant to hide is back.

The latency cost is structural: a read_committed consumer sees TenantCostMinute records only when a checkpoint commits, so with module 7's 10-second interval the invoice runs up to ten seconds behind the window. For billing that is nothing. For alerts it is the reason that topic stays where it is.

GatewayMeter, stage 8

env.enable_checkpointing(10_000)                                  # module 7: every 10 s
env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)

# transaction.timeout.ms >> checkpoint + restart; see below
tenant_cost_sink = KafkaSink.builder() \
    .set_bootstrap_servers(brokers) \
    .set_record_serializer(
        KafkaRecordSerializationSchema.builder()
            .set_topic("tenant-cost")
            .set_value_serialization_schema(SimpleStringSchema())  # TenantCostMinute as JSON
            .build()
    ) \
    .set_delivery_guarantee(DeliveryGuarantee.EXACTLY_ONCE) \
    .set_transactional_id_prefix("gatewaymeter-tenant-cost") \
    .set_property("transaction.timeout.ms", str(15 * 60 * 1000)) \
    .build()

tenant_cost_minutes.sink_to(tenant_cost_sink).uid("sink-tenant-cost")

The two builder calls after set_delivery_guarantee are the snake_case forms of the Java setTransactionalIdPrefix(String) and of the producer properties the connector doc refers to; upstream's Python tab shows only the at-least-once shape, so check them against your installed pyflink. set_partitioner is not available in PyFlink; that does not affect the guarantee.

The billing consumer changes one line — isolation.level=read_committed — and nothing else. It still appends every TenantCostMinute it reads; it simply never reads one Flink did not commit.

alerts keeps DeliveryGuarantee.AT_LEAST_ONCE. An error-rate alert delivered twice after a restart pages the same on-call twice, which is a nuisance and not a chargeback, while an alert held back until the next checkpoint is delayed by exactly the interval it is supposed to beat. Where duplicates are cheap and latency is not, at-least-once is the right guarantee, said out loud.

The alternative: an idempotent sink

TenantCostMinute has a natural key, (tenant_id, window_end), and a re-emitted record carries the same usd and requests as the first — module 7 guarantees that. Write it to a keyed store where a second write with the same key overwrites the first, and duplicates stop being a question: no transactions, no read_committed, no timeout to tune, visibility at write time rather than commit time. It also absorbs a duplicate transactions do not: module 6's late re-fire, where allowed lateness makes the window emit a corrected record for the same key. An upsert replaces the first value with the better one; a transactional append delivers both, once each, and leaves the consumer to prefer the later.

The price is on the reading side. The consumer must read current values by key, not an append log: a ledger that sums every record it has ever seen is not helped by a store that overwrote one. Upstream's example is the Cassandra row, "exactly once only for idempotent updates". If the billing system is a table keyed by tenant and minute, this is the simpler design.

The failure

Checkpoint 41 completed at t = 0. The 10:01:00 window for acme is in TenantCost's state at $4.10, waiting on the watermark.

At-least-once sink, append-only consumer:

Module 7's guarantee held throughout: the accumulator was reset to its checkpoint-41 value before the replay, so the second record says $4.10, not $8.20. The double-count happened downstream of Flink, in a consumer that treats every record as new money.

EXACTLY_ONCE sink, read_committed consumer:

The record reached the ledger once, when checkpoint 42 committed (4 s after the re-fire here; up to one 10 s interval in general) instead of immediately.

The transaction-timeout trap: same job, but the outage is longer.

Upstream's diagnosis (kept exact): "The reason for this exception is most likely a transaction timeout on the broker side. With the implementation of KAFKA-6119, the (producerId, epoch) will be fenced off after a transaction timeout and all of its pending transactions are aborted". The job may recover and continue; the 10:01:00 minute for acme is still gone, because checkpoint 42 records it as emitted and no replay will produce it again. The guidance was "much greater than" checkpoint plus restart, and 15 minutes is not much greater than 16.

Which one does the invoice need?

State the sink and the consumer, and the answer falls out. tenant-cost into a keyed table the billing system reads by (tenant_id, window_end): exactly-once state semantics plus an idempotent upsert is enough, and cheaper. tenant-cost into a topic an append-only ledger consumes: the sink must be transactional, the consumer read_committed, and transaction.timeout.ms becomes a parameter someone owns. Neither answer is "Flink is exactly-once".

That is why the unqualified phrase is a red flag in a design review: it usually offers scope 1 as the answer to a scope 3 question. Three short follow-ups expose it — which sink, what does the consumer read, and what happens to an open transaction during your longest plausible restart.

Temporal, one paragraph

Module 4 of the Temporal course reached the same discipline from the other side: an Activity executes at-least-once, so a side effect needs an idempotency key that the callee enforces. Flink moves the mechanism, not the shape. Temporal retries a call and dedupes at the target with a key; Flink replays a stream and dedupes at the sink, with a key (the upsert) or with a transaction the checkpoint commits. Replay of a source is not retry of a call, but in both runtimes the world outside is at-least-once and correctness is bought at the boundary, not in the middle.

If a TaskManager dies right now

• Survives: every operator's state as of checkpoint k; the source positions S_k; every Kafka transaction the committer committed — that is, every TenantCostMinute a read_committed consumer has ever seen.
• Recomputed: everything after S_k — assembled requests, the open window accumulators, and the sink writes that followed. The recomputed values equal the first ones.
• Emitted twice: at read_committed on the EXACTLY_ONCE sink, nothing; the transaction that was open at the failure is aborted, never read. At an idempotent upsert, nothing visible — the second write lands on the first. On the at-least-once alerts topic, everything the sink wrote since k.

Try it

Lab 8 is in labs/08-exactly-once/:

• 8.1 — leave tenant-cost at DeliveryGuarantee.AT_LEAST_ONCE, run the append-only billing consumer, docker kill a TaskManager six seconds after a checkpoint, and find the minute that billed twice.
• 8.2 — switch the sink to EXACTLY_ONCE with a transactional id prefix, set the consumer to isolation.level=read_committed, repeat the kill, and confirm one record; measure how long after window close it became visible and relate it to the checkpoint interval.
• 8.3 — replace the topic with the upsert sink keyed (tenant_id, window_end), repeat the kill, then inject module 6's late RequestCompleted and watch the corrected value replace the first.
• 8.4 — set transaction.timeout.ms below your restart time, stop and resume the job from a retained checkpoint after the timeout has elapsed, and read the ProducerFencedException — then find which minute is missing from the ledger.

Lab page: lab-08-exactly-once.html — the exercises, every source file, and the run-it-locally steps.