Flink Checkpoints Stuck Behind a Backpressured Sink

A checkpoint has a trigger phase, an alignment phase (each subtask waits until all its input channels have delivered the barrier), a synchronous snapshot phase, and an asynchronous upload phase. Flink exposes these separately: checkpointAlignmentTime, checkpointStartDelay, syncDuration, asyncDuration, and the per-subtask checkpointStartDelayNanos metrics. If alignment time dominates while syncDuration and asyncDuration are modest, the problem is backpressure, not state size. If async duration dominates, it is the HDFS upload of 40 GB per checkpoint.

The symptom set points strongly at alignment. Sustained backpressure at the HBase sink means the slowest operator cannot drain its input buffers, so barriers queue behind in-flight records and every upstream subtask waits. Consumer lag growth and 25-minute watermark lag confirm the pipeline is throughput-bound, not checkpoint-bound. The 40 GB figure also matters: 40 GB per checkpoint is a full, non-incremental RocksDB snapshot upload, which alone can take minutes at typical HDFS write throughput. So both phases likely contribute, with alignment dominant during peak.

Raising the timeout to 30 minutes is harmful for three reasons. First, Flink does not trigger a new checkpoint while one is in progress, so a 12-minute snapshot silently turns a 5-minute interval into a 12-minute interval, increasing the data volume replayed after any failure. Second, a longer timeout allows snapshots to hold state and HDFS files for longer, growing storage pressure. Third, it does not remove the backpressure, so recovery time and lag keep worsening. The timeout should be a safety net, not the tuning knob; fix throughput and alignment first, then keep the timeout near 2x the expected checkpoint duration.

Address throughput first, because checkpoint alignment recovers on its own once the sink is no longer the bottleneck.

Throughput fixes:

1. The HBase sink. If it performs a synchronous Put per record, replace it with BufferedMutator batching (100-500 rows or 2-4 MB per flush) and raise hbase.client.write.buffer. Check whether the sink is hitting a hot region - a monitor-style table with a timestamp-first rowkey will concentrate writes on one RegionServer and no amount of Flink parallelism will help.
2. The Kafka producer. Default batch.size is small; move to 128 KB with linger.ms 50-100 and enable compression (lz4). This is usually a large win for a 480k events/s sink.
3. Parallelism. 40 with a 60-partition source is below the partition count; raise to 60 so every partition can be consumed without one subtask driving two. Adding TaskManagers with the same slot count only adds slots, which is useful only if parallelism is also raised and the external sink scales.
4. RocksDB memory. Check that managed memory is adequate; an undersized block cache and write buffer turns every state access into disk IO, which shows up as backpressure inside the job rather than at the sink.

Checkpoint mechanics: 5. Enable unaligned checkpoints (execution.checkpointing.unaligned=true), available since 1.11. Barriers then overtake in-flight records, removing alignment time entirely under backpressure. Note the cost: in-flight data is included in the snapshot, so checkpoint size grows, and the storage must support it. 6. Enable RocksDB incremental checkpoints (state.backend.incremental=true) and local recovery for the state backend, so only new SST files are uploaded and a TaskManager can restore locally after a failure. 7. Check execution.buffer-timeout and network buffer counts; a 100 ms buffer timeout with too few buffers adds latency and backpressure independent of the sink.

Order matters: steps 1-2 usually drop alignment time enough that step 5 becomes optional.

A watermark is the minimum across all input channels of a subtask. If one Kafka partition stops producing at 01:00, its watermark stops advancing, and the operator's watermark is pinned at that old value even though the other 59 partitions are current. Session windows that should fire based on event time therefore never reach their end timestamp for any key, which is exactly the 'results stop firing' symptom. It is not a state or checkpoint bug.

The standard fix is idleness detection on the watermark strategy, for example WatermarkStrategy.forBoundedOutOfOrderness(...).withIdleness(Duration.ofMinutes(2)). A partition with no records for the configured duration is marked idle and temporarily excluded from the minimum, so the operator watermark advances. The trade-off is correctness: if data later arrives on that partition with timestamps older than the watermark, it is late and is dropped by the window unless you configure allowedLateness and route late events to a side output. Choose the idleness timeout larger than the normal inter-arrival gap for the partition, and monitor late-event counters.

An alternative is a heartbeat or a synthetic record per partition per minute, which keeps watermarks honest and avoids the drop, at the cost of extra traffic.

For the exactly-once design: exactly-once is provided by the checkpoint plus the two-phase-commit Kafka sink and by idempotent/transactional writes to HBase. Turning off exactly-once to reduce checkpoint cost would trade correctness for throughput, and would require downstream deduplication by event id (for example an HBase upsert keyed by event id, or a dedup table in the serving store). If the checkpoint fix is only partially successful, prefer keeping exactly-once and reducing state (shorter session windows, TTL state) over dropping the guarantee, because deduplication in the serving layer is harder to operate than a well-sized checkpoint.