[FLINK-4451] [rpc] Throw RpcConnectionException when rpc endpoint is not 
reachable

This PR introduces a RpcConnectionException which is thrown if the rpc endpoint
is not reachable when calling RpcService.connect.

This closes #2405.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9718dcd3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9718dcd3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9718dcd3

Branch: refs/heads/flip-6
Commit: 9718dcd39742fc542b64d5a7b97a14dc2dd21380
Parents: 12af3b1
Author: Till Rohrmann <trohrm...@apache.org>
Authored: Tue Aug 23 17:59:54 2016 +0200
Committer: Till Rohrmann <trohrm...@apache.org>
Committed: Wed Sep 21 11:39:17 2016 +0200

----------------------------------------------------------------------
 .../registration/RetryingRegistration.java      |  2 +-
 .../apache/flink/runtime/rpc/RpcService.java    |  7 +++-
 .../flink/runtime/rpc/akka/AkkaRpcService.java  | 38 +++++++++++-------
 .../rpc/exceptions/RpcConnectionException.java  | 41 ++++++++++++++++++++
 .../runtime/rpc/akka/AkkaRpcActorTest.java      | 18 +++++++++
 5 files changed, 88 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/9718dcd3/flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java
index 88fe9b5..ea49e42 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java
@@ -197,7 +197,7 @@ public abstract class RetryingRegistration<Gateway extends 
RpcGateway, Success e
                                @Override
                                public void onFailure(Throwable failure) {
                                        if (!isCanceled()) {
-                                               log.warn("Could not resolve {} 
address {}, retrying...", targetName, targetAddress);
+                                               log.warn("Could not resolve {} 
address {}, retrying...", targetName, targetAddress, failure);
                                                startRegistration();
                                        }
                                }

http://git-wip-us.apache.org/repos/asf/flink/blob/9718dcd3/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
index bc0f5cb..78c1cec 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcService.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.rpc;
 
+import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException;
 import scala.concurrent.ExecutionContext;
 import scala.concurrent.Future;
 
@@ -32,12 +33,14 @@ public interface RpcService {
 
        /**
         * Connect to a remote rpc server under the provided address. Returns a 
rpc gateway which can
-        * be used to communicate with the rpc server.
+        * be used to communicate with the rpc server. If the connection 
failed, then the returned
+        * future is failed with a {@link RpcConnectionException}.
         *
         * @param address Address of the remote rpc server
         * @param clazz Class of the rpc gateway to return
         * @param <C> Type of the rpc gateway to return
-        * @return Future containing the rpc gateway
+        * @return Future containing the rpc gateway or an {@link 
RpcConnectionException} if the
+        * connection attempt failed
         */
        <C extends RpcGateway> Future<C> connect(String address, Class<C> 
clazz);
 

http://git-wip-us.apache.org/repos/asf/flink/blob/9718dcd3/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 00a6932..060a1ef 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
@@ -35,6 +35,7 @@ import org.apache.flink.runtime.rpc.RpcGateway;
 import org.apache.flink.runtime.rpc.RpcEndpoint;
 import org.apache.flink.runtime.rpc.RpcService;
 import org.apache.flink.runtime.rpc.StartStoppable;
+import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -99,25 +100,32 @@ public class AkkaRpcService implements RpcService {
                final Future<Object> identify = asker.ask(new Identify(42), 
timeout);
                return identify.map(new Mapper<Object, C>(){
                        @Override
-                       public C apply(Object obj) {
-                               ActorRef actorRef = ((ActorIdentity) 
obj).getRef();
+                       public C checkedApply(Object obj) throws Exception {
 
-                               final String address = 
AkkaUtils.getAkkaURL(actorSystem, actorRef);
+                               ActorIdentity actorIdentity = (ActorIdentity) 
obj;
 
-                               InvocationHandler akkaInvocationHandler = new 
AkkaInvocationHandler(address, actorRef, timeout, maximumFramesize);
+                               if (actorIdentity.getRef() == null) {
+                                       throw new RpcConnectionException("Could 
not connect to rpc endpoint under address " + address + '.');
+                               } else {
+                                       ActorRef actorRef = 
actorIdentity.getRef();
+
+                                       final String address = 
AkkaUtils.getAkkaURL(actorSystem, actorRef);
+
+                                       InvocationHandler akkaInvocationHandler 
= new AkkaInvocationHandler(address, actorRef, timeout, maximumFramesize);
 
-                               // 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();
-                               
-                               @SuppressWarnings("unchecked")
-                               C proxy = (C) Proxy.newProxyInstance(
-                                       classLoader,
-                                       new Class<?>[] {clazz},
-                                       akkaInvocationHandler);
+                                       // 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();
 
-                               return proxy;
+                                       @SuppressWarnings("unchecked")
+                                       C proxy = (C) Proxy.newProxyInstance(
+                                               classLoader,
+                                               new Class<?>[]{clazz},
+                                               akkaInvocationHandler);
+
+                                       return proxy;
+                               }
                        }
                }, actorSystem.dispatcher());
        }

http://git-wip-us.apache.org/repos/asf/flink/blob/9718dcd3/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/exceptions/RpcConnectionException.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/exceptions/RpcConnectionException.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/exceptions/RpcConnectionException.java
new file mode 100644
index 0000000..a22ebe7
--- /dev/null
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/exceptions/RpcConnectionException.java
@@ -0,0 +1,41 @@
+/*
+ * 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.exceptions;
+
+import java.util.concurrent.ExecutionException;
+
+/**
+ * Exception class which is thrown if a rpc connection failed. Usually this 
happens if the remote
+ * host cannot be reached.
+ */
+public class RpcConnectionException extends ExecutionException {
+       private static final long serialVersionUID = -5500560405481142472L;
+
+       public RpcConnectionException(String message) {
+               super(message);
+       }
+
+       public RpcConnectionException(String message, Throwable cause) {
+               super(message, cause);
+       }
+
+       public RpcConnectionException(Throwable cause) {
+               super(cause);
+       }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/9718dcd3/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 82d13f0..a6ceb91 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
@@ -25,6 +25,7 @@ import org.apache.flink.runtime.rpc.RpcEndpoint;
 import org.apache.flink.runtime.rpc.RpcGateway;
 import org.apache.flink.runtime.rpc.RpcMethod;
 import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException;
 import org.apache.flink.util.TestLogger;
 import org.hamcrest.core.Is;
 import org.junit.AfterClass;
@@ -36,6 +37,7 @@ import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
 
 public class AkkaRpcActorTest extends TestLogger {
 
@@ -73,6 +75,22 @@ public class AkkaRpcActorTest extends TestLogger {
        }
 
        /**
+        * Tests that a {@link RpcConnectionException} is thrown if the rpc 
endpoint cannot be connected to.
+        */
+       @Test
+       public void testFailingAddressResolution() throws Exception {
+               Future<DummyRpcGateway> futureRpcGateway = 
akkaRpcService.connect("foobar", DummyRpcGateway.class);
+
+               try {
+                       DummyRpcGateway gateway = 
Await.result(futureRpcGateway, timeout.duration());
+
+                       fail("The rpc connection resolution should have 
failed.");
+               } catch (RpcConnectionException exception) {
+                       // we're expecting a RpcConnectionException
+               }
+       }
+
+       /**
         * Tests that the {@link AkkaRpcActor} stashes messages until the 
corresponding
         * {@link RpcEndpoint} has been started.
         */

Reply via email to