Fixing Flink SQL FileNotFoundException in S3 Incremental Checkpoints

by

Summary

A long-running Flink SQL job (running >1 year) failed during recovery, throwing a java.io.FileNotFoundException when attempting to restore from a checkpoint. The failure occurred specifically when the job tried to access a file in the shared directory of an incremental checkpoint stored on S3 (s3a://). This prevented the MiniBatchStreamingJoinOperator from initializing its keyed state, leading to a full job crash and preventing recovery from the latest successful checkpoint.

Root Cause

The root cause is state inconsistency caused by the deletion or corruption of shared incremental checkpoint data.

When using RocksDB with Incremental Checkpointing enabled, Flink does not upload the full state every time. Instead:

  • It uploads only the SST (Sorted String Table) files that have changed since the last checkpoint.
  • These files are stored in a shared directory to be reused across multiple checkpoint versions.
  • The failure java.io.FileNotFoundException: .../shared/34d2d0e5... indicates that the metadata for a specific checkpoint version pointed to a file in the shared/ folder that no longer exists on S3.

This typically happens due to:

  • Aggressive Lifecycle Policies: An S3 Lifecycle Rule (TTL) was configured to delete objects in the checkpoint directory to save costs, accidentally deleting files still referenced by active checkpoints.
  • Manual Cleanup: An engineer or an automated script manually deleted “old” files in the S3 bucket, unaware that incremental checkpoints rely on those “old” files for their current state.
  • S3 Consistency/Race Conditions: Though rare with modern S3, distributed systems can occasionally experience issues where metadata is written but the underlying object is not immediately available or is overwritten.

Why This Happens in Real Systems

In large-scale production environments, state management is often a battle between durability and storage costs.

  • State Bloat: High-throughput jobs with large joins (like this one) accumulate massive amounts of state. Over a year, this can reach terabytes.
  • Cost Optimization Pressure: Teams are often pressured to reduce S3 costs. This leads to the implementation of S3 Lifecycle Policies to transition objects to Glacier or delete them after $N$ days.
  • The Incremental Trap: Most developers understand that checkpoints are snapshots. However, they often fail to realize that incremental checkpoints are a graph of dependencies. A “new” checkpoint is actually a collection of new files plus a pointer to many “old” files in the shared directory. If you delete the “old” files, you break the new checkpoints.

Real-World Impact

  • Recovery Failure (The Death Spiral): The job cannot start from the latest checkpoint. If the corruption exists in all recent checkpoints, the job may be unable to restart at all without a cold start (losing all state).
  • Data Inconsistency: If the job is forced to restart from an older, valid checkpoint, it may reprocess massive amounts of Kafka data, leading to duplicate outputs or late-arriving data issues.
  • Operational Downtime: In a streaming pipeline, this stops all downstream consumers, potentially causing a massive backlog in Kafka and impacting real-time dashboards or downstream databases.

Example or Code (if necessary and relevant)

To prevent this, ensure your S3 Lifecycle configuration explicitly excludes the active checkpoint directory or is aware of Flink’s retention settings.

{
  "Rules": [
    {
      "ID": "DeleteOldCheckpoints",
      "Status": "Enabled",
      "Filter": {
        "Prefix": "flink/state/checkpoints/"
      },
      "Expiration": {
        "Days": 30
      }
    }
  ]
}

The configuration above is dangerous because it ignores the fact that incremental files in shared/ are needed by checkpoints created up to 30 days ago.

How Senior Engineers Fix It

  1. Audit S3 Lifecycle Policies: Immediately check for any Expiration or Transition rules on the S3 bucket that target the Flink state path.
  2. Implement “Safe” Retention: Instead of S3-level TTLs, rely on Flink’s internal checkpoint retention. Configure state.checkpoints.num-retained in flink-conf.yaml to manage how many checkpoints Flink keeps.
  3. Decouple Storage from Cleanup: If storage costs are too high, use a dedicated tool to identify orphaned files (files in shared/ not referenced by any _metadata file) rather than using blunt-force S3 expiration rules.
  4. Enable Versioning/Object Lock: For mission-critical state, enable S3 Versioning or Object Lock to prevent accidental deletions from automated scripts or human error.
  5. Recovery Strategy: If the latest checkpoint is corrupted, perform a point-in-time recovery to the last known healthy checkpoint. If all checkpoints are corrupted, prepare for a stateful reset (re-processing from Kafka offsets).

Why Juniors Miss It

  • Treating Checkpoints as Monolithic: Juniors often think a checkpoint is a single folder that can be deleted independently. They don’t realize the interdependency between the v1/, v2/ directories and the shared/ directory.
  • Focusing only on the Exception: A junior might see FileNotFoundException and assume it’s a network glitch or an S3 permission issue, rather than investigating the lifecycle of the data itself.
  • Infrastructure/Application Silos: Juniors often treat the Flink job and the S3 bucket as two unrelated entities. Senior engineers understand that the infrastructure configuration (S3) is a direct dependency of the application logic (Flink State).

Leave a Comment