Fault Tolerance & Checkpointing

What can go wrong in a streaming job:

Hardware failures:
  - TaskManager JVM crashes (OOM, segfault)
  - Machine dies (power, hardware failure)
  - Disk corruption or full

Software failures:
  - User code throws exception (NPE, parsing error)
  - Serialization failure (schema mismatch)
  - Timeout (checkpoint takes too long)

Infrastructure failures:
  - Network partition between TaskManagers
  - Kafka broker unavailable
  - S3/HDFS checkpoint storage unreachable

Without fault tolerance:
  - All in-flight state is LOST
  - Events processed since last output are LOST
  - Kafka offsets not committed → reprocessing → DUPLICATES
  - Hours of computation wasted

With Flink checkpointing:
  ✅ State restored to last consistent snapshot
  ✅ Source rewound to checkpoint position
  ✅ Processing resumes from consistent point
  ✅ No data loss, no duplicates (exactly-once)

// Enable checkpointing
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Checkpoint every 60 seconds
env.enableCheckpointing(60_000);

// Checkpoint configuration
CheckpointConfig config = env.getCheckpointConfig();

// Exactly-once vs at-least-once
config.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// Minimum time between checkpoints (prevents overlap)
config.setMinPauseBetweenCheckpoints(30_000);

// Checkpoint timeout (fail if not complete in 10 min)
config.setCheckpointTimeout(600_000);

// Max concurrent checkpoints
config.setMaxConcurrentCheckpoints(1);

// Tolerate N checkpoint failures before job fails
config.setTolerableCheckpointFailureNumber(3);

// Keep checkpoints on cancellation (for manual recovery)
config.setExternalizedCheckpointCleanup(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

// Checkpoint storage
config.setCheckpointStorage("s3://my-bucket/checkpoints/");

Aligned Checkpoint Barriers:

Source 1: [e1] [e2] [B_n] [e3] [e4] ...
Source 2: [e5] [e6] [e7] [B_n] [e8] ...

Operator with 2 inputs:
  1. Receives B_n from Source 1 first
  2. BLOCKS Source 1 input (buffers events after barrier)
  3. Continues processing Source 2 until B_n arrives
  4. Receives B_n from Source 2
  5. NOW: takes state snapshot (consistent point)
  6. Emits B_n downstream
  7. Unblocks Source 1, processes buffered events

Timeline:
  Source 1: ──[e1]──[e2]──[B]──[e3]──[e4]──────────→
                              ↓
  Operator:  processes e1,e2  │ BLOCKS  │ snapshot │ resume
                              ↓         ↓          ↓
  Source 2: ──[e5]──[e6]──[e7]──[B]────────────────→
                                   ↑
                              still processing

Problem: blocking causes backpressure during alignment!

// Enable unaligned checkpoints (Flink 1.11+)
env.getCheckpointConfig().enableUnalignedCheckpoints();

// Unaligned checkpoints:
// - Don't block inputs during barrier alignment
// - Instead, snapshot in-flight data (buffers) as part of checkpoint
// - Faster checkpoint completion under backpressure
// - Larger checkpoint size (includes network buffers)
// - Recommended when checkpoint duration is high due to alignment

Checkpoint Storage Options:

1. JobManager Heap (development only):
   state.checkpoint-storage: jobmanager
   - Stored in JobManager memory
   - Lost if JobManager crashes
   - Only for testing!

2. Filesystem (production):
   state.checkpoint-storage: filesystem
   state.checkpoints.dir: s3://bucket/checkpoints/
   - Durable, survives any single failure
   - Supports S3, HDFS, GCS, Azure Blob

Full vs Incremental Checkpoints:
═══════════════════════════════════════════════════
Full checkpoint (HashMap backend):
  Checkpoint 1: [████████████] 10 GB (full state)
  Checkpoint 2: [████████████] 10 GB (full state again)
  Checkpoint 3: [████████████] 10 GB (full state again)
  Total I/O per checkpoint: 10 GB

Incremental checkpoint (RocksDB backend):
  Checkpoint 1: [████████████] 10 GB (initial full)
  Checkpoint 2: [██]           2 GB  (only changed SST files)
  Checkpoint 3: [█]            1 GB  (only changed SST files)
  Total I/O per checkpoint: 1-2 GB (90% reduction!)
═══════════════════════════════════════════════════

# Trigger savepoint (job keeps running)
flink savepoint <job-id> s3://savepoints/

# Stop job with savepoint (graceful shutdown)
flink stop --savepointPath s3://savepoints/ <job-id>

# Cancel job with savepoint (immediate)
flink cancel --withSavepoint s3://savepoints/ <job-id>

# Start job from savepoint
flink run -s s3://savepoints/savepoint-abc123 job.jar

# Start with different parallelism (rescaling)
flink run -s s3://savepoints/savepoint-abc123 -p 32 job.jar

# Common workflow: upgrade job version
# 1. Stop old version with savepoint
flink stop --savepointPath s3://savepoints/ <old-job-id>
# 2. Deploy new version from savepoint
flink run -s s3://savepoints/savepoint-xyz789 new-job-v2.jar

End-to-End Exactly-Once Requirements:

Layer 1: Source (replayable)
  ✅ Kafka — can replay from committed offset
  ✅ Kinesis — can replay from sequence number
  ❌ Socket source — cannot replay (at-most-once)

Layer 2: Internal Processing (Flink checkpoints)
  ✅ Guaranteed by checkpoint + barrier alignment
  ✅ On failure: restore state + rewind source to checkpoint
  ✅ Events between checkpoint and failure are reprocessed
  ✅ State is restored → reprocessing produces same result

Layer 3: Sink (idempotent or transactional)
  Option A: Idempotent sink
    ✅ Writes are idempotent (same write twice = same result)
    ✅ Example: upsert to database with primary key
    ✅ Simple but requires idempotent operations

  Option B: Transactional sink (two-phase commit)
    ✅ Writes committed only when checkpoint completes
    ✅ Example: Kafka sink with transactions
    ✅ True exactly-once but higher latency

End-to-end exactly-once = replayable source
                        + Flink checkpoints
                        + transactional/idempotent sink

Two-Phase Commit Protocol (Kafka Sink Example):

Phase 1: Pre-commit (during checkpoint)
  ┌─────────────────────────────────────────────┐
  │ Checkpoint N triggered                       │
  │ Kafka Sink:                                  │
  │   1. Flush all buffered records to Kafka     │
  │   2. Records written but NOT committed       │
  │   3. Start new transaction for next records  │
  │   4. Report "pre-commit success" to JM       │
  └─────────────────────────────────────────────┘

Phase 2: Commit (after checkpoint confirmed)
  ┌─────────────────────────────────────────────┐
  │ JobManager confirms checkpoint N complete    │
  │ Kafka Sink:                                  │
  │   1. commitTransaction() on pre-committed tx │
  │   2. Records now visible to consumers        │
  │   3. (read_committed consumers see them now) │
  └─────────────────────────────────────────────┘

On failure (between pre-commit and commit):
  ┌─────────────────────────────────────────────┐
  │ TaskManager crashes after pre-commit         │
  │ Recovery:                                    │
  │   1. Restore from checkpoint N-1             │
  │   2. Abort uncommitted transaction           │
  │   3. Reprocess events from checkpoint N-1    │
  │   4. No duplicates! (aborted tx was invisible│
  │      to read_committed consumers)            │
  └─────────────────────────────────────────────┘

// Kafka sink with exactly-once (two-phase commit)
KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(
        KafkaRecordSerializationSchema.builder()
            .setTopic("output-topic")
            .setValueSerializationSchema(
                new SimpleStringSchema())
            .build())
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .setTransactionalIdPrefix("flink-job-1")
    .setProperty("transaction.timeout.ms", "900000") // 15 min
    .build();

// IMPORTANT: Kafka transaction.timeout.ms must be >
// checkpoint interval + checkpoint timeout
// Otherwise Kafka aborts the transaction before Flink commits!

// Consumer side: must use read_committed isolation
// props.put("isolation.level", "read_committed");