Flink Event Time and Watermarks: Checkpoints, Savepoints, and the Sink Key You Still Need

Finance closed Tuesday $2.4 million light. The Flink job was healthy. Checkpoints succeeded every ten seconds. The watermark was BoundedOutOfOrderness of ten seconds because a tutorial used ten seconds. Card-network events arrive 45 to 90 seconds late on a normal afternoon. Flink did what we configured: it closed the window and dropped the late payments. The dashboard used the sink table, not the Kafka topic, so the topic still had the money.

I changed the watermark to two minutes and replayed from a savepoint taken before the bad config. The money came back. The lesson was not “Flink is wrong.” The lesson was that a watermark is a drop policy with a friendly name.

flowchart TD
  ev[Event in Kafka] --> ts[Event-time timestamp]
  ts --> wm[Watermark]
  wm -->|ts <= watermark| late[Dropped or side output]
  wm -->|ts > watermark| win[Window / join]
  win --> ckpt[Checkpoint]
  ckpt --> sink[Sink]
  sink --> key{Idempotency key?}
  key -->|yes| once[Safe replay]
  key -->|no| dup[Duplicates on recovery]

Watermarks decide which events exist. Checkpoints decide what you can replay. The sink key decides whether replay duplicates.

Event time versus processing time

Event time is payment.authorized_at. Processing time is System.currentTimeMillis() on the task manager. They agree when the job is caught up and the clock is honest. They diverge on restart, on backpressure, and on any source that can stall.

I have seen a processing-time tumbling window “work” for months and then double-count a burst after a failover, because the burst was old events arriving on a new clock. Use event time for anything that money, inventory, or a regulator will read. Use processing time for “page me if this operator is idle,” not for the aggregate.

from pyflink.common import Duration, Types, WatermarkStrategy
from pyflink.common.watermark_strategy import TimestampAssigner
from pyflink.datastream import StreamExecutionEnvironment


class PaymentTs(TimestampAssigner):
    def extract_timestamp(self, value: dict, record_timestamp: int) -> int:
        # milliseconds since epoch from the payload, not the Kafka timestamp
        return int(value["authorized_at_ms"])


env = StreamExecutionEnvironment.get_execution_environment()
env.enable_checkpointing(10_000)
env.get_checkpoint_config().set_min_pause_between_checkpoints(5_000)
env.get_checkpoint_config().set_checkpoint_timeout(60_000)

# 90s delay matches the p99 we measured. 10s was a guess.
watermark = (
    WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(90))
    .with_timestamp_assigner(PaymentTs())
    .with_idleness(Duration.of_minutes(2))
)

with_idleness matters. One silent Kafka partition holds the watermark back for the whole stream. Without idleness, a low-volume tenant stalls every window. With it, you are promising that silence means “advance,” which is wrong if the producer actually died. Pair it with a source-lag alert so idleness is not how you discover an outage.

Measure the lag before you pick the bound. I take p99 of now - event_time on a day’s worth of the topic and add a small buffer. A round number from a blog is how you get a $2.4 million hole.

Checkpoint versus savepoint

A checkpoint is automatic, aligned, and owned by the job. Recovery after a task-manager death starts from the last successful checkpoint. If you delete the checkpoint directory, you have a new job that does not remember Kafka offsets or keyed state.

A savepoint is an operator-triggered, portable snapshot. You take one to upgrade the job, change parallelism, or move clusters. You restore from it on purpose. I take a savepoint before every config change that touches watermarks, keys, or sinks. The Tuesday replay was a savepoint restore, not a hopeful checkpoint.

Incompatible state after a key change will refuse to restore. That is the same class of incident as a Spark checkpoint that no longer matches the query — walked through in Structured Streaming. Plan the migration: new uid, dual-write, or accept a gap.

Exactly-once at the sink still needs a key

Flink’s two-phase commit sink can give you exactly-once into Kafka (or another transactional sink) together with the checkpoint. The moment the sink is INSERT into Postgres, the transaction coordinator does not know about that row. Recovery will write it again.

This is the same boundary as Kafka exactly-once. set_delivery_guarantee(DeliveryGuarantee.EXACTLY_ONCE) on a Kafka sink does not travel with a JDBC statement.

Give every external row a natural key and MERGE. Or write only to Kafka and let a batch job land the warehouse. I will not run an exactly-once story that ends in an unqualified INSERT.

When Spark Structured Streaming is enough

If the real shape is “every 15 minutes, process what arrived, write a partition,” Spark’s AvailableNow trigger on a scheduled job is cheaper to operate than an always-on Flink cluster. You still need a watermark and an idempotent sink. You do not need Flink’s keyed state for a 15-minute microbatch over a day’s keys.

Pick Flink when the job is actually streaming: large keyed state, event-time joins, low-latency windows, or a CDC stream that cannot wait for the next batch. Pick Spark when your team already debugs Spark and the SLA is minutes. The wrong reason is “Flink is the real streaming engine.”

Failure modes

Tutorial watermarks. Ten seconds is not a measurement.

Idle sources without a lag page. The watermark advances. The producer is dead.

Checkpoint directory reused across job graphs. You get a state-incompatibility error or a silent wrong restore.

Exactly-once plus INSERT. Duplicates on every failover.

Processing-time windows for money. They lie after a restart.

What to do Monday

Plot event-time lag. Set the watermark from that plot. Turn on checkpoints with a timeout you have tested under load. Take a savepoint before you change keys or watermarks. Put a merge key on every non-Kafka sink. If the job can be a scheduled microbatch, run Spark and save the Flink on-call for the streams that need it.