You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by wangzhijiang999 <gi...@git.apache.org> on 2017/02/17 09:17:18 UTC
[GitHub] flink pull request #3340: [FLINK-5703][runtime]Job manager failure recovery ...
GitHub user wangzhijiang999 opened a pull request:
https://github.com/apache/flink/pull/3340
[FLINK-5703][runtime]Job manager failure recovery via reconciliation with TaskManager reports
This is part of [Non-disruptive JobManager Failures via Reconciliation ](https://issues.apache.org/jira/browse/FLINK-4911).
The design doc for this part is attached in the JIRA and it mainly contains the following work:
- **RECONCILING** state transition for job status and execution.
- The data structure recovery for **ExecutionGraph**, **ExecutionVertex**,**Execution** based on **TaskManager** reports.
Two parts are left for the whole feature:
- The related modifications in **TaskManger** side will be submitted in another JIRA, including not cancel the tasks when notified job leader changed and report the task status to **JobManager**, etc.
- This PR will not make an effect on the current master branch, so it is safe to merge. The reconcile logic should be triggered by **JobManagerRunner** when grants leadership, but it is dependent on [Determine whether the job starts from last JobManager failure](https://issues.apache.org/jira/browse/FLINK-5501) , so I will modify the **JobManagerRunner** logic after [FLINK-5501](https://issues.apache.org/jira/browse/FLINK-5501) is merged.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/wangzhijiang999/flink FLINK-5703
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/3340.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #3340
----
commit 7337754bd8d12808a328b55e184b0fe59cf6e11d
Author: \u6dd8\u6c5f <ta...@alibaba-inc.com>
Date: 2017-02-17T08:41:44Z
[FLINK-5703][runtime]Job manager failure recovery based on reconciliation with TaskManager reports
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3340: [FLINK-5703][runtime] ExecutionGraph recovery via reconci...
Posted by zhijiangW <gi...@git.apache.org>.
Github user zhijiangW commented on the issue:
https://github.com/apache/flink/pull/3340
@StephanEwen , thanks for professional reviews!
I attached the design doc in this jira for the implementation consideration.
[https://docs.google.com/document/d/1rm3BYQyC-8GrLgdwSUAbkJ0aV8IaCneF6c5dk4G3SXY/edit](https://docs.google.com/document/d/1rm3BYQyC-8GrLgdwSUAbkJ0aV8IaCneF6c5dk4G3SXY/edit)
For your above suggestions:
1. I totally agree with your point of keeping the current final variables related with **Execution** and **IntermediateResult**, etc. Although I already tried to avoid concurrent issues for these variables in implementation, it is still a good choice of creating the new ones to replace the previous ones, and I am willing to re-implement the related logics for this good idea.
2. In the process of JobManagerRunner **grantLeadership**, considering not expose the new job leader to TaskManagers before reconcile, so the reconciliation process in JobManager will be called before **confirmLeadership** to avoid this issue. As you suggest, it can also make sense if creating JobManager and ExecutionGraph during reconciliation. Considering minimum changes, I will keep the current mode for this issue if no other reasons. What do you think?
3. Two phases of the reconciliation mainly concerns about some actions during state transition. For example, if transition from **RECONCILING** to **RUNNING**, it should **sendPartitionInfos** in previous logic and this action relies on other upstream executions reconciliation. In other words, the upstream execution reconciliation would try to cache the partition infos in the **PartialInputChannelDeploymentDescriptor** of downstream execution. And when the downstream execution switches to **RUNNING**, this partition info will be sent to it. Because we can not confirm all the sequences of execution reconciliation, dividing the whole process into two steps can make sense for that.
But after your latest improvements for **EAGER** **ScheduleMode**, I think the downstream task is submitted already with final upstream partition infos, so maybe it is no need to send partition infos for running downstream task again. If my understanding is correct, I can re-implement this part for merging the current two phases into one based on TaskManager reports.
In addition , for FAILED or CANCELED reported execution states, it would fail the whole ExecutionGraph as a result, so we can end the reconciliation period immediately, then the following reports will be refused and the corresponding tasks will be failed by TaskManagers. From JobManager side, when process the first FAILED or CANCELED reported state to trigger failing the ExecutionGraph, the other RECONCILING state executions will be transition to CANCELED state directly, no need to send cancel rpc message. What do you think?
I will finish the above modifications during this week, and welcome any other advices!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3340: [FLINK-5703][runtime] ExecutionGraph recovery via ...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/3340#discussion_r102995512
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java ---
@@ -440,6 +440,72 @@ public void testSendCancelAndReceiveFail() {
}
}
+ /**
+ * For job manager failure recovery case, the execution may still in reconciling state but already recovered
+ * basic information including slot, when process the failed execution, it will trigger to cancel all the current
+ * executions. It is necessary to send cancel rpc to reconciling state execution with slot because the task manger
+ * already reports its status for recovery.
+ */
+ @Test
+ public void testCancelFromReconcilingWithSlot() {
+ try {
+ final JobVertexID jid = new JobVertexID();
+ final ExecutionJobVertex ejv = getExecutionVertex(jid, new DirectScheduledExecutorService());
+ final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0], AkkaUtils.getDefaultTimeout());
+ final ActorGateway actorGateway = new CancelSequenceActorGateway(TestingUtils.directExecutionContext(), 1);
+
+ final Instance instance = getInstance(new ActorTaskManagerGateway(actorGateway));
+ final SimpleSlot slot = instance.allocateSimpleSlot(new JobID());
+
+ setVertexState(vertex, ExecutionState.RECONCILING);
+ setVertexResource(vertex, slot);
+
+ assertEquals(ExecutionState.RECONCILING, vertex.getExecutionState());
+
+ vertex.cancel();
+ vertex.getCurrentExecutionAttempt().cancelingComplete(); // response by task manager once actually canceled
+
+ assertEquals(ExecutionState.CANCELED, vertex.getExecutionState());
+ assertTrue(slot.isReleased());
+ assertNull(vertex.getFailureCause());
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CREATED) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELING) > 0);
+ assertTrue(vertex.getStateTimestamp(ExecutionState.CANCELED) > 0);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * For job manager failure recovery case, the execution may still in reconciling state because the task manager
+ * does not report its status within duration time. It is no need to send cancel rpc for such execution with no real
+ * attempt id and slot. And it can be transition to canceled state directly, the same with the cases of scheduled or created.
+ */
+ @Test
+ public void testCancelFromReconcilingNoSlot() {
+ try {
--- End diff --
We are trying to avoid this pattern now (we used it in the earlier days).
It is better for logging and debugging to simply declare `throws Exception` on the test method.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3340: [FLINK-5703] [Job Manager] ExecutionGraph recovery via re...
Posted by zhijiangW <gi...@git.apache.org>.
Github user zhijiangW commented on the issue:
https://github.com/apache/flink/pull/3340
@StephanEwen , sorry for delay update of this feature.
This modification covers almost all the work on job manager side concerning about the suggestions you mentioned before, including:
1. Once `JobManagerRunner` granting leadership, it will check the job status by `RunningJobsRegistry`. If the job is already running, then the `JobManager` will enter reconciling process and confirm leadership last.
To do so, we can confirm that the `TaskManager` will not report task status before `JobManager` reconciles, and it is also reasonable for 'JobManager` exposing its address to outside `TaskManager` after internal initialization. To keep the behavior consistency, I modified the `EmbeddedLeaderService` to set leader flag as true in the process of `grantLeadership` instead of 'confirmLeadership`.
2. 'ExecutionGraph' recovery based on task status report form `TaskManager`, and I created the new execution attempt to replace the current one for recovery, also the same case for `IntermediateResultPartition` to avoid updating the final fields directly in previous way. And also integrating two phases of recovery into one phase you mentioned before.
3. After duration time, the `ExecutionGraph` decides to enter `RUNNING` or `FAILING` from `RECONCILING` based on reconciliation results.
Currently I have not considered the improvements for `LAZY_FROM_SOURCES` schedule mode in this version, because it needs to remember how many vertexes have already been scheduled before `JobManager` failure. Maybe we can improve it in next version if needed.
I may supplement some extra tests and update the attached design doc after you approve this realization, and the 'TaskManager' part will be submitted in another separate PR which mainly includes:
1. `TaskManager` will not fail task directly after notified of `JobManager` leader lost, and triggers a timer to later check whether reconnects to the new job leader successfully.
2. If reconnection success, it will report the current internal task and slot status that belongs to this job leader, otherwise it will fail the tasks as the current behavior.
Wish your further response, and I will begin to consider resource issues of FLIP-6, thank you!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3340: [FLINK-5703][runtime] ExecutionGraph recovery via ...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/3340#discussion_r102993824
--- Diff: flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java ---
@@ -110,7 +110,13 @@
public static final String EXECUTION_RETRY_DELAY_KEY = "execution-retries.delay";
// -------------------------------- Runtime -------------------------------
-
+
+ /**
+ * The maximum time job manager defines to wait for task manager reporting task status, if exceed this duration,
+ * job manager will refuse the reporting.
+ */
+ public static final String JOB_MANAGER_RECONCILE_DURATION = "jobmanager.reconcile-duration";
--- End diff --
Can we add this to the `JobManagerOptions` instead?
https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManagerOptions.java
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3340: [FLINK-5703][runtime] ExecutionGraph recovery via reconci...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/3340
There is a lot of good code in this PR.
What I would suggest to make different is to NOT make `Execution`, `IntermediateResult` partition, etc mutable. There is a big benefit to having them immutable, both for safety against accidental bugs, and also from a concurrency standpoint. Regular mutable non-volatile variables have visibility problems in concurrent access, which leads to very subtle bugs that are hard to trace and recover: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/package-summary.html#MemoryVisibility
For `Execution`, I would change it such that the `ExecutionVertex` swaps the current `Execution` (in state `RECONCILING`) with the a new `Execution` that has all the fields properly set. That way there is a quasi atomic replacement of the previous execution, which is safer and easier to reason about.
It may be worth to even create the JobManager (from the JobManagerRunner) and the ExecutionGraph directly in reconciliation mode. That way, you don't have to worry what to do with TaskManagers that register while the JobMaster has not yet switched to reconciliation.
Can you explain a bit why the reconciliation need the two phases:
1. Receive reports, set the execution attempt and state
2. Once all are there, transition the state to running
Could it be done such that on report the `Execution` is directly set to the reported state? Then once all reports are there, everything is just fine. After the reconciliation timeout expires, you would just iterate over all executions and `fail()` the ones that are still in `RECONCILE` - recovery logic takes care of the rest.
That may also handle the case where some already reconciled tasks fail while reconciliation still happens. The ExecutionGraph would already recover the local tasks that are affected by the failure.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3340: [FLINK-5703][runtime]Job manager failure recovery via rec...
Posted by wangzhijiang999 <gi...@git.apache.org>.
Github user wangzhijiang999 commented on the issue:
https://github.com/apache/flink/pull/3340
@StephanEwen , wish your reviews. Thank you for any comments!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---