[
https://issues.apache.org/jira/browse/FLINK-6014?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15948423#comment-15948423
]
ASF GitHub Bot commented on FLINK-6014:
---------------------------------------
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/3524
@StephanEwen Thanks very much for your valuable comments. The following
are some of my thoughts.
* Now the registration of shared states is put in `CheckpointCoordinator`
because it's needed whenever a `PendingCheckpoint` receives a state handle or a
`CompletedCheckpoint` is recovered. But I think it does make sense to put both
the registration and unregistration of shared states in the same place. I will
update the PR so that the logics are put in `PendingCheckpoint`s and
`CompletedCheckpoint`s.
* When a `SubtaskState` is not successfully added to the
`PendingCheckpoint`, the state objects in the `SubtaskState` should be
correctly deleted. The discarding of these `SubtaskState`s varies in different
cases. In the case where the `PendingCheckpoint` fails, the `SubtaskState`
should delete both its private states and shared states. But in the case where
the `CompletedCheckpoint` is subsumed, the `SubtaskState` should delete those
unreferenced shared states (possibly created by others) instead of its shared
states.
By registering the shared states first, we can unify the implementation
in the two cases. Those shared states in the failed `PendingCheckpoint` are
always not referenced by other checkpoints. So they can be correctly discarded
by the registry when the `PendingCheckpoint` unregisters its shared states,
just like a subsumed `CompletedCheckpoint` does.
Another choice is refactoring the interface of `CompositeStateHandle`.
Three methods, namely `onComplete()`, `onFail()` and `onSubsume()`, will be
provided. A`CompositeStateHandle` can implement these methods to correctly deal
with its states under these cases. What do you think?
* It's a good idea to introduce `SharedStateHandle` for shared states. It
can improve the performance and allow safety checks. I will add it in the
update.
> 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)
