[
https://issues.apache.org/jira/browse/FLINK-6014?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15939966#comment-15939966
]
ASF GitHub Bot commented on FLINK-6014:
---------------------------------------
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/3524
@StephanEwen I have updated the PR as suggested. Changes include
1. Make `StateRegistry` to be `SharedStateRegistry` where only shared
states are registered. Now the `discardState()` method is supposed to delete
those private states in the checkpoint.
2. `SharedStateRegistry` now is deployed by the `CheckpointCoordinator`.
The state handle will register its shared states once it is received by the
coordinator. In another words, all shared states in a completed checkpoint are
registered when the checkpoint is added into the `CompletedCheckpointStore`.
All checkpoints (including savepoints) will unregister shared states when they
are removed from the store. Savepoints should not contain any shared state.
Therefore the unregistration will not discard any state in the savepoints.
3. When recovering from failures or restarting from a savepoint, the
`CheckpointCoordinator` will rebuild the registry with the checkpoints
recovered in the `CompletedCheckpointStore`.
4. Related tests are added to ensure correctness.
5. The conflicts with the master branch are resolved.
> Allow the registration of state objects in checkpoints
> ------------------------------------------------------
>
> Key: FLINK-6014
> URL: https://issues.apache.org/jira/browse/FLINK-6014
> Project: Flink
> Issue Type: Sub-task
> Components: State Backends, Checkpointing
> Reporter: Xiaogang Shi
> Assignee: Xiaogang Shi
>
> This issue is the very first step towards incremental checkpointing. We
> introduce a new state handle named {{CompositeStateHandle}} to be the base of
> the snapshots taken by task components. Known implementation may include
> {{KeyedStateHandle}} (for {{KeyedStateBackend}}s), {{SubtaskState}} (for
> subtasks, splits of {{JobVertex}}) and {{TaskState}} (for {{JobVertex}}s).
> Each {{CompositeStateHandle}} is composed of a collection of {{StateObject}s.
> It should register all its state objects in {{StateRegistry}} when its
> checkpoint is added into {{CompletedCheckpointStore}} (i.e., a pending
> checkpoint completes or a complete checkpoint is reloaded in the recovery).
> When a completed checkpoint is moved out of the {{CompletedCheckpointStore}},
> we should not simply discard all state objects in the checkpoint. With the
> introduction of incremental checkpointing, a {{StateObject}} may be
> referenced by different checkpoints. We should unregister all the state
> objects contained in the {{StateRegistry}} first. Only those state objects
> that are not referenced by any checkpoint can be deleted.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
