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 2017/10/05 12:53:10 UTC
flink git commit: [FLINK-7754] [rpc] Complete termination future
after actor has been stopped
Repository: flink
Updated Branches:
refs/heads/master dae21da7e -> 4947ee669
[FLINK-7754] [rpc] Complete termination future after actor has been stopped
This commit waits not only until the Actor has called postStop but also until the actor
has been completely shut down by the ActorSystem before completing the termination
future.
This closes #4770.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4947ee66
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4947ee66
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4947ee66
Branch: refs/heads/master
Commit: 4947ee669fe267e3f71853cd14228b350cf10ac8
Parents: dae21da
Author: Till Rohrmann <tr...@apache.org>
Authored: Mon Oct 2 22:29:12 2017 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Oct 5 14:52:43 2017 +0200
----------------------------------------------------------------------
.../apache/flink/runtime/rpc/RpcEndpoint.java | 5 +-
.../org/apache/flink/runtime/rpc/RpcServer.java | 2 +-
.../runtime/rpc/akka/AkkaBasedEndpoint.java | 45 ++++++++
.../flink/runtime/rpc/akka/AkkaGateway.java | 30 ------
.../runtime/rpc/akka/AkkaInvocationHandler.java | 24 +++--
.../flink/runtime/rpc/akka/AkkaRpcActor.java | 10 +-
.../flink/runtime/rpc/akka/AkkaRpcService.java | 105 +++++++++++++++----
.../rpc/akka/FencedAkkaInvocationHandler.java | 13 +--
.../runtime/rpc/akka/FencedAkkaRpcActor.java | 4 +-
.../runtime/taskexecutor/TaskManagerRunner.java | 2 +-
.../runtime/rpc/akka/AkkaRpcActorTest.java | 32 +++++-
.../runtime/rpc/akka/AkkaRpcServiceTest.java | 8 +-
12 files changed, 203 insertions(+), 77 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java
----------------------------------------------------------------------
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 563674a..9c27c95 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
@@ -223,11 +223,12 @@ public abstract class RpcEndpoint implements RpcGateway {
}
/**
- * Return a future which is completed when the rpc endpoint has been terminated.
+ * Return a future which is completed with true when the rpc endpoint has been terminated.
+ * In case of a failure, this future is completed with the occurring exception.
*
* @return Future which is completed when the rpc endpoint has been terminated.
*/
- public CompletableFuture<Void> getTerminationFuture() {
+ public CompletableFuture<Boolean> getTerminationFuture() {
return rpcServer.getTerminationFuture();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java
index 9c645cc..25ce10a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java
@@ -30,5 +30,5 @@ public interface RpcServer extends StartStoppable, MainThreadExecutable, RpcGate
*
* @return Future indicating when the rpc endpoint has been terminated
*/
- CompletableFuture<Void> getTerminationFuture();
+ CompletableFuture<Boolean> getTerminationFuture();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaBasedEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaBasedEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaBasedEndpoint.java
new file mode 100644
index 0000000..499de1e
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaBasedEndpoint.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rpc.akka;
+
+import org.apache.flink.runtime.rpc.RpcGateway;
+
+import akka.actor.ActorRef;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Interface for Akka based rpc gateways
+ */
+interface AkkaBasedEndpoint extends RpcGateway {
+
+ /**
+ * Returns the {@link ActorRef} of the underlying RPC actor.
+ *
+ * @return the {@link ActorRef} of the underlying RPC actor
+ */
+ ActorRef getActorRef();
+
+ /**
+ * Returns the internal termination future.
+ *
+ * @return Internal termination future
+ */
+ CompletableFuture<Void> getInternalTerminationFuture();
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaGateway.java
deleted file mode 100644
index f6125dc..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaGateway.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.rpc.akka;
-
-import akka.actor.ActorRef;
-import org.apache.flink.runtime.rpc.RpcGateway;
-
-/**
- * Interface for Akka based rpc gateways
- */
-interface AkkaGateway extends RpcGateway {
-
- ActorRef getRpcEndpoint();
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java
----------------------------------------------------------------------
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 fc785cb..37f46e3 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
@@ -57,7 +57,7 @@ import static org.apache.flink.util.Preconditions.checkArgument;
* rpc in a {@link LocalRpcInvocation} message and then sends it to the {@link AkkaRpcActor} where it is
* executed.
*/
-class AkkaInvocationHandler implements InvocationHandler, AkkaGateway, RpcServer {
+class AkkaInvocationHandler implements InvocationHandler, AkkaBasedEndpoint, RpcServer {
private static final Logger LOG = LoggerFactory.getLogger(AkkaInvocationHandler.class);
/**
@@ -82,7 +82,12 @@ class AkkaInvocationHandler implements InvocationHandler, AkkaGateway, RpcServer
private final long maximumFramesize;
// null if gateway; otherwise non-null
- private final CompletableFuture<Void> terminationFuture;
+ @Nullable
+ private final CompletableFuture<Boolean> terminationFuture;
+
+ // null if gateway; otherwise non-null
+ @Nullable
+ private final CompletableFuture<Void> internalTerminationFuture;
AkkaInvocationHandler(
String address,
@@ -90,7 +95,8 @@ class AkkaInvocationHandler implements InvocationHandler, AkkaGateway, RpcServer
ActorRef rpcEndpoint,
Time timeout,
long maximumFramesize,
- @Nullable CompletableFuture<Void> terminationFuture) {
+ @Nullable CompletableFuture<Boolean> terminationFuture,
+ @Nullable CompletableFuture<Void> internalTerminationFuture) {
this.address = Preconditions.checkNotNull(address);
this.hostname = Preconditions.checkNotNull(hostname);
@@ -99,6 +105,7 @@ class AkkaInvocationHandler implements InvocationHandler, AkkaGateway, RpcServer
this.timeout = Preconditions.checkNotNull(timeout);
this.maximumFramesize = maximumFramesize;
this.terminationFuture = terminationFuture;
+ this.internalTerminationFuture = internalTerminationFuture;
}
@Override
@@ -107,7 +114,7 @@ class AkkaInvocationHandler implements InvocationHandler, AkkaGateway, RpcServer
Object result;
- if (declaringClass.equals(AkkaGateway.class) ||
+ if (declaringClass.equals(AkkaBasedEndpoint.class) ||
declaringClass.equals(Object.class) ||
declaringClass.equals(RpcGateway.class) ||
declaringClass.equals(StartStoppable.class) ||
@@ -127,7 +134,7 @@ class AkkaInvocationHandler implements InvocationHandler, AkkaGateway, RpcServer
}
@Override
- public ActorRef getRpcEndpoint() {
+ public ActorRef getActorRef() {
return rpcEndpoint;
}
@@ -339,7 +346,12 @@ class AkkaInvocationHandler implements InvocationHandler, AkkaGateway, RpcServer
}
@Override
- public CompletableFuture<Void> getTerminationFuture() {
+ public CompletableFuture<Boolean> getTerminationFuture() {
return terminationFuture;
}
+
+ @Override
+ public CompletableFuture<Void> getInternalTerminationFuture() {
+ return internalTerminationFuture;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
----------------------------------------------------------------------
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 f6c2e8b..f7488ab 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
@@ -77,12 +77,12 @@ class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends UntypedActor {
/** the helper that tracks whether calls come from the main thread. */
private final MainThreadValidatorUtil mainThreadValidator;
- private final CompletableFuture<Void> terminationFuture;
+ private final CompletableFuture<Void> internalTerminationFuture;
- AkkaRpcActor(final T rpcEndpoint, final CompletableFuture<Void> terminationFuture) {
+ AkkaRpcActor(final T rpcEndpoint, final CompletableFuture<Void> internalTerminationFuture) {
this.rpcEndpoint = checkNotNull(rpcEndpoint, "rpc endpoint");
this.mainThreadValidator = new MainThreadValidatorUtil(rpcEndpoint);
- this.terminationFuture = checkNotNull(terminationFuture);
+ this.internalTerminationFuture = checkNotNull(internalTerminationFuture);
}
@Override
@@ -106,9 +106,9 @@ class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends UntypedActor {
// Complete the termination future so that others know that we've stopped.
if (shutdownThrowable != null) {
- terminationFuture.completeExceptionally(shutdownThrowable);
+ internalTerminationFuture.completeExceptionally(shutdownThrowable);
} else {
- terminationFuture.complete(null);
+ internalTerminationFuture.complete(null);
}
} finally {
mainThreadValidator.exitMainThread();
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java
----------------------------------------------------------------------
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 259f0ee..2d98552 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
@@ -40,6 +40,7 @@ import akka.actor.ActorSelection;
import akka.actor.ActorSystem;
import akka.actor.Address;
import akka.actor.Identify;
+import akka.actor.Kill;
import akka.actor.Props;
import akka.dispatch.Futures;
import akka.dispatch.Mapper;
@@ -47,18 +48,24 @@ import akka.pattern.Patterns;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+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.HashMap;
import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import scala.Option;
@@ -83,7 +90,10 @@ public class AkkaRpcService implements RpcService {
private final ActorSystem actorSystem;
private final Time timeout;
- private final Set<ActorRef> actors = new HashSet<>(4);
+
+ @GuardedBy("lock")
+ private final Map<ActorRef, RpcEndpoint> actors = new HashMap<>(4);
+
private final long maximumFramesize;
private final String address;
@@ -119,6 +129,8 @@ public class AkkaRpcService implements RpcService {
}
internalScheduledExecutor = new ActorSystemScheduledExecutorAdapter(actorSystem);
+
+ stopped = false;
}
@Override
@@ -149,6 +161,7 @@ public class AkkaRpcService implements RpcService {
actorRef,
timeout,
maximumFramesize,
+ null,
null);
});
}
@@ -169,6 +182,7 @@ public class AkkaRpcService implements RpcService {
timeout,
maximumFramesize,
null,
+ null,
() -> fencingToken);
});
}
@@ -177,13 +191,14 @@ public class AkkaRpcService implements RpcService {
public <C extends RpcEndpoint & RpcGateway> RpcServer startServer(C rpcEndpoint) {
checkNotNull(rpcEndpoint, "rpc endpoint");
- CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ CompletableFuture<Boolean> terminationFuture = new CompletableFuture<>();
+ CompletableFuture<Void> internalTerminationFuture = new CompletableFuture<>();
final Props akkaRpcActorProps;
if (rpcEndpoint instanceof FencedRpcEndpoint) {
- akkaRpcActorProps = Props.create(FencedAkkaRpcActor.class, rpcEndpoint, terminationFuture);
+ akkaRpcActorProps = Props.create(FencedAkkaRpcActor.class, rpcEndpoint, internalTerminationFuture);
} else {
- akkaRpcActorProps = Props.create(AkkaRpcActor.class, rpcEndpoint, terminationFuture);
+ akkaRpcActorProps = Props.create(AkkaRpcActor.class, rpcEndpoint, internalTerminationFuture);
}
ActorRef actorRef;
@@ -191,12 +206,12 @@ public class AkkaRpcService implements RpcService {
synchronized (lock) {
checkState(!stopped, "RpcService is stopped");
actorRef = actorSystem.actorOf(akkaRpcActorProps, rpcEndpoint.getEndpointId());
- actors.add(actorRef);
+ actors.put(actorRef, rpcEndpoint);
}
LOG.info("Starting RPC endpoint for {} at {} .", rpcEndpoint.getClass().getName(), actorRef.path());
- final String address = AkkaUtils.getAkkaURL(actorSystem, actorRef);
+ final String akkaAddress = AkkaUtils.getAkkaURL(actorSystem, actorRef);
final String hostname;
Option<String> host = actorRef.path().address().host();
if (host.isEmpty()) {
@@ -208,30 +223,32 @@ public class AkkaRpcService implements RpcService {
Set<Class<?>> implementedRpcGateways = new HashSet<>(RpcUtils.extractImplementedRpcGateways(rpcEndpoint.getClass()));
implementedRpcGateways.add(RpcServer.class);
- implementedRpcGateways.add(AkkaGateway.class);
+ implementedRpcGateways.add(AkkaBasedEndpoint.class);
final InvocationHandler akkaInvocationHandler;
if (rpcEndpoint instanceof FencedRpcEndpoint) {
// a FencedRpcEndpoint needs a FencedAkkaInvocationHandler
akkaInvocationHandler = new FencedAkkaInvocationHandler<>(
- address,
+ akkaAddress,
hostname,
actorRef,
timeout,
maximumFramesize,
terminationFuture,
+ internalTerminationFuture,
((FencedRpcEndpoint<?>) rpcEndpoint)::getFencingToken);
implementedRpcGateways.add(FencedMainThreadExecutable.class);
} else {
akkaInvocationHandler = new AkkaInvocationHandler(
- address,
+ akkaAddress,
hostname,
actorRef,
timeout,
maximumFramesize,
- terminationFuture);
+ terminationFuture,
+ internalTerminationFuture);
}
// Rather than using the System ClassLoader directly, we derive the ClassLoader
@@ -250,15 +267,16 @@ public class AkkaRpcService implements RpcService {
@Override
public <F extends Serializable> RpcServer fenceRpcServer(RpcServer rpcServer, F fencingToken) {
- if (rpcServer instanceof AkkaGateway) {
+ if (rpcServer instanceof AkkaBasedEndpoint) {
InvocationHandler fencedInvocationHandler = new FencedAkkaInvocationHandler<>(
rpcServer.getAddress(),
rpcServer.getHostname(),
- ((AkkaGateway) rpcServer).getRpcEndpoint(),
+ ((AkkaBasedEndpoint) rpcServer).getActorRef(),
timeout,
maximumFramesize,
null,
+ null,
() -> fencingToken);
// Rather than using the System ClassLoader directly, we derive the ClassLoader
@@ -268,7 +286,7 @@ public class AkkaRpcService implements RpcService {
return (RpcServer) Proxy.newProxyInstance(
classLoader,
- new Class<?>[]{RpcServer.class, AkkaGateway.class},
+ new Class<?>[]{RpcServer.class, AkkaBasedEndpoint.class},
fencedInvocationHandler);
} else {
throw new RuntimeException("The given RpcServer must implement the AkkaGateway in order to fence it.");
@@ -277,24 +295,45 @@ public class AkkaRpcService implements RpcService {
@Override
public void stopServer(RpcServer selfGateway) {
- if (selfGateway instanceof AkkaGateway) {
- AkkaGateway akkaClient = (AkkaGateway) selfGateway;
+ if (selfGateway instanceof AkkaBasedEndpoint) {
+ AkkaBasedEndpoint akkaClient = (AkkaBasedEndpoint) selfGateway;
boolean fromThisService;
synchronized (lock) {
if (stopped) {
return;
} else {
- fromThisService = actors.remove(akkaClient.getRpcEndpoint());
+ fromThisService = actors.remove(akkaClient.getActorRef()) != null;
}
}
if (fromThisService) {
- ActorRef selfActorRef = akkaClient.getRpcEndpoint();
+ ActorRef selfActorRef = akkaClient.getActorRef();
LOG.info("Trigger shut down of RPC endpoint {}.", selfActorRef.path());
- actorSystem.stop(selfActorRef);
+
+ CompletableFuture<Boolean> akkaTerminationFuture = FutureUtils.toJava(
+ Patterns.gracefulStop(
+ selfActorRef,
+ FutureUtils.toFiniteDuration(timeout),
+ Kill.getInstance()));
+
+ akkaTerminationFuture
+ .thenCombine(
+ akkaClient.getInternalTerminationFuture(),
+ (Boolean terminated, Void ignored) -> true)
+ .whenComplete(
+ (Boolean terminated, Throwable throwable) -> {
+ if (throwable != null) {
+ LOG.debug("Graceful RPC endpoint shutdown failed.", throwable);
+
+ actorSystem.stop(selfActorRef);
+ selfGateway.getTerminationFuture().completeExceptionally(throwable);
+ } else {
+ selfGateway.getTerminationFuture().complete(null);
+ }
+ });
} else {
- LOG.debug("RPC endpoint {} already stopped or from different RPC service");
+ LOG.debug("RPC endpoint {} already stopped or from different RPC service", selfGateway.getAddress());
}
}
}
@@ -303,6 +342,8 @@ public class AkkaRpcService implements RpcService {
public void stopService() {
LOG.info("Stopping Akka RPC service.");
+ final List<RpcEndpoint> actorsToTerminate;
+
synchronized (lock) {
if (stopped) {
return;
@@ -311,11 +352,36 @@ public class AkkaRpcService implements RpcService {
stopped = true;
actorSystem.shutdown();
+
+ actorsToTerminate = new ArrayList<>(actors.values());
+
actors.clear();
}
actorSystem.awaitTermination();
+ // complete the termination futures of all actors
+ for (RpcEndpoint rpcEndpoint : actorsToTerminate) {
+ final CompletableFuture<Boolean> terminationFuture = rpcEndpoint.getTerminationFuture();
+
+ AkkaBasedEndpoint akkaBasedEndpoint = rpcEndpoint.getSelfGateway(AkkaBasedEndpoint.class);
+
+ CompletableFuture<Void> internalTerminationFuture = akkaBasedEndpoint.getInternalTerminationFuture();
+
+ internalTerminationFuture.whenComplete(
+ (Void ignored, Throwable throwable) -> {
+ if (throwable != null) {
+ terminationFuture.completeExceptionally(throwable);
+ } else {
+ terminationFuture.complete(true);
+ }
+ });
+
+ // make sure that if the internal termination futures haven't completed yet, then they time out
+ internalTerminationFuture.completeExceptionally(
+ new TimeoutException("The RpcEndpoint " + rpcEndpoint.getAddress() + " did not terminate in time."));
+ }
+
LOG.info("Stopped Akka RPC service.");
}
@@ -331,6 +397,7 @@ public class AkkaRpcService implements RpcService {
return actorSystem.dispatcher();
}
+ @Override
public ScheduledExecutor getScheduledExecutor() {
return internalScheduledExecutor;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/FencedAkkaInvocationHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/FencedAkkaInvocationHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/FencedAkkaInvocationHandler.java
index 9d2c295..03534ae 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/FencedAkkaInvocationHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/FencedAkkaInvocationHandler.java
@@ -60,9 +60,10 @@ public class FencedAkkaInvocationHandler<F extends Serializable> extends AkkaInv
ActorRef rpcEndpoint,
Time timeout,
long maximumFramesize,
- @Nullable CompletableFuture<Void> terminationFuture,
+ @Nullable CompletableFuture<Boolean> terminationFuture,
+ @Nullable CompletableFuture<Void> internalTerminationFuture,
Supplier<F> fencingTokenSupplier) {
- super(address, hostname, rpcEndpoint, timeout, maximumFramesize, terminationFuture);
+ super(address, hostname, rpcEndpoint, timeout, maximumFramesize, terminationFuture, internalTerminationFuture);
this.fencingTokenSupplier = Preconditions.checkNotNull(fencingTokenSupplier);
}
@@ -84,11 +85,11 @@ public class FencedAkkaInvocationHandler<F extends Serializable> extends AkkaInv
checkNotNull(runnable, "runnable");
if (isLocal) {
- getRpcEndpoint().tell(
+ getActorRef().tell(
new UnfencedMessage<>(new RunAsync(runnable, 0L)), ActorRef.noSender());
} else {
throw new RuntimeException("Trying to send a Runnable to a remote actor at " +
- getRpcEndpoint().path() + ". This is not supported.");
+ getActorRef().path() + ". This is not supported.");
}
}
@@ -101,14 +102,14 @@ public class FencedAkkaInvocationHandler<F extends Serializable> extends AkkaInv
@SuppressWarnings("unchecked")
CompletableFuture<V> resultFuture = (CompletableFuture<V>) FutureUtils.toJava(
Patterns.ask(
- getRpcEndpoint(),
+ getActorRef(),
new UnfencedMessage<>(new CallAsync(callable)),
timeout.toMilliseconds()));
return resultFuture;
} else {
throw new RuntimeException("Trying to send a Runnable to a remote actor at " +
- getRpcEndpoint().path() + ". This is not supported.");
+ getActorRef().path() + ". This is not supported.");
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/FencedAkkaRpcActor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/FencedAkkaRpcActor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/FencedAkkaRpcActor.java
index 369af6a..fa83e4f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/FencedAkkaRpcActor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/FencedAkkaRpcActor.java
@@ -38,8 +38,8 @@ import java.util.concurrent.CompletableFuture;
*/
public class FencedAkkaRpcActor<F extends Serializable, T extends FencedRpcEndpoint<F> & RpcGateway> extends AkkaRpcActor<T> {
- public FencedAkkaRpcActor(T rpcEndpoint, CompletableFuture<Void> terminationFuture) {
- super(rpcEndpoint, terminationFuture);
+ public FencedAkkaRpcActor(T rpcEndpoint, CompletableFuture<Void> internalTerminationFuture) {
+ super(rpcEndpoint, internalTerminationFuture);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index 2fe523c..26671a8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -163,7 +163,7 @@ public class TaskManagerRunner implements FatalErrorHandler {
}
// export the termination future for caller to know it is terminated
- public CompletableFuture<Void> getTerminationFuture() {
+ public CompletableFuture<Boolean> getTerminationFuture() {
return taskManager.getTerminationFuture();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActorTest.java
----------------------------------------------------------------------
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 56d17e9..dd49ee1 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
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.rpc.akka;
import akka.actor.ActorSystem;
import org.apache.flink.api.common.time.Time;
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;
@@ -35,6 +36,7 @@ import org.junit.Test;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -174,7 +176,7 @@ public class AkkaRpcActorTest extends TestLogger {
final DummyRpcEndpoint rpcEndpoint = new DummyRpcEndpoint(akkaRpcService);
rpcEndpoint.start();
- CompletableFuture<Void> terminationFuture = rpcEndpoint.getTerminationFuture();
+ CompletableFuture<Boolean> terminationFuture = rpcEndpoint.getTerminationFuture();
assertFalse(terminationFuture.isDone());
@@ -235,7 +237,7 @@ public class AkkaRpcActorTest extends TestLogger {
rpcEndpoint.shutDown();
- CompletableFuture<Void> terminationFuture = rpcEndpoint.getTerminationFuture();
+ CompletableFuture<Boolean> terminationFuture = rpcEndpoint.getTerminationFuture();
try {
terminationFuture.get();
@@ -254,13 +256,37 @@ public class AkkaRpcActorTest extends TestLogger {
simpleRpcEndpoint.shutDown();
- CompletableFuture<Void> terminationFuture = simpleRpcEndpoint.getTerminationFuture();
+ CompletableFuture<Boolean> terminationFuture = simpleRpcEndpoint.getTerminationFuture();
// check that we executed the postStop method in the main thread, otherwise an exception
// would be thrown here.
terminationFuture.get();
}
+ /**
+ * Tests that actors are properly terminated when the AkkaRpcService is shut down.
+ */
+ @Test
+ public void testActorTerminationWhenServiceShutdown() throws Exception {
+ final ActorSystem rpcActorSystem = AkkaUtils.createDefaultActorSystem();
+ final RpcService rpcService = new AkkaRpcService(rpcActorSystem, timeout);
+
+ try {
+ SimpleRpcEndpoint rpcEndpoint = new SimpleRpcEndpoint(rpcService, SimpleRpcEndpoint.class.getSimpleName());
+
+ rpcEndpoint.start();
+
+ CompletableFuture<Boolean> terminationFuture = rpcEndpoint.getTerminationFuture();
+
+ rpcService.stopService();
+
+ terminationFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ } finally {
+ rpcActorSystem.shutdown();
+ rpcActorSystem.awaitTermination(FutureUtils.toFiniteDuration(timeout));
+ }
+ }
+
// ------------------------------------------------------------------------
// Test Actors and Interfaces
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4947ee66/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceTest.java
----------------------------------------------------------------------
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 e0d1110..d20a090 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
@@ -18,13 +18,14 @@
package org.apache.flink.runtime.rpc.akka;
-import akka.actor.ActorSystem;
import org.apache.flink.api.common.time.Time;
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.util.TestLogger;
+import akka.actor.ActorSystem;
import org.junit.AfterClass;
import org.junit.Test;
@@ -49,13 +50,16 @@ public class AkkaRpcServiceTest extends TestLogger {
private static ActorSystem actorSystem = AkkaUtils.createDefaultActorSystem();
+ private static final Time timeout = Time.milliseconds(10000);
+
private static AkkaRpcService akkaRpcService =
- new AkkaRpcService(actorSystem, Time.milliseconds(10000));
+ new AkkaRpcService(actorSystem, timeout);
@AfterClass
public static void shutdown() {
akkaRpcService.stopService();
actorSystem.shutdown();
+ actorSystem.awaitTermination(FutureUtils.toFiniteDuration(timeout));
}
// ------------------------------------------------------------------------