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 2018/05/15 12:24:23 UTC
[2/3] flink git commit: [hotfix] Resolve compiler warnings in
AkkaRpcService
[hotfix] Resolve compiler warnings in AkkaRpcService
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5a093e5f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5a093e5f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5a093e5f
Branch: refs/heads/master
Commit: 5a093e5f90a142396f7978ee1d8ec51379cbdde6
Parents: 17f0e85
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue May 15 10:41:15 2018 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Tue May 15 14:24:02 2018 +0200
----------------------------------------------------------------------
.../flink/runtime/rpc/akka/AkkaRpcService.java | 26 ++++++++++----------
.../flink/runtime/rpc/RpcConnectionTest.java | 2 +-
2 files changed, 14 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5a093e5f/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 8e96492..1151389 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
@@ -44,7 +44,6 @@ import akka.actor.PoisonPill;
import akka.actor.Props;
import akka.actor.Terminated;
import akka.dispatch.Futures;
-import akka.dispatch.Mapper;
import akka.pattern.Patterns;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -61,6 +60,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -68,6 +68,7 @@ import java.util.function.Function;
import scala.Option;
import scala.concurrent.Future;
+import scala.reflect.ClassTag$;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -378,7 +379,7 @@ public class AkkaRpcService implements RpcService {
@Override
public <T> CompletableFuture<T> execute(Callable<T> callable) {
- Future<T> scalaFuture = Futures.future(callable, actorSystem.dispatcher());
+ Future<T> scalaFuture = Futures.<T>future(callable, actorSystem.dispatcher());
return FutureUtils.toJava(scalaFuture);
}
@@ -411,15 +412,16 @@ public class AkkaRpcService implements RpcService {
final ActorSelection actorSel = actorSystem.actorSelection(address);
- final Future<Object> identify = Patterns.ask(actorSel, new Identify(42), timeout.toMilliseconds());
- final Future<C> resultFuture = identify.map(new Mapper<Object, C>(){
- @Override
- public C checkedApply(Object obj) throws Exception {
+ final Future<ActorIdentity> identify = Patterns
+ .ask(actorSel, new Identify(42), timeout.toMilliseconds())
+ .<ActorIdentity>mapTo(ClassTag$.MODULE$.<ActorIdentity>apply(ActorIdentity.class));
- ActorIdentity actorIdentity = (ActorIdentity) obj;
+ final CompletableFuture<ActorIdentity> identifyFuture = FutureUtils.toJava(identify);
+ return identifyFuture.thenApplyAsync(
+ (ActorIdentity actorIdentity) -> {
if (actorIdentity.getRef() == null) {
- throw new RpcConnectionException("Could not connect to rpc endpoint under address " + address + '.');
+ throw new CompletionException(new RpcConnectionException("Could not connect to rpc endpoint under address " + address + '.'));
} else {
ActorRef actorRef = actorIdentity.getRef();
@@ -428,7 +430,7 @@ public class AkkaRpcService implements RpcService {
// Rather than using the System ClassLoader directly, we derive the ClassLoader
// from this class . That works better in cases where Flink runs embedded and all Flink
// code is loaded dynamically (for example from an OSGI bundle) through a custom ClassLoader
- ClassLoader classLoader = AkkaRpcService.this.getClass().getClassLoader();
+ ClassLoader classLoader = getClass().getClassLoader();
@SuppressWarnings("unchecked")
C proxy = (C) Proxy.newProxyInstance(
@@ -438,9 +440,7 @@ public class AkkaRpcService implements RpcService {
return proxy;
}
- }
- }, actorSystem.dispatcher());
-
- return FutureUtils.toJava(resultFuture);
+ },
+ actorSystem.dispatcher());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5a093e5f/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java
index 78137e8..2b88a17 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java
@@ -58,7 +58,7 @@ public class RpcConnectionTest extends TestLogger {
RpcService rpcService = null;
try {
actorSystem = AkkaUtils.createActorSystem(
- new Configuration(), Option.apply(new Tuple2<String, Object>("localhost", 0)));
+ new Configuration(), Option.<Tuple2<String, Object>>apply(new Tuple2<>("localhost", 0)));
// we start the RPC service with a very long timeout to ensure that the test
// can only pass if the connection problem is not recognized merely via a timeout