You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2019/02/12 20:57:57 UTC
[flink] 03/05: [FLINK-11551][rpc] Allow RpcEndpoint to execute
asynchronous stop operations
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3d200070c4404dbce61d80fcdda4509ece383937
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Feb 7 11:01:42 2019 +0100
[FLINK-11551][rpc] Allow RpcEndpoint to execute asynchronous stop operations
Replace RpcEndpoin#postStop with RpcEndpoint#onStop method which can execute asynchronous
operations in the main thread executor. onStop returns a future which will terminate the
RpcEndpoint once it is completed.
The new stop semantics allow to execute complex shut down logic which requires asynchronous
operations.
This closes #7665.
---
.../clusterframework/MesosResourceManager.java | 4 +-
.../flink/runtime/dispatcher/Dispatcher.java | 2 +-
.../apache/flink/runtime/jobmaster/JobMaster.java | 2 +-
.../flink/runtime/jobmaster/slotpool/SlotPool.java | 2 +-
.../runtime/resourcemanager/ResourceManager.java | 2 +-
.../org/apache/flink/runtime/rpc/RpcEndpoint.java | 9 +-
.../runtime/rpc/akka/AkkaInvocationHandler.java | 5 +-
.../flink/runtime/rpc/akka/AkkaRpcActor.java | 190 +++++++++++++++++----
.../flink/runtime/rpc/akka/AkkaRpcService.java | 44 +++--
.../Processing.java => ControlMessages.java} | 11 +-
.../exceptions/AkkaRpcInvalidStateException.java} | 24 +--
.../flink/runtime/taskexecutor/TaskExecutor.java | 2 +-
.../apache/flink/runtime/rpc/AsyncCallsTest.java | 10 --
.../flink/runtime/rpc/FencedRpcEndpointTest.java | 5 -
.../apache/flink/runtime/rpc/RpcEndpointTest.java | 5 -
.../apache/flink/runtime/rpc/RpcSSLAuthITCase.java | 5 -
.../AkkaRpcActorOversizedResponseMessageTest.java | 4 +-
.../flink/runtime/rpc/akka/AkkaRpcActorTest.java | 95 +++++++----
.../flink/runtime/rpc/akka/AkkaRpcServiceTest.java | 168 ++++++++++++++++--
.../runtime/rpc/akka/MainThreadValidationTest.java | 5 -
.../runtime/rpc/akka/MessageSerializationTest.java | 27 +--
.../retriever/impl/RpcGatewayRetrieverTest.java | 5 -
.../org/apache/flink/yarn/YarnResourceManager.java | 4 +-
23 files changed, 449 insertions(+), 181 deletions(-)
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
index ae51a20..04cf7cc 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
@@ -400,8 +400,8 @@ public class MesosResourceManager extends ResourceManager<RegisteredMesosWorkerN
}
@Override
- public CompletableFuture<Void> postStop() {
- return stopSupportingActorsAsync().thenCompose((ignored) -> super.postStop());
+ public CompletableFuture<Void> onStop() {
+ return stopSupportingActorsAsync().thenCompose((ignored) -> super.onStop());
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
index 07223c3..39a496d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -183,7 +183,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
//------------------------------------------------------
@Override
- public CompletableFuture<Void> postStop() {
+ public CompletableFuture<Void> onStop() {
log.info("Stopping dispatcher {}.", getAddress());
final CompletableFuture<Void> allJobManagerRunnersTerminationFuture = terminateJobManagerRunnersAndGetTerminationFuture();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index 29f8914..e2030a6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -347,7 +347,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
* Suspend the job and shutdown all other services including rpc.
*/
@Override
- public CompletableFuture<Void> postStop() {
+ public CompletableFuture<Void> onStop() {
log.info("Stopping the JobMaster for job {}({}).", jobGraph.getName(), jobGraph.getJobID());
// disconnect from all registered TaskExecutors
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
index 9ac7556..f69c1a4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
@@ -217,7 +217,7 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
}
@Override
- public CompletableFuture<Void> postStop() {
+ public CompletableFuture<Void> onStop() {
log.info("Stopping SlotPool.");
// cancel all pending allocations
Set<AllocationID> allocationIds = pendingRequests.keySetB();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
index 8632afe..4ef1858 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -226,7 +226,7 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
}
@Override
- public CompletableFuture<Void> postStop() {
+ public CompletableFuture<Void> onStop() {
Exception exception = null;
taskManagerHeartbeatManager.stop();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java
index 89526e6..764f969 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java
@@ -147,7 +147,9 @@ public abstract class RpcEndpoint implements RpcGateway {
* @return Future which is completed once all post stop actions are completed. If an error
* occurs this future is completed exceptionally
*/
- public abstract CompletableFuture<Void> postStop();
+ public CompletableFuture<Void> onStop() {
+ return CompletableFuture.completedFuture(null);
+ }
/**
* Triggers the shut down of the rpc endpoint. The shut down is executed asynchronously.
@@ -159,6 +161,11 @@ public abstract class RpcEndpoint implements RpcGateway {
rpcService.stopServer(rpcServer);
}
+ public final CompletableFuture<Void> terminate() {
+ rpcService.stopServer(rpcServer);
+ return getTerminationFuture();
+ }
+
// ------------------------------------------------------------------------
// Basic RPC endpoint properties
// ------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java
index e147cf6..15c5f88 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java
@@ -26,7 +26,6 @@ import org.apache.flink.runtime.rpc.RpcGateway;
import org.apache.flink.runtime.rpc.RpcServer;
import org.apache.flink.runtime.rpc.RpcTimeout;
import org.apache.flink.runtime.rpc.StartStoppable;
-import org.apache.flink.runtime.rpc.akka.messages.Processing;
import org.apache.flink.runtime.rpc.exceptions.RpcException;
import org.apache.flink.runtime.rpc.messages.CallAsync;
import org.apache.flink.runtime.rpc.messages.LocalRpcInvocation;
@@ -172,12 +171,12 @@ class AkkaInvocationHandler implements InvocationHandler, AkkaBasedEndpoint, Rpc
@Override
public void start() {
- rpcEndpoint.tell(Processing.START, ActorRef.noSender());
+ rpcEndpoint.tell(ControlMessages.START, ActorRef.noSender());
}
@Override
public void stop() {
- rpcEndpoint.tell(Processing.STOP, ActorRef.noSender());
+ rpcEndpoint.tell(ControlMessages.STOP, ActorRef.noSender());
}
// ------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
index b17222e..f79c550 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
@@ -24,8 +24,8 @@ import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcGateway;
import org.apache.flink.runtime.rpc.akka.exceptions.AkkaHandshakeException;
import org.apache.flink.runtime.rpc.akka.exceptions.AkkaRpcException;
+import org.apache.flink.runtime.rpc.akka.exceptions.AkkaRpcInvalidStateException;
import org.apache.flink.runtime.rpc.akka.exceptions.AkkaUnknownMessageException;
-import org.apache.flink.runtime.rpc.akka.messages.Processing;
import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException;
import org.apache.flink.runtime.rpc.messages.CallAsync;
import org.apache.flink.runtime.rpc.messages.HandshakeSuccessMessage;
@@ -38,12 +38,16 @@ import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.SerializedValue;
import akka.actor.ActorRef;
+import akka.actor.Kill;
import akka.actor.Status;
import akka.actor.UntypedActor;
import akka.pattern.Patterns;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
@@ -59,7 +63,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* Akka rpc actor which receives {@link LocalRpcInvocation}, {@link RunAsync} and {@link CallAsync}
- * {@link Processing} messages.
+ * {@link ControlMessages} messages.
*
* <p>The {@link LocalRpcInvocation} designates a rpc and is dispatched to the given {@link RpcEndpoint}
* instance.
@@ -67,8 +71,8 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* <p>The {@link RunAsync} and {@link CallAsync} messages contain executable code which is executed
* in the context of the actor thread.
*
- * <p>The {@link Processing} message controls the processing behaviour of the akka rpc actor. A
- * {@link Processing#START} starts processing incoming messages. A {@link Processing#STOP} message
+ * <p>The {@link ControlMessages} message controls the processing behaviour of the akka rpc actor. A
+ * {@link ControlMessages#START} starts processing incoming messages. A {@link ControlMessages#STOP} message
* stops processing messages. All messages which arrive when the processing is stopped, will be
* discarded.
*
@@ -90,8 +94,12 @@ class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends UntypedActor {
private final long maximumFramesize;
+ @Nonnull
private State state;
+ @Nullable
+ private CompletableFuture<Void> rpcEndpointTerminationFuture;
+
AkkaRpcActor(
final T rpcEndpoint,
final CompletableFuture<Boolean> terminationFuture,
@@ -104,29 +112,16 @@ class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends UntypedActor {
this.terminationFuture = checkNotNull(terminationFuture);
this.version = version;
this.maximumFramesize = maximumFramesize;
- this.state = State.STOPPED;
+ this.state = StoppedState.INSTANCE;
+ this.rpcEndpointTerminationFuture = null;
}
@Override
public void postStop() throws Exception {
- mainThreadValidator.enterMainThread();
+ super.postStop();
- try {
- CompletableFuture<Void> postStopFuture;
- try {
- postStopFuture = rpcEndpoint.postStop();
- } catch (Throwable throwable) {
- postStopFuture = FutureUtils.completedExceptionally(throwable);
- }
-
- super.postStop();
-
- // IMPORTANT: This only works if we don't use a restarting supervisor strategy. Otherwise
- // we would complete the future and let the actor system restart the actor with a completed
- // future.
- // Complete the termination future so that others know that we've stopped.
-
- postStopFuture.whenComplete(
+ if (rpcEndpointTerminationFuture != null && rpcEndpointTerminationFuture.isDone()) {
+ rpcEndpointTerminationFuture.whenComplete(
(Void value, Throwable throwable) -> {
if (throwable != null) {
terminationFuture.completeExceptionally(throwable);
@@ -134,20 +129,22 @@ class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends UntypedActor {
terminationFuture.complete(null);
}
});
- } finally {
- mainThreadValidator.exitMainThread();
+ } else {
+ terminationFuture.completeExceptionally(
+ new AkkaRpcException(
+ String.format("RpcEndpoint %s has not been properly stopped.", rpcEndpoint.getEndpointId())));
}
+
+ state = state.finishTermination();
}
@Override
public void onReceive(final Object message) {
if (message instanceof RemoteHandshakeMessage) {
handleHandshakeMessage((RemoteHandshakeMessage) message);
- } else if (message.equals(Processing.START)) {
- state = State.STARTED;
- } else if (message.equals(Processing.STOP)) {
- state = State.STOPPED;
- } else if (state == State.STARTED) {
+ } else if (message instanceof ControlMessages) {
+ handleControlMessage(((ControlMessages) message));
+ } else if (state.isRunning()) {
mainThreadValidator.enterMainThread();
try {
@@ -165,6 +162,28 @@ class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends UntypedActor {
}
}
+ private void handleControlMessage(ControlMessages controlMessage) {
+ switch (controlMessage) {
+ case START:
+ state = state.start();
+ break;
+ case STOP:
+ state = state.stop();
+ break;
+ case TERMINATE:
+ state.terminate(this);
+ break;
+ default:
+ handleUnknownControlMessage(controlMessage);
+ }
+ }
+
+ private void handleUnknownControlMessage(ControlMessages controlMessage) {
+ final String message = String.format("Received unknown control message %s. Dropping this message!", controlMessage);
+ log.warn(message);
+ sendErrorIfSender(new AkkaUnknownMessageException(message));
+ }
+
protected void handleRpcMessage(Object message) {
if (message instanceof RunAsync) {
handleRunAsync((RunAsync) message);
@@ -438,8 +457,115 @@ class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends UntypedActor {
return message;
}
- enum State {
- STARTED,
- STOPPED
+ // ---------------------------------------------------------------------------
+ // Internal state machine
+ // ---------------------------------------------------------------------------
+
+ interface State {
+ default State start() {
+ throw new AkkaRpcInvalidStateException(invalidStateTransitionMessage(StartedState.INSTANCE));
+ }
+
+ default State stop() {
+ throw new AkkaRpcInvalidStateException(invalidStateTransitionMessage(StoppedState.INSTANCE));
+ }
+
+ default State terminate(AkkaRpcActor<?> akkaRpcActor) {
+ throw new AkkaRpcInvalidStateException(invalidStateTransitionMessage(TerminatingState.INSTANCE));
+ }
+
+ default State finishTermination() {
+ return TerminatedState.INSTANCE;
+ }
+
+ default boolean isRunning() {
+ return false;
+ }
+
+ default String invalidStateTransitionMessage(State targetState) {
+ return String.format("AkkaRpcActor is currently in state %s and cannot go into state %s.", this, targetState);
+ }
+ }
+
+ @SuppressWarnings("Singleton")
+ enum StartedState implements State {
+ INSTANCE;
+
+ @Override
+ public State start() {
+ return INSTANCE;
+ }
+
+ @Override
+ public State stop() {
+ return StoppedState.INSTANCE;
+ }
+
+ @Override
+ public State terminate(AkkaRpcActor<?> akkaRpcActor) {
+ akkaRpcActor.mainThreadValidator.enterMainThread();
+
+ try {
+ akkaRpcActor.rpcEndpointTerminationFuture = akkaRpcActor.rpcEndpoint.onStop();
+ } catch (Throwable t) {
+ akkaRpcActor.rpcEndpointTerminationFuture = FutureUtils.completedExceptionally(
+ new AkkaRpcException(
+ String.format("Failure while stopping RpcEndpoint %s.", akkaRpcActor.rpcEndpoint.getEndpointId()),
+ t));
+ } finally {
+ akkaRpcActor.mainThreadValidator.exitMainThread();
+ }
+
+ // IMPORTANT: This only works if we don't use a restarting supervisor strategy. Otherwise
+ // we would complete the future and let the actor system restart the actor with a completed
+ // future.
+ // Complete the termination future so that others know that we've stopped.
+
+ akkaRpcActor.rpcEndpointTerminationFuture.whenComplete((ignored, throwable) -> akkaRpcActor.getSelf().tell(Kill.getInstance(), ActorRef.noSender()));
+
+ return TerminatingState.INSTANCE;
+ }
+
+ @Override
+ public boolean isRunning() {
+ return true;
+ }
+ }
+
+ @SuppressWarnings("Singleton")
+ enum StoppedState implements State {
+ INSTANCE;
+
+ @Override
+ public State start() {
+ return StartedState.INSTANCE;
+ }
+
+ @Override
+ public State stop() {
+ return INSTANCE;
+ }
+
+ @Override
+ public State terminate(AkkaRpcActor<?> akkaRpcActor) {
+ akkaRpcActor.rpcEndpointTerminationFuture = CompletableFuture.completedFuture(null);
+ akkaRpcActor.getSelf().tell(Kill.getInstance(), ActorRef.noSender());
+
+ return TerminatingState.INSTANCE;
+ }
+ }
+
+ @SuppressWarnings("Singleton")
+ enum TerminatingState implements State {
+ INSTANCE;
+
+ @Override
+ public boolean isRunning() {
+ return true;
+ }
+ }
+
+ enum TerminatedState implements State {
+ INSTANCE
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
index 6548513..344f96b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
@@ -41,20 +41,21 @@ import akka.actor.ActorSelection;
import akka.actor.ActorSystem;
import akka.actor.Address;
import akka.actor.Identify;
-import akka.actor.PoisonPill;
import akka.actor.Props;
-import akka.actor.Terminated;
import akka.dispatch.Futures;
import akka.pattern.Patterns;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nonnull;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import java.io.Serializable;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@@ -315,7 +316,7 @@ public class AkkaRpcService implements RpcService {
}
if (rpcEndpoint != null) {
- akkaClient.getActorRef().tell(PoisonPill.getInstance(), ActorRef.noSender());
+ terminateAkkaRpcActor(akkaClient.getActorRef(), rpcEndpoint);
} else {
LOG.debug("RPC endpoint {} already stopped or from different RPC service", selfGateway.getAddress());
}
@@ -324,24 +325,26 @@ public class AkkaRpcService implements RpcService {
@Override
public CompletableFuture<Void> stopService() {
+ final CompletableFuture<Void> akkaRpcActorsTerminationFuture;
+
synchronized (lock) {
if (stopped) {
return terminationFuture;
}
+ LOG.info("Stopping Akka RPC service.");
+
stopped = true;
- }
- LOG.info("Stopping Akka RPC service.");
+ akkaRpcActorsTerminationFuture = terminateAkkaRpcActors();
+ }
- final CompletableFuture<Terminated> actorSystemTerminationFuture = FutureUtils.toJava(actorSystem.terminate());
+ final CompletableFuture<Void> actorSystemTerminationFuture = FutureUtils.composeAfterwards(
+ akkaRpcActorsTerminationFuture,
+ () -> FutureUtils.toJava(actorSystem.terminate()));
actorSystemTerminationFuture.whenComplete(
- (Terminated ignored, Throwable throwable) -> {
- synchronized (lock) {
- actors.clear();
- }
-
+ (Void ignored, Throwable throwable) -> {
if (throwable != null) {
terminationFuture.completeExceptionally(throwable);
} else {
@@ -354,6 +357,25 @@ public class AkkaRpcService implements RpcService {
return terminationFuture;
}
+ @GuardedBy("lock")
+ @Nonnull
+ private CompletableFuture<Void> terminateAkkaRpcActors() {
+ final Collection<CompletableFuture<Void>> akkaRpcActorTerminationFutures = new ArrayList<>(actors.size());
+
+ for (Map.Entry<ActorRef, RpcEndpoint> actorRefRpcEndpointEntry : actors.entrySet()) {
+ akkaRpcActorTerminationFutures.add(terminateAkkaRpcActor(actorRefRpcEndpointEntry.getKey(), actorRefRpcEndpointEntry.getValue()));
+ }
+ actors.clear();
+
+ return FutureUtils.waitForAll(akkaRpcActorTerminationFutures);
+ }
+
+ private CompletableFuture<Void> terminateAkkaRpcActor(ActorRef akkaRpcActorRef, RpcEndpoint rpcEndpoint) {
+ akkaRpcActorRef.tell(ControlMessages.TERMINATE, ActorRef.noSender());
+
+ return rpcEndpoint.getTerminationFuture();
+ }
+
@Override
public CompletableFuture<Void> getTerminationFuture() {
return terminationFuture;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/messages/Processing.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/ControlMessages.java
similarity index 71%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/messages/Processing.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/ControlMessages.java
index 030ff60..2cfd62d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/messages/Processing.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/ControlMessages.java
@@ -16,12 +16,13 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.rpc.akka.messages;
+package org.apache.flink.runtime.rpc.akka;
/**
- * Controls the processing behaviour of the {@link org.apache.flink.runtime.rpc.akka.AkkaRpcActor}
+ * Control messages for the {@link AkkaRpcActor}.
*/
-public enum Processing {
- START, // Unstashes all stashed messages and starts processing incoming messages
- STOP // Stop processing messages and stashes all incoming messages
+public enum ControlMessages {
+ START, // Start processing incoming messages
+ STOP, // Stop processing messages and drop all newly incoming messages
+ TERMINATE, // Terminate the AkkaRpcActor
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/TestRpcEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/exceptions/AkkaRpcInvalidStateException.java
similarity index 59%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/TestRpcEndpoint.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/exceptions/AkkaRpcInvalidStateException.java
index 073dabe..83e7d12 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/TestRpcEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/exceptions/AkkaRpcInvalidStateException.java
@@ -16,21 +16,25 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.rpc.akka;
+package org.apache.flink.runtime.rpc.akka.exceptions;
-import org.apache.flink.runtime.rpc.RpcEndpoint;
-import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.FlinkRuntimeException;
-import java.util.concurrent.CompletableFuture;
+/**
+ * Exception which indicates an invalid state.
+ */
+public class AkkaRpcInvalidStateException extends FlinkRuntimeException {
+ private static final long serialVersionUID = 7771345513700817242L;
-class TestRpcEndpoint extends RpcEndpoint {
+ public AkkaRpcInvalidStateException(String message) {
+ super(message);
+ }
- protected TestRpcEndpoint(RpcService rpcService) {
- super(rpcService);
+ public AkkaRpcInvalidStateException(Throwable cause) {
+ super(cause);
}
- @Override
- public CompletableFuture<Void> postStop() {
- return CompletableFuture.completedFuture(null);
+ public AkkaRpcInvalidStateException(String message, Throwable cause) {
+ super(message, cause);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index 79176c8..ed90b0e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -294,7 +294,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
* Called to shut down the TaskManager. The method closes all TaskManager services.
*/
@Override
- public CompletableFuture<Void> postStop() {
+ public CompletableFuture<Void> onStop() {
log.info("Stopping TaskExecutor {}.", getAddress());
Throwable throwable = null;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java
index 1233306..3927b0c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/AsyncCallsTest.java
@@ -367,11 +367,6 @@ public class AsyncCallsTest extends TestLogger {
concurrentAccess.set(true);
}
}
-
- @Override
- public CompletableFuture<Void> postStop() {
- return CompletableFuture.completedFuture(null);
- }
}
public interface FencedTestGateway extends FencedRpcGateway<UUID>, TestGateway {
@@ -453,11 +448,6 @@ public class AsyncCallsTest extends TestLogger {
}
@Override
- public CompletableFuture<Void> postStop() {
- return CompletableFuture.completedFuture(null);
- }
-
- @Override
public void someCall() {
boolean holdsLock = lock.tryLock();
if (holdsLock) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java
index 81407e5..344a05a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/FencedRpcEndpointTest.java
@@ -300,11 +300,6 @@ public class FencedRpcEndpointTest extends TestLogger {
this(rpcService, value, null);
}
- @Override
- public CompletableFuture<Void> postStop() {
- return CompletableFuture.completedFuture(null);
- }
-
protected FencedTestingEndpoint(RpcService rpcService, String value, UUID initialFencingToken) {
super(rpcService);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java
index 0081d58..ba0f9ca 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java
@@ -161,11 +161,6 @@ public class RpcEndpointTest extends TestLogger {
public CompletableFuture<Integer> foobar() {
return CompletableFuture.completedFuture(foobarValue);
}
-
- @Override
- public CompletableFuture<Void> postStop() {
- return CompletableFuture.completedFuture(null);
- }
}
public static class ExtendedEndpoint extends BaseEndpoint implements ExtendedGateway, DifferentGateway {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcSSLAuthITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcSSLAuthITCase.java
index 138cf98..141db35 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcSSLAuthITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcSSLAuthITCase.java
@@ -145,11 +145,6 @@ public class RpcSSLAuthITCase extends TestLogger {
}
@Override
- public CompletableFuture<Void> postStop() {
- return CompletableFuture.completedFuture(null);
- }
-
- @Override
public CompletableFuture<String> foo() {
return CompletableFuture.completedFuture("bar");
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorOversizedResponseMessageTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorOversizedResponseMessageTest.java
index 73de0a1..119f37e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorOversizedResponseMessageTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorOversizedResponseMessageTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.rpc.akka;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcGateway;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.RpcUtils;
@@ -170,7 +171,7 @@ public class AkkaRpcActorOversizedResponseMessageTest extends TestLogger {
String messageSync() throws RpcException;
}
- static class MessageRpcEndpoint extends TestRpcEndpoint implements MessageRpcGateway {
+ static class MessageRpcEndpoint extends RpcEndpoint implements MessageRpcGateway {
@Nonnull
private final String message;
@@ -189,6 +190,5 @@ public class AkkaRpcActorOversizedResponseMessageTest extends TestLogger {
public String messageSync() throws RpcException {
return message;
}
-
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java
index 27eca97..1d4a1c6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java
@@ -162,7 +162,7 @@ public class AkkaRpcActorTest extends TestLogger {
assertFalse(terminationFuture.isDone());
CompletableFuture.runAsync(
- () -> rpcEndpoint.shutDown(),
+ rpcEndpoint::shutDown,
akkaRpcService.getExecutor());
// wait until the rpc endpoint has terminated
@@ -208,12 +208,12 @@ public class AkkaRpcActorTest extends TestLogger {
}
/**
- * Tests that exception thrown in the postStop method are returned by the termination
+ * Tests that exception thrown in the onStop method are returned by the termination
* future.
*/
@Test
- public void testPostStopExceptionPropagation() throws Exception {
- FailingPostStopEndpoint rpcEndpoint = new FailingPostStopEndpoint(akkaRpcService, "FailingPostStopEndpoint");
+ public void testOnStopExceptionPropagation() throws Exception {
+ FailingOnStopEndpoint rpcEndpoint = new FailingOnStopEndpoint(akkaRpcService, "FailingOnStopEndpoint");
rpcEndpoint.start();
rpcEndpoint.shutDown();
@@ -223,15 +223,15 @@ public class AkkaRpcActorTest extends TestLogger {
try {
terminationFuture.get();
} catch (ExecutionException e) {
- assertTrue(e.getCause() instanceof FailingPostStopEndpoint.PostStopException);
+ assertTrue(e.getCause() instanceof FailingOnStopEndpoint.OnStopException);
}
}
/**
- * Checks that the postStop callback is executed within the main thread.
+ * Checks that the onStop callback is executed within the main thread.
*/
@Test
- public void testPostStopExecutedByMainThread() throws Exception {
+ public void testOnStopExecutedByMainThread() throws Exception {
SimpleRpcEndpoint simpleRpcEndpoint = new SimpleRpcEndpoint(akkaRpcService, "SimpleRpcEndpoint");
simpleRpcEndpoint.start();
@@ -239,7 +239,7 @@ public class AkkaRpcActorTest extends TestLogger {
CompletableFuture<Void> terminationFuture = simpleRpcEndpoint.getTerminationFuture();
- // check that we executed the postStop method in the main thread, otherwise an exception
+ // check that we executed the onStop method in the main thread, otherwise an exception
// would be thrown here.
terminationFuture.get();
}
@@ -274,9 +274,9 @@ public class AkkaRpcActorTest extends TestLogger {
* post stop action has completed.
*/
@Test
- public void testActorTerminationWithAsynchronousPostStopAction() throws Exception {
- final CompletableFuture<Void> postStopFuture = new CompletableFuture<>();
- final AsynchronousPostStopEndpoint endpoint = new AsynchronousPostStopEndpoint(akkaRpcService, postStopFuture);
+ public void testActorTerminationWithAsynchronousOnStopAction() throws Exception {
+ final CompletableFuture<Void> onStopFuture = new CompletableFuture<>();
+ final AsynchronousOnStopEndpoint endpoint = new AsynchronousOnStopEndpoint(akkaRpcService, onStopFuture);
try {
endpoint.start();
@@ -287,9 +287,28 @@ public class AkkaRpcActorTest extends TestLogger {
assertFalse(terminationFuture.isDone());
- postStopFuture.complete(null);
+ onStopFuture.complete(null);
+
+ // the onStopFuture completion should allow the endpoint to terminate
+ terminationFuture.get();
+ } finally {
+ RpcUtils.terminateRpcEndpoint(endpoint, timeout);
+ }
+ }
+
+ /**
+ * Tests that we can still run commands via the main thread executor when the onStop method
+ * is called.
+ */
+ @Test
+ public void testMainThreadExecutionOnStop() throws Exception {
+ final MainThreadExecutorOnStopEndpoint endpoint = new MainThreadExecutorOnStopEndpoint(akkaRpcService);
+
+ try {
+ endpoint.start();
+
+ CompletableFuture<Void> terminationFuture = endpoint.terminate();
- // the postStopFuture completion should allow the endpoint to terminate
terminationFuture.get();
} finally {
RpcUtils.terminateRpcEndpoint(endpoint, timeout);
@@ -304,7 +323,7 @@ public class AkkaRpcActorTest extends TestLogger {
CompletableFuture<Integer> foobar();
}
- static class DummyRpcEndpoint extends TestRpcEndpoint implements DummyRpcGateway {
+ static class DummyRpcEndpoint extends RpcEndpoint implements DummyRpcGateway {
private volatile int foobar = 42;
@@ -328,7 +347,7 @@ public class AkkaRpcActorTest extends TestLogger {
CompletableFuture<Integer> doStuff();
}
- private static class ExceptionalEndpoint extends TestRpcEndpoint implements ExceptionalGateway {
+ private static class ExceptionalEndpoint extends RpcEndpoint implements ExceptionalGateway {
protected ExceptionalEndpoint(RpcService rpcService) {
super(rpcService);
@@ -340,7 +359,7 @@ public class AkkaRpcActorTest extends TestLogger {
}
}
- private static class ExceptionalFutureEndpoint extends TestRpcEndpoint implements ExceptionalGateway {
+ private static class ExceptionalFutureEndpoint extends RpcEndpoint implements ExceptionalGateway {
protected ExceptionalFutureEndpoint(RpcService rpcService) {
super(rpcService);
@@ -372,32 +391,26 @@ public class AkkaRpcActorTest extends TestLogger {
protected SimpleRpcEndpoint(RpcService rpcService, String endpointId) {
super(rpcService, endpointId);
}
-
- @Override
- public CompletableFuture<Void> postStop() {
- validateRunsInMainThread();
- return CompletableFuture.completedFuture(null);
- }
}
// ------------------------------------------------------------------------
- private static class FailingPostStopEndpoint extends RpcEndpoint implements RpcGateway {
+ private static class FailingOnStopEndpoint extends RpcEndpoint implements RpcGateway {
- protected FailingPostStopEndpoint(RpcService rpcService, String endpointId) {
+ protected FailingOnStopEndpoint(RpcService rpcService, String endpointId) {
super(rpcService, endpointId);
}
@Override
- public CompletableFuture<Void> postStop() {
- return FutureUtils.completedExceptionally(new PostStopException("Test exception."));
+ public CompletableFuture<Void> onStop() {
+ return FutureUtils.completedExceptionally(new OnStopException("Test exception."));
}
- private static class PostStopException extends FlinkException {
+ private static class OnStopException extends FlinkException {
private static final long serialVersionUID = 6701096588415871592L;
- public PostStopException(String message) {
+ public OnStopException(String message) {
super(message);
}
}
@@ -405,19 +418,33 @@ public class AkkaRpcActorTest extends TestLogger {
// ------------------------------------------------------------------------
- private static class AsynchronousPostStopEndpoint extends RpcEndpoint {
+ static class AsynchronousOnStopEndpoint extends RpcEndpoint {
- private final CompletableFuture<Void> postStopFuture;
+ private final CompletableFuture<Void> onStopFuture;
- protected AsynchronousPostStopEndpoint(RpcService rpcService, CompletableFuture<Void> postStopFuture) {
+ protected AsynchronousOnStopEndpoint(RpcService rpcService, CompletableFuture<Void> onStopFuture) {
super(rpcService);
- this.postStopFuture = Preconditions.checkNotNull(postStopFuture);
+ this.onStopFuture = Preconditions.checkNotNull(onStopFuture);
+ }
+
+ @Override
+ public CompletableFuture<Void> onStop() {
+ return onStopFuture;
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static class MainThreadExecutorOnStopEndpoint extends RpcEndpoint {
+
+ protected MainThreadExecutorOnStopEndpoint(RpcService rpcService) {
+ super(rpcService);
}
@Override
- public CompletableFuture<Void> postStop() {
- return postStopFuture;
+ public CompletableFuture<Void> onStop() {
+ return CompletableFuture.runAsync(() -> {}, getMainThreadExecutor());
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceTest.java
index 1b81331..4be670b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceTest.java
@@ -23,14 +23,24 @@ import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.concurrent.ScheduledExecutor;
+import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.FlinkException;
import org.apache.flink.util.TestLogger;
import akka.actor.ActorSystem;
import akka.actor.Terminated;
import org.junit.AfterClass;
+import org.junit.BeforeClass;
import org.junit.Test;
+import javax.annotation.Nonnull;
+
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
@@ -38,8 +48,10 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -48,24 +60,33 @@ import static org.junit.Assert.fail;
*/
public class AkkaRpcServiceTest extends TestLogger {
+ private static final Time TIMEOUT = Time.milliseconds(10000L);
+
// ------------------------------------------------------------------------
// shared test members
// ------------------------------------------------------------------------
- private static final ActorSystem ACTOR_SYSTEM = AkkaUtils.createDefaultActorSystem();
+ private static ActorSystem actorSystem;
- private static final Time TIMEOUT = Time.milliseconds(10000L);
+ private static AkkaRpcService akkaRpcService;
- private static final AkkaRpcService AKKA_RPC_SERVICE = new AkkaRpcService(ACTOR_SYSTEM, AkkaRpcServiceConfiguration.defaultConfiguration());
+ @BeforeClass
+ public static void setup() {
+ actorSystem = AkkaUtils.createDefaultActorSystem();
+ akkaRpcService = new AkkaRpcService(actorSystem, AkkaRpcServiceConfiguration.defaultConfiguration());
+ }
@AfterClass
public static void shutdown() throws InterruptedException, ExecutionException, TimeoutException {
- final CompletableFuture<Void> rpcTerminationFuture = AKKA_RPC_SERVICE.stopService();
- final CompletableFuture<Terminated> actorSystemTerminationFuture = FutureUtils.toJava(ACTOR_SYSTEM.terminate());
+ final CompletableFuture<Void> rpcTerminationFuture = akkaRpcService.stopService();
+ final CompletableFuture<Terminated> actorSystemTerminationFuture = FutureUtils.toJava(actorSystem.terminate());
FutureUtils
.waitForAll(Arrays.asList(rpcTerminationFuture, actorSystemTerminationFuture))
.get(TIMEOUT.toMilliseconds(), TimeUnit.MILLISECONDS);
+
+ actorSystem = null;
+ akkaRpcService = null;
}
// ------------------------------------------------------------------------
@@ -78,7 +99,7 @@ public class AkkaRpcServiceTest extends TestLogger {
final long delay = 100L;
final long start = System.nanoTime();
- ScheduledFuture<?> scheduledFuture = AKKA_RPC_SERVICE.scheduleRunnable(latch::trigger, delay, TimeUnit.MILLISECONDS);
+ ScheduledFuture<?> scheduledFuture = akkaRpcService.scheduleRunnable(latch::trigger, delay, TimeUnit.MILLISECONDS);
scheduledFuture.get();
@@ -95,7 +116,7 @@ public class AkkaRpcServiceTest extends TestLogger {
public void testExecuteRunnable() throws Exception {
final OneShotLatch latch = new OneShotLatch();
- AKKA_RPC_SERVICE.execute(latch::trigger);
+ akkaRpcService.execute(latch::trigger);
latch.await(30L, TimeUnit.SECONDS);
}
@@ -109,7 +130,7 @@ public class AkkaRpcServiceTest extends TestLogger {
final OneShotLatch latch = new OneShotLatch();
final int expected = 42;
- CompletableFuture<Integer> result = AKKA_RPC_SERVICE.execute(() -> {
+ CompletableFuture<Integer> result = akkaRpcService.execute(() -> {
latch.trigger();
return expected;
});
@@ -122,12 +143,12 @@ public class AkkaRpcServiceTest extends TestLogger {
@Test
public void testGetAddress() {
- assertEquals(AkkaUtils.getAddress(ACTOR_SYSTEM).host().get(), AKKA_RPC_SERVICE.getAddress());
+ assertEquals(AkkaUtils.getAddress(actorSystem).host().get(), akkaRpcService.getAddress());
}
@Test
public void testGetPort() {
- assertEquals(AkkaUtils.getAddress(ACTOR_SYSTEM).port().get(), AKKA_RPC_SERVICE.getPort());
+ assertEquals(AkkaUtils.getAddress(actorSystem).port().get(), akkaRpcService.getPort());
}
/**
@@ -135,15 +156,13 @@ public class AkkaRpcServiceTest extends TestLogger {
*/
@Test(timeout = 60000)
public void testTerminationFuture() throws Exception {
- final ActorSystem actorSystem = AkkaUtils.createDefaultActorSystem();
- final AkkaRpcService rpcService = new AkkaRpcService(
- actorSystem, AkkaRpcServiceConfiguration.defaultConfiguration());
+ final AkkaRpcService rpcService = startAkkaRpcService();
CompletableFuture<Void> terminationFuture = rpcService.getTerminationFuture();
assertFalse(terminationFuture.isDone());
- CompletableFuture.runAsync(rpcService::stopService, actorSystem.dispatcher());
+ rpcService.stopService();
terminationFuture.get();
}
@@ -154,7 +173,7 @@ public class AkkaRpcServiceTest extends TestLogger {
*/
@Test(timeout = 60000)
public void testScheduledExecutorServiceSimpleSchedule() throws Exception {
- ScheduledExecutor scheduledExecutor = AKKA_RPC_SERVICE.getScheduledExecutor();
+ ScheduledExecutor scheduledExecutor = akkaRpcService.getScheduledExecutor();
final OneShotLatch latch = new OneShotLatch();
@@ -175,7 +194,7 @@ public class AkkaRpcServiceTest extends TestLogger {
*/
@Test(timeout = 60000)
public void testScheduledExecutorServicePeriodicSchedule() throws Exception {
- ScheduledExecutor scheduledExecutor = AKKA_RPC_SERVICE.getScheduledExecutor();
+ ScheduledExecutor scheduledExecutor = akkaRpcService.getScheduledExecutor();
final int tries = 4;
final long delay = 10L;
@@ -210,7 +229,7 @@ public class AkkaRpcServiceTest extends TestLogger {
*/
@Test(timeout = 60000)
public void testScheduledExecutorServiceWithFixedDelaySchedule() throws Exception {
- ScheduledExecutor scheduledExecutor = AKKA_RPC_SERVICE.getScheduledExecutor();
+ ScheduledExecutor scheduledExecutor = akkaRpcService.getScheduledExecutor();
final int tries = 4;
final long delay = 10L;
@@ -244,7 +263,7 @@ public class AkkaRpcServiceTest extends TestLogger {
*/
@Test
public void testScheduledExecutorServiceCancelWithFixedDelay() throws InterruptedException {
- ScheduledExecutor scheduledExecutor = AKKA_RPC_SERVICE.getScheduledExecutor();
+ ScheduledExecutor scheduledExecutor = akkaRpcService.getScheduledExecutor();
long delay = 10L;
@@ -285,4 +304,117 @@ public class AkkaRpcServiceTest extends TestLogger {
// expected
}
}
+
+ /**
+ * Tests that the {@link AkkaRpcService} terminates all its RpcEndpoints when shutting down.
+ */
+ @Test
+ public void testAkkaRpcServiceShutDownWithRpcEndpoints() throws Exception {
+ final AkkaRpcService akkaRpcService = startAkkaRpcService();
+
+ try {
+ final int numberActors = 5;
+
+ CompletableFuture<Void> terminationFuture = akkaRpcService.getTerminationFuture();
+
+ final Collection<CompletableFuture<Void>> onStopFutures = startStopNCountingAsynchronousOnStopEndpoints(akkaRpcService, numberActors);
+
+ for (CompletableFuture<Void> onStopFuture : onStopFutures) {
+ onStopFuture.complete(null);
+ }
+
+ terminationFuture.get();
+ assertThat(akkaRpcService.getActorSystem().isTerminated(), is(true));
+ } finally {
+ RpcUtils.terminateRpcService(akkaRpcService, TIMEOUT);
+ }
+ }
+
+ /**
+ * Tests that {@link AkkaRpcService} terminates all its RpcEndpoints and also stops
+ * the underlying {@link ActorSystem} if one of the RpcEndpoints fails while stopping.
+ */
+ @Test
+ public void testAkkaRpcServiceShutDownWithFailingRpcEndpoints() throws Exception {
+ final AkkaRpcService akkaRpcService = startAkkaRpcService();
+
+ final int numberActors = 5;
+
+ CompletableFuture<Void> terminationFuture = akkaRpcService.getTerminationFuture();
+
+ final Collection<CompletableFuture<Void>> onStopFutures = startStopNCountingAsynchronousOnStopEndpoints(akkaRpcService, numberActors);
+
+ Iterator<CompletableFuture<Void>> iterator = onStopFutures.iterator();
+
+ for (int i = 0; i < numberActors - 1; i++) {
+ iterator.next().complete(null);
+ }
+
+ iterator.next().completeExceptionally(new OnStopException("onStop exception occurred."));
+
+ for (CompletableFuture<Void> onStopFuture : onStopFutures) {
+ onStopFuture.complete(null);
+ }
+
+ try {
+ terminationFuture.get();
+ fail("Expected the termination future to complete exceptionally.");
+ } catch (ExecutionException e) {
+ assertThat(ExceptionUtils.findThrowable(e, OnStopException.class).isPresent(), is(true));
+ }
+
+ assertThat(akkaRpcService.getActorSystem().isTerminated(), is(true));
+ }
+
+ private Collection<CompletableFuture<Void>> startStopNCountingAsynchronousOnStopEndpoints(AkkaRpcService akkaRpcService, int numberActors) throws Exception {
+ final Collection<CompletableFuture<Void>> onStopFutures = new ArrayList<>(numberActors);
+
+ final CountDownLatch countDownLatch = new CountDownLatch(numberActors);
+
+ for (int i = 0; i < numberActors; i++) {
+ CompletableFuture<Void> onStopFuture = new CompletableFuture<>();
+ final CountingAsynchronousOnStopEndpoint endpoint = new CountingAsynchronousOnStopEndpoint(akkaRpcService, onStopFuture, countDownLatch);
+ endpoint.start();
+ onStopFutures.add(onStopFuture);
+ }
+
+ CompletableFuture<Void> terminationFuture = akkaRpcService.stopService();
+
+ assertThat(terminationFuture.isDone(), is(false));
+ assertThat(akkaRpcService.getActorSystem().isTerminated(), is(false));
+
+ countDownLatch.await();
+
+ return onStopFutures;
+ }
+
+ @Nonnull
+ private AkkaRpcService startAkkaRpcService() {
+ final ActorSystem actorSystem = AkkaUtils.createDefaultActorSystem();
+ return new AkkaRpcService(actorSystem, AkkaRpcServiceConfiguration.defaultConfiguration());
+ }
+
+ private static class CountingAsynchronousOnStopEndpoint extends AkkaRpcActorTest.AsynchronousOnStopEndpoint {
+
+ private final CountDownLatch countDownLatch;
+
+ protected CountingAsynchronousOnStopEndpoint(RpcService rpcService, CompletableFuture<Void> onStopFuture, CountDownLatch countDownLatch) {
+ super(rpcService, onStopFuture);
+ this.countDownLatch = countDownLatch;
+ }
+
+ @Override
+ public CompletableFuture<Void> onStop() {
+ countDownLatch.countDown();
+ return super.onStop();
+ }
+ }
+
+ private static class OnStopException extends FlinkException {
+ private static final long serialVersionUID = 7136609202083168954L;
+
+ public OnStopException(String message) {
+ super(message);
+ }
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MainThreadValidationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MainThreadValidationTest.java
index c13abdd..1c9f5ca 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MainThreadValidationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MainThreadValidationTest.java
@@ -91,11 +91,6 @@ public class MainThreadValidationTest extends TestLogger {
}
@Override
- public CompletableFuture<Void> postStop() {
- return CompletableFuture.completedFuture(null);
- }
-
- @Override
public void someConcurrencyCriticalFunction() {
validateRunsInMainThread();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MessageSerializationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MessageSerializationTest.java
index e60c675..50cdb2b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MessageSerializationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/MessageSerializationTest.java
@@ -21,14 +21,12 @@ package org.apache.flink.runtime.rpc.akka;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcGateway;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.util.TestLogger;
-import akka.actor.ActorSystem;
import org.hamcrest.core.Is;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@@ -50,35 +48,27 @@ import static org.junit.Assert.fail;
* Tests that akka rpc invocation messages are properly serialized and errors reported
*/
public class MessageSerializationTest extends TestLogger {
- private static ActorSystem actorSystem1;
- private static ActorSystem actorSystem2;
- private static AkkaRpcService akkaRpcService1;
- private static AkkaRpcService akkaRpcService2;
+ private static RpcService akkaRpcService1;
+ private static RpcService akkaRpcService2;
private static final Time timeout = Time.seconds(10L);
private static final int maxFrameSize = 32000;
@BeforeClass
- public static void setup() {
+ public static void setup() throws Exception {
Configuration configuration = new Configuration();
configuration.setString(AkkaOptions.FRAMESIZE, maxFrameSize + "b");
- actorSystem1 = AkkaUtils.createDefaultActorSystem();
- actorSystem2 = AkkaUtils.createDefaultActorSystem();
-
- AkkaRpcServiceConfiguration akkaRpcServiceConfig = AkkaRpcServiceConfiguration.fromConfiguration(configuration);
- akkaRpcService1 = new AkkaRpcService(actorSystem1, akkaRpcServiceConfig);
- akkaRpcService2 = new AkkaRpcService(actorSystem2, akkaRpcServiceConfig);
+ akkaRpcService1 = AkkaRpcServiceUtils.createRpcService("localhost", 0, configuration);
+ akkaRpcService2 = AkkaRpcServiceUtils.createRpcService("localhost", 0, configuration);
}
@AfterClass
public static void teardown() throws InterruptedException, ExecutionException, TimeoutException {
- final Collection<CompletableFuture<?>> terminationFutures = new ArrayList<>(4);
+ final Collection<CompletableFuture<?>> terminationFutures = new ArrayList<>(2);
terminationFutures.add(akkaRpcService1.stopService());
- terminationFutures.add(FutureUtils.toJava(actorSystem1.terminate()));
terminationFutures.add(akkaRpcService2.stopService());
- terminationFutures.add(FutureUtils.toJava(actorSystem2.terminate()));
FutureUtils
.waitForAll(terminationFutures)
@@ -191,11 +181,6 @@ public class MessageSerializationTest extends TestLogger {
public void foobar(Object object) throws InterruptedException {
queue.put(object);
}
-
- @Override
- public CompletableFuture<Void> postStop() {
- return CompletableFuture.completedFuture(null);
- }
}
private static class NonSerializableObject {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/retriever/impl/RpcGatewayRetrieverTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/retriever/impl/RpcGatewayRetrieverTest.java
index f4d66d1..d359cca 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/retriever/impl/RpcGatewayRetrieverTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/retriever/impl/RpcGatewayRetrieverTest.java
@@ -138,10 +138,5 @@ public class RpcGatewayRetrieverTest extends TestLogger {
public UUID getFencingToken() {
return HighAvailabilityServices.DEFAULT_LEADER_ID;
}
-
- @Override
- public CompletableFuture<Void> postStop() {
- return CompletableFuture.completedFuture(null);
- }
}
}
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
index 5a85daa..1f5a2f9 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
@@ -254,7 +254,7 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
}
@Override
- public CompletableFuture<Void> postStop() {
+ public CompletableFuture<Void> onStop() {
// shut down all components
Throwable firstException = null;
@@ -274,7 +274,7 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
}
}
- final CompletableFuture<Void> terminationFuture = super.postStop();
+ final CompletableFuture<Void> terminationFuture = super.onStop();
if (firstException != null) {
return FutureUtils.completedExceptionally(new FlinkException("Error while shutting down YARN resource manager", firstException));