[FLINK-15347] Add SupervisorActor which monitors the proper termination of AkkaRpcActors #11683

tillrohrmann · 2020-04-09T08:38:50Z

What is the purpose of the change

In order to properly complete the termination future of an RpcEndpoint, we need to monitor
when the underlying AkkaRpcActor has been removed from the ActorSystem. If this is not done,
then it can happen that another RpcEndpoint using the same name cannot be started because
the old RpcEndpoint is still registered.

The way we achieve this monitoring is to introduce a helper actor which is responsible for
starting the AkkaRpcActors for all RpcEndpoints. Since the SupervisorActor is the parent
of all RpcEndpoints, it can tell when they are being removed from the ActorSystem through
the SupervisorStrategy.

A consequence of the new actor is that the akka urls change from akka://flink@actorsystem:port/user/xyz
to akka://flink@actorsystem:port/user/rpc/xyz. The respective method AkkaRpcServiceUtils.getRpcUrl
has been updated to reflect this change. This hierarchy change also warrants the bump of the
AkkaRpcService.VERSION.

The failure behaviour of the underlying ActorSystem has been changed so that it terminates
all running actors if an exception is thrown from the SupervisorActor. The assumption is that
such an exception always indicates a programming error and is unrecoverable.

The same applies to failure originating from an AkkaRpcActor (children of the SupervisorActor).
If such an exception is thrown, then we assume that the system is in an illegal state and shut it
down. The way it works is by recording the failure cause for the respective AkkaRpcActor and
then terminating the ActorSystem.

Verifying this change

Added AkkaActorSystemTest, SupervisorActorTest and AkkaRpcActorTest.canReuseEndpointNameAfterTermination.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (yes)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable)

flinkbot · 2020-04-09T08:41:49Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 07cde12 (Thu Apr 09 08:41:49 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-04-09T09:12:03Z

CI report:

fade42d UNKNOWN
80b7f1a Travis: SUCCESS Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

tillrohrmann · 2020-04-09T12:30:02Z

This PR has still a problem with a potential deadlock because the SupervisorActor uses the same Dispatcher as the other actors. I'm working on a fix.

tillrohrmann · 2020-04-09T13:11:00Z

Resolved the problem (hopefully) by introducing a separate supervisor-dispatcher.

tillrohrmann · 2020-04-10T09:52:22Z

@flinkbot run travis

tillrohrmann · 2020-04-10T09:52:30Z

@flinkbot run azure

GJL · 2020-04-15T08:26:22Z

flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/ActorSystemResource.java

+		Preconditions.checkState(actorSystem != null, "ActorSystem must be initialized when calling after.");
+		AkkaUtils.terminateActorSystem(actorSystem).join();
+
+		super.after();


Not needed to call super.before() and super.after(). Also not done in their own code (e.g., TemporaryFolder) and in the examples in Junit's Javadocs.

Good point. I will remove it.

GJL · 2020-04-15T13:16:42Z

.../src/main/java/org/apache/flink/runtime/rpc/akka/exceptions/AkkaUnknownMessageException.java

@@ -22,7 +22,7 @@
 * Exception which indicates that the AkkaRpcActor has received an
 * unknown message type.
 */
-public class AkkaUnknownMessageException extends AkkaRpcException {
+public class AkkaUnknownMessageException extends AkkaRpcRuntimeException {


I think the commit message should better reflect why this change was done.

I will add a more detailed commit message

GJL · 2020-04-16T16:45:20Z

flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/SupervisorActorTest.java

+			terminationFuture.get();
+			fail("Expected the termination future being completed exceptionally");
+		} catch (ExecutionException expected) {
+			final Throwable ignored = ExceptionUtils.findThrowable(expected, e -> e.equals(cause))


Why assign the throwable to ignored in the first place? Also in completesTerminationFutureExceptionallyIfActorStopsExceptionally()

There is no reason other than stupidity.

GJL · 2020-04-16T16:46:55Z

flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/SupervisorActorTest.java

+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+public class SupervisorActorTest extends TestLogger {


We normally add a comment.

True, I will add one. Not sure whether this all that meaningful, though.

GJL · 2020-04-16T16:54:25Z

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/SupervisorActor.java

+			if (errorCause != null) {
+				if (!internalTerminationFuture.completeExceptionally(errorCause)) {
+					// we have another failure reason -> let's add it
+					terminationFuture = internalTerminationFuture.<Void>handle(


I think <Void> can be inferred by the compiler:

Suggested change

terminationFuture = internalTerminationFuture.<Void>handle(

terminationFuture = internalTerminationFuture.handle(

I will remove it.

GJL · 2020-04-16T17:33:37Z

flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/SupervisorActorTest.java

+			terminationFuture -> Props.create(SimpleActor.class, terminationFuture),
+			"foobar");
+
+		final SupervisorActor.ActorRegistration actorRegistration = startResponse.orElseThrow(cause -> new AssertionError("Expected the start to succeed.", cause));


All the implementations for the throwableFunction are the same in this test. Maybe there is a way to de-deduplicate this line:

final SupervisorActor.ActorRegistration actorRegistration = startResponse.orElseThrow(cause -> new AssertionError("Expected the start to succeed.", cause));

Good point. I'll look into it.

GJL · 2020-04-16T17:34:31Z

Looks good all in all.

tillrohrmann · 2020-04-17T12:22:47Z

Thanks for the review @GJL. I've addressed your comments (which unfortunately required a force push).

GJL

LGTM.

tillrohrmann · 2020-04-20T09:55:55Z

Thanks for the review @GJL. I'll rebase the PR and once Travis gives green light, then I will merge it.

AkkaUnknownMessageException represents a programming error which is not recoverable. Consequently, this exception becomes an unchecked exception with this commit.

…on of AkkaRpcActors In order to properly complete the termination future of an RpcEndpoint, we need to monitor when the underlying AkkaRpcActor has been removed from the ActorSystem. If this is not done, then it can happen that another RpcEndpoint using the same name cannot be started because the old RpcEndpoint is still registered. The way we achieve this monitoring is to introduce a helper actor which is responsible for starting the AkkaRpcActors for all RpcEndpoints. Since the SupervisorActor is the parent of all RpcEndpoints, it can tell when they are being removed from the ActorSystem through the SupervisorStrategy. A consequence of the new actor is that the akka urls change from akka://flink@actorsystem:port/user/xyz to akka://flink@actorsystem:port/user/rpc/xyz. The respective method AkkaRpcServiceUtils.getRpcUrl has been updated to reflect this change. This hierarchy change also warrants the bump of the AkkaRpcService.VERSION. The failure behaviour of the underlying ActorSystem has been changed so that it terminates all running actors if an exception is thrown from the SupervisorActor. The assumption is that such an exception always indicates a programming error and is unrecoverable. The same applies to failure originating from an AkkaRpcActor (children of the SupervisorActor). If such an exception is thrown, then we assume that the system is in an illegal state and shut it down. The way it works is by recording the failure cause for the respective AkkaRpcActor and then terminating the ActorSystem.

The separate dispatcher ensures that the SupervisorActor can always make progress. This again ensures that the other RpcEndpoints won't deadlock when starting a new RpcEndpoint. The reason for this is because a newly started RpcEndpoint needs to register at the SupervisorActor.

FutureUtils.forwardAsync forwards the source value to the target future using the provided executor.

…Actor The SupervisorActor must never block its own main thread. Due to this constraint, we must not complete the external termination futures from the main thread, because one does not know what the user does with this future. Hence, this commit introduces a dedicated termination future executor which is responsible for completing the termination futures. This closes apache#11683.

rmetzger added the review=description? label Apr 9, 2020

rmetzger added the component=Runtime/Coordination label Apr 9, 2020

tillrohrmann mentioned this pull request Apr 9, 2020

[FLINK-17072] Let Dispatcher and ResourceManager use random endpoint id #11693

Closed

tillrohrmann force-pushed the FLINK-15347 branch from fade42d to a4765c1 Compare April 11, 2020 11:54

GJL self-assigned this Apr 14, 2020

GJL reviewed Apr 16, 2020

View reviewed changes

tillrohrmann force-pushed the FLINK-15347 branch from 2aba0b8 to 7be78ae Compare April 17, 2020 12:22

GJL approved these changes Apr 20, 2020

View reviewed changes

tillrohrmann added 11 commits April 20, 2020 11:56

[hotfix] Add RpcUtils#terminateRpcEndpoints

a3c268b

[hotfix] Add RpcUtils#INF_DURATION

a62f9c5

[hotfix] Add RpcRuntimeException

4443559

[hotfix] Add AkkRpcRuntimeException

25fa5d8

[hotfix] Let AkkaUnknownMessageException extend AkkaRpcRuntimeException

dab8a40

AkkaUnknownMessageException represents a programming error which is not recoverable. Consequently, this exception becomes an unchecked exception with this commit.

[FLINK-15347] Bump AkkRpcVersion to 2

66c0da7

[FLINK-15347][tests] Add ActorSystemResource

1e6d5d1

[hotfix] Add FutureUtils.forwardAsync

38ffb2c

FutureUtils.forwardAsync forwards the source value to the target future using the provided executor.

tillrohrmann force-pushed the FLINK-15347 branch from 7be78ae to 80b7f1a Compare April 20, 2020 09:57

tillrohrmann closed this in 422aa89 Apr 20, 2020

tillrohrmann deleted the FLINK-15347 branch December 15, 2020 10:31

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-15347] Add SupervisorActor which monitors the proper termination of AkkaRpcActors #11683

[FLINK-15347] Add SupervisorActor which monitors the proper termination of AkkaRpcActors #11683

tillrohrmann commented Apr 9, 2020

flinkbot commented Apr 9, 2020

flinkbot commented Apr 9, 2020 •

edited

tillrohrmann commented Apr 9, 2020

tillrohrmann commented Apr 9, 2020

tillrohrmann commented Apr 10, 2020

tillrohrmann commented Apr 10, 2020

GJL Apr 15, 2020

tillrohrmann Apr 17, 2020

GJL Apr 15, 2020

tillrohrmann Apr 17, 2020

GJL Apr 16, 2020

tillrohrmann Apr 17, 2020

GJL Apr 16, 2020

tillrohrmann Apr 17, 2020

GJL Apr 16, 2020

tillrohrmann Apr 17, 2020

GJL Apr 16, 2020 •

edited

tillrohrmann Apr 17, 2020

GJL commented Apr 16, 2020

tillrohrmann commented Apr 17, 2020

GJL left a comment

tillrohrmann commented Apr 20, 2020

	terminationFuture = internalTerminationFuture.<Void>handle(
	terminationFuture = internalTerminationFuture.handle(

[FLINK-15347] Add SupervisorActor which monitors the proper termination of AkkaRpcActors #11683

[FLINK-15347] Add SupervisorActor which monitors the proper termination of AkkaRpcActors #11683

Conversation

tillrohrmann commented Apr 9, 2020

What is the purpose of the change

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Apr 9, 2020

Automated Checks

Review Progress

flinkbot commented Apr 9, 2020 • edited

CI report:

tillrohrmann commented Apr 9, 2020

tillrohrmann commented Apr 9, 2020

tillrohrmann commented Apr 10, 2020

tillrohrmann commented Apr 10, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GJL Apr 16, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GJL commented Apr 16, 2020

tillrohrmann commented Apr 17, 2020

GJL left a comment

Choose a reason for hiding this comment

tillrohrmann commented Apr 20, 2020

flinkbot commented Apr 9, 2020 •

edited

GJL Apr 16, 2020 •

edited