Flink try to recover checkpoint from deleted directories

Viewed 152

After cleaning the s3 bucket, which is used to store checkpoints from old files (files that have been accessed for more than a month), when restarting or when restoring from actual checkpoints, some processes of job do not start due to some missing old files

Job works well and save actual checkpoins (save path s3://flink-checkpoints/check/af8b0712ae0c1f20d2226b86e6bddb60/chk-100274)

2022-04-24 03:58:32.892 Triggering checkpoint 100273 @ 1653353912890 for job af8b0712ae0c1f20d2226b86e6bddb60.
2022-04-24 03:58:55.317 Completed checkpoint 100273 for job af8b0712ae0c1f20d2226b86e6bddb60 (679053131 bytes in 22090 ms).
2022-04-24 04:03:32.892 Triggering checkpoint 100274 @ 1653354212890 for job af8b0712ae0c1f20d2226b86e6bddb60.
2022-04-24 04:03:35.844 Completed checkpoint 100274 for job af8b0712ae0c1f20d2226b86e6bddb60 (9606712 bytes in 2494 ms).

After one taskmanager switched off and job restarted

2022-04-24 04:04:40.936 Job test-job (af8b0712ae0c1f20d2226b86e6bddb60) switched from state RUNNING to RESTARTING.
2022-04-24 04:05:14.150 Job test-job (af8b0712ae0c1f20d2226b86e6bddb60) switched from state RESTARTING to RUNNING.
2022-04-24 04:05:14.198 Restoring job af8b0712ae0c1f20d2226b86e6bddb60 from latest valid checkpoint: Checkpoint 100274 @ 1653354212890 for af8b0712ae0c1f20d2226b86e6bddb60.

after some time job failed because some process can't restore state

2022-04-24 04:05:17.095 Job test-job (af8b0712ae0c1f20d2226b86e6bddb60) switched from state RUNNING to RESTARTING.
2022-04-24 04:05:17.093 Process first events -> Sink: Sink to test-job (5/10) (4f9089b1015540eb6e13afe4c07fa97b) switched from RUNNING to FAILED.
java.lang.Exception: Exception while creating StreamOperatorStateContext.
Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for KeyedProcessOperator_f1d5710fb330fd579d15b292e305802c_(5/10) from any of the 1 provided restore options.
Caused by: org.apache.flink.runtime.state.BackendBuildingException: Caught unexpected exception.
Caused by: org.apache.flink.util.FlinkRuntimeException: Failed to download data for state handles.
Caused by: com.facebook.presto.hive.s3.PrestoS3FileSystem$UnrecoverableS3OperationException: com.amazonaws.services.s3.model.AmazonS3Exception: null (Service: Amazon S3; Status Code: 404; Error Code: NoSuchKey; Request ID: tx0000000000000f0652d11-00628c2f4a-51f03da-default; S3 Extended Request ID: 51f03da-default-default), S3 Extended Request ID: 51f03da-default-default (Path: s3://flink-checkpoints/check/e3d82336005fc40be9af536938716199/shared/64452a30-c8a0-454f-8164-34d9e70142e0)
Caused by: com.amazonaws.services.s3.model.AmazonS3Exception: null (Service: Amazon S3; Status Code: 404; Error Code: NoSuchKey; Request ID: tx0000000000000f0652d11-00628c2f4a-51f03da-default; S3 Extended Request ID: 51f03da-default-default)
2022-04-24 04:05:17.095 Job test-job (af8b0712ae0c1f20d2226b86e6bddb60) switched from state RUNNING to RESTARTING.

if I completely cancel the job and start a new one with the savepoint set to the path of the last checkpoint, I get the same errors.

Why when working with checkpoint from af8b0712ae0c1f20d2226b86e6bddb60 folder job tries to get some files from e3d82336005fc40be9af536938716199 folder and what are the rules for clearing old checkpoints from storage?

UPDATE I found that flink save s3 paths for all TaskManager's rocksdb files in chk-*/_metadata file.

3 Answers

This is something that was quite ambiguous for a long time and has recently been addressed in Flink 1.15. I would recommend reading https://flink.apache.org/news/2022/05/05/1.15-announcement.html on the section 'Clarification of checkpoint and savepoint semantics', including the section where the comparison between checkpoints and savepoints is made.

The behaviour you've experienced depends on your checkpointing setup (aligned vs unaligned).

By default cancelling job removes old checkpoints. There is a configuration flag to control it execution.checkpointing.externalized-checkpoint-retention. As mentioned by Martijn, normally you would resort to savepoints for controlled job upgrades / restarts.

I found that flink save s3 paths for all TaskManager's rocksdb files in chk-*/_metadata file.

Related