[
https://issues.apache.org/jira/browse/FLINK-10712?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16699080#comment-16699080
]
ASF GitHub Bot commented on FLINK-10712:
----------------------------------------
StefanRRichter commented on a change in pull request #7009: [FLINK-10712]
Support to restore state when using RestartPipelinedRegionStrategy
URL: https://github.com/apache/flink/pull/7009#discussion_r236285508
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
##########
@@ -105,13 +106,70 @@ public boolean assignStates() throws Exception {
continue;
}
- assignAttemptState(task.getValue(), operatorStates);
+ Set<Integer> executionVertexIndices = new HashSet<>();
+ for (ExecutionVertex executionVertex :
task.getValue().getTaskVertices()) {
+
executionVertexIndices.add(executionVertex.getParallelSubtaskIndex());
+ }
+ assignAttemptState(task.getValue(), operatorStates,
executionVertexIndices);
+ }
+
+ return true;
+ }
+
+ /**
+ * Assign states to given execution vertices.
+ */
+ public boolean assignStates(List<ExecutionVertex> executionVertices)
throws Exception {
+ Map<OperatorID, OperatorState> localOperators = new
HashMap<>(operatorStates);
+ Map<JobVertexID, ExecutionJobVertex> localTasks = this.tasks;
+
+ checkStateMappingCompleteness(allowNonRestoredState,
operatorStates, tasks);
+
+ // get job vertex and its subTaskIndex from given
executionVertices.
+ Map<JobVertexID, Set<Integer>> jobVertexIDSetMap = new
HashMap<>();
+ for (ExecutionVertex executionVertex : executionVertices) {
+ JobVertexID jobvertexId =
executionVertex.getJobvertexId();
+ jobVertexIDSetMap.putIfAbsent(jobvertexId, new
HashSet<>());
+
jobVertexIDSetMap.get(jobvertexId).add(executionVertex.getParallelSubtaskIndex());
+ }
+
+ for (Map.Entry<JobVertexID, ExecutionJobVertex> task :
localTasks.entrySet()) {
+ final ExecutionJobVertex executionJobVertex =
task.getValue();
+
+ // find the states of all operators belonging to this
task
+ List<OperatorID> operatorIDs =
executionJobVertex.getOperatorIDs();
+ List<OperatorID> altOperatorIDs =
executionJobVertex.getUserDefinedOperatorIDs();
+ List<OperatorState> operatorStates = new ArrayList<>();
+ boolean statelessTask = true;
+ for (int x = 0; x < operatorIDs.size(); x++) {
+ OperatorID operatorID = altOperatorIDs.get(x)
== null
+ ? operatorIDs.get(x)
+ : altOperatorIDs.get(x);
+
+ OperatorState operatorState =
localOperators.remove(operatorID);
+ if (operatorState == null) {
+ operatorState = new OperatorState(
+ operatorID,
+
executionJobVertex.getParallelism(),
+
executionJobVertex.getMaxParallelism());
+ } else {
+ statelessTask = false;
+ }
+ operatorStates.add(operatorState);
+ }
+ if (statelessTask) { // skip tasks where no operator
has any state
+ continue;
+ }
+
+ if
(jobVertexIDSetMap.containsKey(executionJobVertex.getJobVertexId())) {
+ assignAttemptState(executionJobVertex,
operatorStates, jobVertexIDSetMap.get(executionJobVertex.getJobVertexId()));
+ }
}
return true;
}
- private void assignAttemptState(ExecutionJobVertex executionJobVertex,
List<OperatorState> operatorStates) {
+ private void assignAttemptState(ExecutionJobVertex executionJobVertex,
List<OperatorState> operatorStates, Set<Integer> subTaskIndices) {
Review comment:
I doubt that `Set<Integer>` is the best representation of subtask indexes.
At least from the interface leven, an `Iterable<Integer>` could do the job if
we rewite the loop as I suggested. Forthermore, we can have a more memory
friendly implementation to back this, for example `boolean[]` or `Bitset`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> RestartPipelinedRegionStrategy does not restore state
> -----------------------------------------------------
>
> Key: FLINK-10712
> URL: https://issues.apache.org/jira/browse/FLINK-10712
> Project: Flink
> Issue Type: Bug
> Components: Distributed Coordination
> Affects Versions: 1.3.3, 1.4.2, 1.5.5, 1.6.2, 1.7.0
> Reporter: Stefan Richter
> Assignee: Yun Tang
> Priority: Critical
> Labels: pull-request-available
> Fix For: 1.8.0
>
>
> RestartPipelinedRegionStrategy does not perform any state restore. This is
> big problem because all restored regions will be restarted with empty state.
> We need to take checkpoints into account when restoring.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
