Flink savepoint compatibility mode behavior after a breaking schema change – rollback safety?

Answer to the Main Question

In Flink 1.18+ the compatibility mode can only auto‑restore for backward‑compatible changes (e.g., adding fields or widening types). When a field is removed or its type is narrowed, the job must either keep the original savepoint and restore from it, or run a custom state migration before restarting the job. The compatibility mode does not provide automatic rollback for breaking changes.

Confirmed Facts

• The compatibility mode is controlled by state.checkpointing.mode and state.backend.checkpointing.mode but does not affect breaking schema changes.
• State Processor API can transform state between two job versions, but it does not guarantee transactional consistency across multiple state backends.
• Keeping an untouched original savepoint is the safest safety net; a migrated savepoint can be validated and used only when the transformation logic is verified.

Minimal Steps for This Case

1. Take a clean savepoint before the breaking change:

   flink savepoint --target-directory hdfs://... --savepoint <job-id> /path/to/savepoint

2. Write a state transformer: Implement StateTransformer that reads the old state format and writes the new one (or vice‑versa). Package it into a JAR.
3. Run the State Processor API:

   flink run -c org.apache.flink.stateprocessor.StateProcessorJob /path/to/transformer.jar \
     --checkpoint-dir hdfs://... --target-dir hdfs://... \
     --source-savepoint /path/to/savepoint --target-savepoint /path/to/migrated

4. Validate the migrated state: Deserialize a small sample of the migrated state (e.g., via a unit test) to confirm all expected fields are present and correctly typed.
5. Deploy the new job:

   flink run -c com.example.NewJob /path/to/new-job.jar --savepoint /path/to/migrated

6. Rollback if needed: If the new job fails or a revert is required, start the old job with the untouched original savepoint:

   flink run -c com.example.OldJob /path/to/old-job.jar --savepoint /path/to/savepoint

Validation Steps

• Run flink state-tool --metadata on both the original and migrated savepoints to verify state size and structure.
• Monitor Flink’s metrics endpoint for deserialization errors during a staged run.
• Ensure all keyed and operator state types are handled by the transformer.

Community‑Endorsed Practices

• Keep a copy of the original savepoint in a separate, version‑controlled storage bucket.
• Document the transformation logic in a Git repository and run it as part of CI before any schema change.
• Use a staging cluster to test the migrated job against a known input stream.
• If RocksDB is used, run rocksdb-check after migration to detect corruption.

Missing Diagnostic Detail

To tailor the migration script precisely, could you confirm whether your state backend is RocksDB or FsStateBackend? This choice can affect how the State Processor API reads and writes the state files.