[jira] [Comment Edited] (FLINK-10930) Refactor checkpoint directory layout

["Stephan Ewen (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-10930?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16712017#comment-16712017 ] 

Stephan Ewen edited comment on FLINK-10930 at 12/6/18 8:53 PM:
---------------------------------------------------------------

[~yunta] Better exception handling is definitely always a really good idea. I would suggest to proceed as follows:

 - We work towards a model where checkpoint failures never cause task failures, but only "decline messages". As fas as I know, [~azagrebin] mentioned that there is some discussion already. Could you chime in here, Andrey?

 - We fix all cases of state being left in exception cases that we find.

 - We introduce an exclusive state cleanup daemon (see below)

 - We introduce a shared state cleanup daemon (see below)

h3. Cleaning up exclusive state (idea)
 - when the daemon starts a cleanup sweep it determines what the oldest still retained checkpoint number is
 - it enumerates all directories "CHK-XXX" and recursively deletes the ones older than the oldest retained checkpoint

h3. Cleaning up shared state (idea)
 - every shared state file needs to encode the checkpoint-id of the checkpoint during which it was created
 - when the daemon does a cleanup swipe, it atomically grabs the number of the latest completed checkpoint, and a snapshot of all shared state handles in the shared state registry
 - the daemon then lists all files in the shared state directory and deletes all files that are neither referenced by a shared state handle and are older than the latest completed checkpoint

What do you think about that?


was (Author: stephanewen):
[~yunta] Better exception handling is definitely always a really good idea. I would suggest to proceed as follows:

  - We work towards a model where checkpoint failures never cause task failures, but only "decline messages". As fas as I know, [~azagrebin] mentioned that there is some discussion already. Could you chime in here, Andrey?

  - We fix all cases of state being left in exception cases that we find.

  - We introduce an exclusive state cleanup daemon (see below)

  - We introduce a shared state cleanup daemon (see below)

h3. Cleaning up exclusive state (idea)

  - when the daemon starts a cleanup sweep it determines what the oldest still retained checkpoint number is
  - it enumerates all directories "CHK-XXX" and recursively deletes the ones older than the oldest retained checkpoint


h3. Cleaning up shared state (idea)

  - every shared state file needs to encode the checkpoint-id of the checkpoint during which it was created
  - when the daemon does a cleanup swipe, it atomically grabs the number of the latest completed checkpoint, and a snapshot of all shared state handles in the shared state registry
  - the daemon then lists all files in the shared state directory and deletes all files that are neither referenced by a shared state handle and are older than the latest completed checkpoint


What do you think about that?


> Refactor checkpoint directory layout
> ------------------------------------
>
>                 Key: FLINK-10930
>                 URL: https://issues.apache.org/jira/browse/FLINK-10930
>             Project: Flink
>          Issue Type: Improvement
>          Components: State Backends, Checkpointing
>    Affects Versions: 1.8.0
>            Reporter: Yun Tang
>            Assignee: Yun Tang
>            Priority: Major
>             Fix For: 1.8.0
>
>
> The current checkpoint directory layout is introduced from FLINK-8531 with three different scopes for tasks:
>  * *EXCLUSIVE* is for state that belongs to one checkpoint only, meta data and operator state files.
>  * *SHARED* is for state that is possibly part of multiple checkpoints
>  * *TASKOWNED* is for state that must never by dropped by the jobManager.
> {code:java}
> /user-defined-dir/{job-id}
> 		    |
> 		    +-- shared/
> 	            +-- taskowned/
> 		    +-- chk-1/      // metadata and operator-state files
> 	            +-- chk-2/
>                     ...{code}
> If we just retain one complete checkpoint, the expected exclusive directory, which is the {{chk-id}} checkpoint directory, should only be one left. However, as FLINK-10855 interpreted, the failed/expired checkpoint directories would also be left. This is really confusing for users who [uses externalized checkpoint to resume job|https://ci.apache.org/projects/flink/flink-docs-release-1.6/ops/state/checkpoints.html#resuming-from-a-retained-checkpoint], not to mention the checkpoint directory resource leak. 
>  As far as I could know, if the {{chk-id}} checkpoint directory still contains the operator state files, I have no idea how to clean the useless {{chk-id}} checkpoint directory gracefully. Once job manager dispose the failed/expired checkpoint, the target {{chk-id}} checkpoint directory would be deleted by JM. However, this directory would also be create by tasks who having not reported to JM. When {{checkpoint coordinator}} received those late expired tasks, it would discard those useless handles. However, if JM also plans to delete the empty parent folder, which is already unsupported after FLINK-8540, another task uploading operator state files would meet exception due to its writing target's parent directory has just been removed. Currently, we handle task checkpoint failure as task failure and the whole job would failover which is not we want.
> From what I see, I plan to separate *EXCLUSIVE* directory into two kind of exclusive directories, one is still several {{chk-id}} checkpoint directories but only contains its exclusive {{meta data}}, the other is just one directory named {{exclusive}} which containing the operator state files. Operator state files are exclusive to just one specified checkpoint, we could also add {{checkpoint-id}} within their file name to let users easily clean up.
>  The refactored directory layout should be :
> {code:java}
> /user-defined-dir/{job-id}
>                     |
> 		    +-- shared/
> 		    +-- taskowned/
> 	            +-- exclusive/    // operator state files
> 		    +-- chk-1/        // metadata
>                     +-- chk-2/
>                     ...{code}
>  
> This new directory layout would not affect users who use external checkpoint to resume jobs, since they still just give {{/user-defined-dir/job-id/chk-id}} path to resume job.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)