[FLINK-10255] Only react to onAddedJobGraph signal when being leader #6678
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
tisonkun
reviewed
Sep 11, 2018
| @@ -85,5 +87,13 @@ | |||
| return FutureUtils.completeAll(terminationFutures); | |||
| } | |||
|
|
|||
| public static void stopActor(AkkaActorGateway akkaActorGateway) { | |||
There was a problem hiding this comment.
In this PR we introduce stopActor which is used at one place. After checking all our project, we have define many stopActor here and there. Most usages of them are from TestingUtils but there are also some from MesosResourceManager and FlinkUntypedActorTest. Sometimes use PoisonPill and sometimes use Kill.
Apart from this PR, since all stuff interact with Akka would depend on flink-runtime, let's unify stopActor Utils.
I think here, ActorUtils is the best place.
There was a problem hiding this comment.
Good point. I will clean this up as a follow up.
|
Travis show relevant failures, will take a close look later. |
|
Thanks for the comments @tisonkun. I've fixed the failing |
GJL
reviewed
Sep 12, 2018
| @@ -0,0 +1,48 @@ | |||
| /* | |||
There was a problem hiding this comment.
I think adding this file should be in a commit that is before [FLINK-10255] Only react to onAddedJobGraph signal when being leader, or it should be squashed. Without this class your previous commits would not work.
GJL
reviewed
Sep 12, 2018
| return callAsyncWithoutFencing( | ||
| () -> getJobTerminationFuture(jobId), | ||
| timeout).thenCompose(Function.identity()); | ||
| } | ||
|
|
||
| @VisibleForTesting |
There was a problem hiding this comment.
Not sure about @VisibleForTesting here. This class is already a test utility. It is even in the test sources directory.
There was a problem hiding this comment.
You're right. Will change it.
| @@ -861,14 +861,18 @@ public void grantLeadership(final UUID newLeaderSessionID) { | |||
| getMainThreadExecutor()); | |||
| } | |||
|
|
|||
| protected CompletableFuture<Void> getJobTerminationFuture(JobID jobId) { | |||
| CompletableFuture<Void> getJobTerminationFuture(JobID jobId) { | |||
There was a problem hiding this comment.
It cannot be private since the TestingDispatcher needs to access it.
| if (jobManagerRunners.containsKey(jobId)) { | ||
| return FutureUtils.completedExceptionally(new DispatcherException(String.format("Job with job id %s is still running.", jobId))); | ||
| } else { | ||
| return jobManagerTerminationFutures.getOrDefault(jobId, CompletableFuture.completedFuture(null)); | ||
| } | ||
| } | ||
|
|
||
| CompletableFuture<Void> getRecoveryOperation() { |
There was a problem hiding this comment.
This method has wider visibility scope than necessary, and is part of production code. I think @VisibleForTesting should be added.
There was a problem hiding this comment.
Good point. Will add it.
tillrohrmann
force-pushed
the
fixJobRecovery
branch
from
868c7dd
to
d8c544f
Compare
GJL
reviewed
Sep 12, 2018
| final CompletableFuture<Void> submissionFuture = recoveredJob.thenComposeAsync( | ||
| (FunctionWithThrowable<JobGraph, CompletableFuture<Void>, Exception>) (JobGraph jobGraph) -> tryRunRecoveredJobGraph(jobGraph, dispatcherId) | ||
| .thenAcceptAsync( | ||
| (ConsumerWithException<Boolean, Exception>) (Boolean isRecoveredJobRunning) -> { |
There was a problem hiding this comment.
Imo we are doing this wrong. The code would be much more readible with static factory methods:
/**
* {@link Consumer} that can throw checked exceptions.
*/
@FunctionalInterface
public interface CheckedConsumer<T> {
void checkedAccept(T t) throws Exception;
static <T> Consumer<T> unchecked(CheckedConsumer<T> checkedConsumer) {
return (t) -> {
try {
checkedConsumer.checkedAccept(t);
} catch (Exception e) {
ExceptionUtils.rethrow(e);
}
};
}
}
This allows for:
.thenAcceptAsync(CheckedConsumer.unchecked(isRecoveredJobRunning -> {
...
}));
...
No casts are required. Also when interacting with the Java API, it does not matter what exact type of exception can be thrown – what matters is that the checked exception becomes unchecked. We do not need to generify the exception type in ConsumerWithException.
There was a problem hiding this comment.
Good point. I like this approach better. Will adapt the existing interfaces.
GJL
reviewed
Sep 12, 2018
|
|
||
| final DispatcherId dispatcherId = getFencingToken(); | ||
| final CompletableFuture<Void> submissionFuture = recoveredJob.thenComposeAsync( | ||
| (FunctionWithThrowable<JobGraph, CompletableFuture<Void>, Exception>) (JobGraph jobGraph) -> tryRunRecoveredJobGraph(jobGraph, dispatcherId) |
There was a problem hiding this comment.
See my comment regarding the ConsumerWithException.
GJL
reviewed
Sep 12, 2018
| @@ -167,7 +181,7 @@ public void testSubmittedJobGraphRelease() throws Exception { | |||
| // recover the job | |||
| final SubmittedJobGraph submittedJobGraph = otherSubmittedJobGraphStore.recoverJobGraph(jobId); | |||
|
|
|||
| assertThat(submittedJobGraph, Matchers.is(Matchers.notNullValue())); | |||
| assertThat(submittedJobGraph, is(Matchers.notNullValue())); | |||
There was a problem hiding this comment.
You added a static import for is but not for notNullValue. I think this should be consistent.
There was a problem hiding this comment.
Good catch. Will change it.
GJL
approved these changes
Sep 12, 2018
tillrohrmann
force-pushed
the
fixJobRecovery
branch
from
d8c544f
to
0788a54
Compare
|
Thanks for the review @GJL. I've addressed your comments and after Travis gives green light, I'll merge it. |
tillrohrmann
force-pushed
the
fixJobRecovery
branch
3 times, most recently
from
f255bb2
to
3d94a81
Compare
tillrohrmann
added a commit
to tillrohrmann/flink
that referenced
this pull request
Sep 13, 2018
The Dispatcher should only react to the onAddedJobGraph signal if it is the leader. In all other cases the signal should be ignored since the jobs will be recovered once the Dispatcher becomes the leader. In order to still support non-blocking job recoveries, this commit serializes all recovery operations by introducing a recoveryOperation future which first needs to complete before a subsequent operation is started. That way we can avoid race conditions between granting and revoking leadership as well as the onAddedJobGraph signals. This is important since we can only lock each JobGraph once and, thus, need to make sure that we don't release a lock of a properly recovered job in a concurrent operation. This closes apache#6678.
tillrohrmann
added a commit
to tillrohrmann/flink
that referenced
this pull request
Sep 13, 2018
The Dispatcher should only react to the onAddedJobGraph signal if it is the leader. In all other cases the signal should be ignored since the jobs will be recovered once the Dispatcher becomes the leader. In order to still support non-blocking job recoveries, this commit serializes all recovery operations by introducing a recoveryOperation future which first needs to complete before a subsequent operation is started. That way we can avoid race conditions between granting and revoking leadership as well as the onAddedJobGraph signals. This is important since we can only lock each JobGraph once and, thus, need to make sure that we don't release a lock of a properly recovered job in a concurrent operation. This closes apache#6678.
tillrohrmann
force-pushed
the
fixJobRecovery
branch
from
3d94a81
to
6fb8bc2
Compare
tillrohrmann
added a commit
to tillrohrmann/flink
that referenced
this pull request
Sep 13, 2018
The Dispatcher should only react to the onAddedJobGraph signal if it is the leader. In all other cases the signal should be ignored since the jobs will be recovered once the Dispatcher becomes the leader. In order to still support non-blocking job recoveries, this commit serializes all recovery operations by introducing a recoveryOperation future which first needs to complete before a subsequent operation is started. That way we can avoid race conditions between granting and revoking leadership as well as the onAddedJobGraph signals. This is important since we can only lock each JobGraph once and, thus, need to make sure that we don't release a lock of a properly recovered job in a concurrent operation. This closes apache#6678.
tillrohrmann
added a commit
to tillrohrmann/flink
that referenced
this pull request
Sep 13, 2018
The Dispatcher should only react to the onAddedJobGraph signal if it is the leader. In all other cases the signal should be ignored since the jobs will be recovered once the Dispatcher becomes the leader. In order to still support non-blocking job recoveries, this commit serializes all recovery operations by introducing a recoveryOperation future which first needs to complete before a subsequent operation is started. That way we can avoid race conditions between granting and revoking leadership as well as the onAddedJobGraph signals. This is important since we can only lock each JobGraph once and, thus, need to make sure that we don't release a lock of a properly recovered job in a concurrent operation. This closes apache#6678.
tillrohrmann
added a commit
to tillrohrmann/flink
that referenced
this pull request
Sep 13, 2018
The Dispatcher should only react to the onAddedJobGraph signal if it is the leader. In all other cases the signal should be ignored since the jobs will be recovered once the Dispatcher becomes the leader. In order to still support non-blocking job recoveries, this commit serializes all recovery operations by introducing a recoveryOperation future which first needs to complete before a subsequent operation is started. That way we can avoid race conditions between granting and revoking leadership as well as the onAddedJobGraph signals. This is important since we can only lock each JobGraph once and, thus, need to make sure that we don't release a lock of a properly recovered job in a concurrent operation. This closes apache#6678.
tillrohrmann
added a commit
to tillrohrmann/flink
that referenced
this pull request
Sep 14, 2018
The Dispatcher should only react to the onAddedJobGraph signal if it is the leader. In all other cases the signal should be ignored since the jobs will be recovered once the Dispatcher becomes the leader. In order to still support non-blocking job recoveries, this commit serializes all recovery operations by introducing a recoveryOperation future which first needs to complete before a subsequent operation is started. That way we can avoid race conditions between granting and revoking leadership as well as the onAddedJobGraph signals. This is important since we can only lock each JobGraph once and, thus, need to make sure that we don't release a lock of a properly recovered job in a concurrent operation. This closes apache#6678.
…xcpetion into Function
The Dispatcher should only react to the onAddedJobGraph signal if it is the leader. In all other cases the signal should be ignored since the jobs will be recovered once the Dispatcher becomes the leader. In order to still support non-blocking job recoveries, this commit serializes all recovery operations by introducing a recoveryOperation future which first needs to complete before a subsequent operation is started. That way we can avoid race conditions between granting and revoking leadership as well as the onAddedJobGraph signals. This is important since we can only lock each JobGraph once and, thus, need to make sure that we don't release a lock of a properly recovered job in a concurrent operation. This closes apache#6678.
…nsumer ThrowingRunnable#unchecked converts a ThrowingRunnable into a Runnable which throws checked exceptions as unchecked ones. FunctionUtils#uncheckedConsmer(ThrowingConsumer) converts a ThrowingConsumer into a Consumer which throws checked exceptions as unchecked ones. This is necessary because ThrowingConsumer is public and we cannot add new methods to the interface.
…point This is necessary to support the command line syntax used by the multi master standalone start-up scripts.
tillrohrmann
force-pushed
the
fixJobRecovery
branch
from
34e3b73
to
740443e
Compare
tillrohrmann
added a commit
to tillrohrmann/flink
that referenced
this pull request
Sep 14, 2018
The Dispatcher should only react to the onAddedJobGraph signal if it is the leader. In all other cases the signal should be ignored since the jobs will be recovered once the Dispatcher becomes the leader. In order to still support non-blocking job recoveries, this commit serializes all recovery operations by introducing a recoveryOperation future which first needs to complete before a subsequent operation is started. That way we can avoid race conditions between granting and revoking leadership as well as the onAddedJobGraph signals. This is important since we can only lock each JobGraph once and, thus, need to make sure that we don't release a lock of a properly recovered job in a concurrent operation. This closes apache#6678.
|
Hi @tillrohrmann , is it possible that two async operation that modifies the same recoveryOperation at the same time? Would that be serializable in that case? |
|
I think it should not be possible to have two async recovery operations ongoing since either of them will have to wait for the other to complete. That was the idea of the fix. |
|
Thanks for the reply, that'll make sense. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
The Dispatcher should only react to the onAddedJobGraph signal if it is the leader.
In all other cases the signal should be ignored since the jobs will be recovered once
the Dispatcher becomes the leader.
In order to still support non-blocking job recoveries, this commit serializes all
recovery operations by introducing a recoveryOperation future which first needs to
complete before a subsequent operation is started. That way we can avoid race conditions
between granting and revoking leadership as well as the onAddedJobGraph signals. This is
important since we can only lock each JobGraph once and, thus, need to make sure that
we don't release a lock of a properly recovered job in a concurrent operation.
cc @GJL
Brief change log
SubmittedJobGraphListener#onAddedJobGraphwhen being the leaderrecoveryOperationfuture in order to avoid wrong unlocking of guarded resourcesVerifying this change
ZooKeeperHADispatcherTest#testStandbyDispatcherJobExecutionandZooKeeperHADispatcherTest#testStandbyDispatcherJobRecoveryDoes this pull request potentially affect one of the following parts:
@Public(Evolving): (no)Documentation