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 2016/08/10 17:06:17 UTC
flink git commit: Rename RpcProtocol into RpcEndpoint
Repository: flink
Updated Branches:
refs/heads/flip-6 3ee29cecb -> ad6693ee6
Rename RpcProtocol into RpcEndpoint
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ad6693ee
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ad6693ee
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ad6693ee
Branch: refs/heads/flip-6
Commit: ad6693ee60188a09e3acbd0e3a5957a20ef3e184
Parents: 3ee29ce
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed Aug 10 18:49:59 2016 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Aug 10 18:49:59 2016 +0200
----------------------------------------------------------------------
.../flink/runtime/rpc/MainThreadExecutor.java | 2 +-
.../apache/flink/runtime/rpc/RpcEndpoint.java | 182 +++++++++++++++++++
.../org/apache/flink/runtime/rpc/RpcMethod.java | 2 +-
.../apache/flink/runtime/rpc/RpcProtocol.java | 182 -------------------
.../apache/flink/runtime/rpc/RpcService.java | 10 +-
.../flink/runtime/rpc/akka/AkkaRpcService.java | 18 +-
.../flink/runtime/rpc/jobmaster/JobMaster.java | 4 +-
.../rpc/resourcemanager/ResourceManager.java | 4 +-
.../runtime/rpc/taskexecutor/TaskExecutor.java | 4 +-
.../flink/runtime/rpc/RpcCompletenessTest.java | 68 +++----
10 files changed, 238 insertions(+), 238 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ad6693ee/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/MainThreadExecutor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/MainThreadExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/MainThreadExecutor.java
index 5a71382..e06711e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/MainThreadExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/MainThreadExecutor.java
@@ -28,7 +28,7 @@ import java.util.concurrent.TimeoutException;
* Interface to execute {@link Runnable} and {@link Callable} in the main thread of the underlying
* rpc server.
*
- * This interface is intended to be implemented by the self gateway in a {@link RpcProtocol}
+ * This interface is intended to be implemented by the self gateway in a {@link RpcEndpoint}
* implementation which allows to dispatch local procedures to the main thread of the underlying
* rpc server.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/ad6693ee/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
new file mode 100644
index 0000000..3d8757f
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java
@@ -0,0 +1,182 @@
+/*
+ * 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;
+
+import akka.util.Timeout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.concurrent.ExecutionContext;
+import scala.concurrent.Future;
+
+import java.util.concurrent.Callable;
+
+/**
+ * Base class for rpc endpoints. Distributed components which offer remote procedure calls have to
+ * extend the rpc endpoint base class.
+ *
+ * The main idea is that a rpc endpoint is backed by a rpc server which has a single thread
+ * processing the rpc calls. Thus, by executing all state changing operations within the main
+ * thread, we don't have to reason about concurrent accesses. The rpc provides provides
+ * {@link #runAsync(Runnable)}, {@link #callAsync(Callable, Timeout)} and the
+ * {@link #getMainThreadExecutionContext()} to execute code in the rpc server's main thread.
+ *
+ * @param <C> Rpc gateway counterpart for the implementing rpc endpoint
+ */
+public abstract class RpcEndpoint<C extends RpcGateway> {
+
+ protected final Logger log = LoggerFactory.getLogger(getClass());
+
+ /** Rpc service to be used to start the rpc server and to obtain rpc gateways */
+ private final RpcService rpcService;
+
+ /** Self gateway which can be used to schedule asynchronous calls on yourself */
+ private C self;
+
+ /**
+ * The main thread execution context to be used to execute future callbacks in the main thread
+ * of the executing rpc server.
+ *
+ * IMPORTANT: The main thread context is only available after the rpc server has been started.
+ */
+ private MainThreadExecutionContext mainThreadExecutionContext;
+
+ public RpcEndpoint(RpcService rpcService) {
+ this.rpcService = rpcService;
+ }
+
+ /**
+ * Get self-gateway which should be used to run asynchronous rpc calls on this endpoint.
+ *
+ * IMPORTANT: Always issue local method calls via the self-gateway if the current thread
+ * is not the main thread of the underlying rpc server, e.g. from within a future callback.
+ *
+ * @return Self gateway
+ */
+ public C getSelf() {
+ return self;
+ }
+
+ /**
+ * Execute the runnable in the main thread of the underlying rpc server.
+ *
+ * @param runnable Runnable to be executed in the main thread of the underlying rpc server
+ */
+ public void runAsync(Runnable runnable) {
+ ((MainThreadExecutor) self).runAsync(runnable);
+ }
+
+ /**
+ * Execute the callable in the main thread of the underlying rpc server returning a future for
+ * the result of the callable. If the callable is not completed within the given timeout, then
+ * the future will be failed with a {@link java.util.concurrent.TimeoutException}.
+ *
+ * @param callable Callable to be executed in the main thread of the underlying rpc server
+ * @param timeout Timeout for the callable to be completed
+ * @param <V> Return type of the callable
+ * @return Future for the result of the callable.
+ */
+ public <V> Future<V> callAsync(Callable<V> callable, Timeout timeout) {
+ return ((MainThreadExecutor) self).callAsync(callable, timeout);
+ }
+
+ /**
+ * Gets the main thread execution context. The main thread execution context can be used to
+ * execute tasks in the main thread of the underlying rpc server.
+ *
+ * @return Main thread execution context
+ */
+ public ExecutionContext getMainThreadExecutionContext() {
+ return mainThreadExecutionContext;
+ }
+
+ /**
+ * Gets the used rpc service.
+ *
+ * @return Rpc service
+ */
+ public RpcService getRpcService() {
+ return rpcService;
+ }
+
+ /**
+ * Starts the underlying rpc server via the rpc service and creates the main thread execution
+ * context. This makes the rpc endpoint effectively reachable from the outside.
+ *
+ * Can be overriden to add rpc endpoint specific start up code. Should always call the parent
+ * start method.
+ */
+ public void start() {
+ self = rpcService.startServer(this);
+ mainThreadExecutionContext = new MainThreadExecutionContext((MainThreadExecutor) self);
+ }
+
+
+ /**
+ * Shuts down the underlying rpc server via the rpc service.
+ *
+ * Can be overriden to add rpc endpoint specific shut down code. Should always call the parent
+ * shut down method.
+ */
+ public void shutDown() {
+ rpcService.stopServer(self);
+ }
+
+ /**
+ * Gets the address of the underlying rpc server. The address should be fully qualified so that
+ * a remote system can connect to this rpc server via this address.
+ *
+ * @return Fully qualified address of the underlying rpc server
+ */
+ public String getAddress() {
+ return rpcService.getAddress(self);
+ }
+
+ /**
+ * Execution context which executes runnables in the main thread context. A reported failure
+ * will cause the underlying rpc server to shut down.
+ */
+ private class MainThreadExecutionContext implements ExecutionContext {
+ private final MainThreadExecutor gateway;
+
+ MainThreadExecutionContext(MainThreadExecutor gateway) {
+ this.gateway = gateway;
+ }
+
+ @Override
+ public void execute(Runnable runnable) {
+ gateway.runAsync(runnable);
+ }
+
+ @Override
+ public void reportFailure(final Throwable t) {
+ gateway.runAsync(new Runnable() {
+ @Override
+ public void run() {
+ log.error("Encountered failure in the main thread execution context.", t);
+ shutDown();
+ }
+ });
+ }
+
+ @Override
+ public ExecutionContext prepare() {
+ return this;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ad6693ee/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcMethod.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcMethod.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcMethod.java
index 79e1d84..875e557 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcMethod.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcMethod.java
@@ -24,7 +24,7 @@ import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
- * Annotation for rpc method in a {@link RpcProtocol} implementation. Every rpc method must have a
+ * Annotation for rpc method in a {@link RpcEndpoint} implementation. Every rpc method must have a
* respective counterpart in the {@link RpcGateway} implementation for this rpc server. The
* RpcCompletenessTest makes sure that the set of rpc methods in a rpc server and the set of
* gateway methods in the corresponding gateway implementation are identical.
http://git-wip-us.apache.org/repos/asf/flink/blob/ad6693ee/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcProtocol.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcProtocol.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcProtocol.java
deleted file mode 100644
index 328246c..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcProtocol.java
+++ /dev/null
@@ -1,182 +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;
-
-import akka.util.Timeout;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import scala.concurrent.ExecutionContext;
-import scala.concurrent.Future;
-
-import java.util.concurrent.Callable;
-
-/**
- * Base class for rpc protocols. Distributed components which offer remote procedure calls have to
- * extend the rpc protocol base class.
- *
- * The main idea is that a rpc protocol is backed by a rpc server which has a single thread
- * processing the rpc calls. Thus, by executing all state changing operations within the main
- * thread, we don't have to reason about concurrent accesses. The rpc provides provides
- * {@link #runAsync(Runnable)}, {@link #callAsync(Callable, Timeout)} and the
- * {@link #getMainThreadExecutionContext()} to execute code in the rpc server's main thread.
- *
- * @param <C> Rpc gateway counterpart for the implementing rpc protocol
- */
-public abstract class RpcProtocol<C extends RpcGateway> {
-
- protected final Logger log = LoggerFactory.getLogger(getClass());
-
- /** Rpc service to be used to start the rpc server and to obtain rpc gateways */
- private final RpcService rpcService;
-
- /** Self gateway which can be used to schedule asynchronous calls on yourself */
- private C self;
-
- /**
- * The main thread execution context to be used to execute future callbacks in the main thread
- * of the executing rpc server.
- *
- * IMPORTANT: The main thread context is only available after the rpc server has been started.
- */
- private MainThreadExecutionContext mainThreadExecutionContext;
-
- public RpcProtocol(RpcService rpcService) {
- this.rpcService = rpcService;
- }
-
- /**
- * Get self-gateway which should be used to run asynchronous rpc calls on this protocol.
- *
- * IMPORTANT: Always issue local method calls via the self-gateway if the current thread
- * is not the main thread of the underlying rpc server, e.g. from within a future callback.
- *
- * @return Self gateway
- */
- public C getSelf() {
- return self;
- }
-
- /**
- * Execute the runnable in the main thread of the underlying rpc server.
- *
- * @param runnable Runnable to be executed in the main thread of the underlying rpc server
- */
- public void runAsync(Runnable runnable) {
- ((MainThreadExecutor) self).runAsync(runnable);
- }
-
- /**
- * Execute the callable in the main thread of the underlying rpc server returning a future for
- * the result of the callable. If the callable is not completed within the given timeout, then
- * the future will be failed with a {@link java.util.concurrent.TimeoutException}.
- *
- * @param callable Callable to be executed in the main thread of the underlying rpc server
- * @param timeout Timeout for the callable to be completed
- * @param <V> Return type of the callable
- * @return Future for the result of the callable.
- */
- public <V> Future<V> callAsync(Callable<V> callable, Timeout timeout) {
- return ((MainThreadExecutor) self).callAsync(callable, timeout);
- }
-
- /**
- * Gets the main thread execution context. The main thread execution context can be used to
- * execute tasks in the main thread of the underlying rpc server.
- *
- * @return Main thread execution context
- */
- public ExecutionContext getMainThreadExecutionContext() {
- return mainThreadExecutionContext;
- }
-
- /**
- * Gets the used rpc service.
- *
- * @return Rpc service
- */
- public RpcService getRpcService() {
- return rpcService;
- }
-
- /**
- * Starts the underlying rpc server via the rpc service and creates the main thread execution
- * context. This makes the rpc protocol effectively reachable from the outside.
- *
- * Can be overriden to add rpc protocol specific start up code. Should always call the parent
- * start method.
- */
- public void start() {
- self = rpcService.startServer(this);
- mainThreadExecutionContext = new MainThreadExecutionContext((MainThreadExecutor) self);
- }
-
-
- /**
- * Shuts down the underlying rpc server via the rpc service.
- *
- * Can be overriden to add rpc protocol specific shut down code. Should always call the parent
- * shut down method.
- */
- public void shutDown() {
- rpcService.stopServer(self);
- }
-
- /**
- * Gets the address of the underlying rpc server. The address should be fully qualified so that
- * a remote system can connect to this rpc server via this address.
- *
- * @return Fully qualified address of the underlying rpc server
- */
- public String getAddress() {
- return rpcService.getAddress(self);
- }
-
- /**
- * Execution context which executes runnables in the main thread context. A reported failure
- * will cause the underlying rpc server to shut down.
- */
- private class MainThreadExecutionContext implements ExecutionContext {
- private final MainThreadExecutor gateway;
-
- MainThreadExecutionContext(MainThreadExecutor gateway) {
- this.gateway = gateway;
- }
-
- @Override
- public void execute(Runnable runnable) {
- gateway.runAsync(runnable);
- }
-
- @Override
- public void reportFailure(final Throwable t) {
- gateway.runAsync(new Runnable() {
- @Override
- public void run() {
- log.error("Encountered failure in the main thread execution context.", t);
- shutDown();
- }
- });
- }
-
- @Override
- public ExecutionContext prepare() {
- return this;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/ad6693ee/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 fdc6100..90ff7b6 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
@@ -21,7 +21,7 @@ package org.apache.flink.runtime.rpc;
import scala.concurrent.Future;
/**
- * Interface for rpc services. An rpc service is used to start and connect to a {@link RpcProtocol}.
+ * Interface for rpc services. An rpc service is used to start and connect to a {@link RpcEndpoint}.
* Connecting to a rpc server will return a {@link RpcGateway} which can be used to call remote
* procedures.
*/
@@ -39,14 +39,14 @@ public interface RpcService {
<C extends RpcGateway> Future<C> connect(String address, Class<C> clazz);
/**
- * Start a rpc server which forwards the remote procedure calls to the provided rpc protocol.
+ * Start a rpc server which forwards the remote procedure calls to the provided rpc endpoint.
*
- * @param rpcProtocol Rpc protocl to dispath the rpcs to
- * @param <S> Type of the rpc protocol
+ * @param rpcEndpoint Rpc protocl to dispath the rpcs to
+ * @param <S> Type of the rpc endpoint
* @param <C> Type of the self rpc gateway associated with the rpc server
* @return Self gateway to dispatch remote procedure calls to oneself
*/
- <S extends RpcProtocol, C extends RpcGateway> C startServer(S rpcProtocol);
+ <S extends RpcEndpoint, C extends RpcGateway> C startServer(S rpcEndpoint);
/**
* Stop the underlying rpc server of the provided self gateway.
http://git-wip-us.apache.org/repos/asf/flink/blob/ad6693ee/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 98eed66..d55bd13 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
@@ -34,7 +34,7 @@ import org.apache.flink.runtime.rpc.jobmaster.JobMasterGateway;
import org.apache.flink.runtime.rpc.resourcemanager.ResourceManager;
import org.apache.flink.runtime.rpc.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.rpc.RpcGateway;
-import org.apache.flink.runtime.rpc.RpcProtocol;
+import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.akka.jobmaster.JobMasterAkkaActor;
import org.apache.flink.runtime.rpc.akka.jobmaster.JobMasterAkkaGateway;
@@ -85,29 +85,29 @@ public class AkkaRpcService implements RpcService {
}
@Override
- public <S extends RpcProtocol, C extends RpcGateway> C startServer(S rpcProtocol) {
+ public <S extends RpcEndpoint, C extends RpcGateway> C startServer(S rpcEndpoint) {
ActorRef ref;
C self;
- if (rpcProtocol instanceof TaskExecutor) {
+ if (rpcEndpoint instanceof TaskExecutor) {
ref = actorSystem.actorOf(
- Props.create(TaskExecutorAkkaActor.class, rpcProtocol)
+ Props.create(TaskExecutorAkkaActor.class, rpcEndpoint)
);
self = (C) new TaskExecutorAkkaGateway(ref, timeout);
- } else if (rpcProtocol instanceof ResourceManager) {
+ } else if (rpcEndpoint instanceof ResourceManager) {
ref = actorSystem.actorOf(
- Props.create(ResourceManagerAkkaActor.class, rpcProtocol)
+ Props.create(ResourceManagerAkkaActor.class, rpcEndpoint)
);
self = (C) new ResourceManagerAkkaGateway(ref, timeout);
- } else if (rpcProtocol instanceof JobMaster) {
+ } else if (rpcEndpoint instanceof JobMaster) {
ref = actorSystem.actorOf(
- Props.create(JobMasterAkkaActor.class, rpcProtocol)
+ Props.create(JobMasterAkkaActor.class, rpcEndpoint)
);
self = (C) new JobMasterAkkaGateway(ref, timeout);
} else {
- throw new RuntimeException("Could not start RPC server for class " + rpcProtocol.getClass());
+ throw new RuntimeException("Could not start RPC server for class " + rpcEndpoint.getClass());
}
actors.add(ref);
http://git-wip-us.apache.org/repos/asf/flink/blob/ad6693ee/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMaster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMaster.java
index 73f1947..b81b19c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMaster.java
@@ -27,7 +27,7 @@ import org.apache.flink.runtime.rpc.RpcMethod;
import org.apache.flink.runtime.rpc.resourcemanager.JobMasterRegistration;
import org.apache.flink.runtime.rpc.resourcemanager.RegistrationResponse;
import org.apache.flink.runtime.rpc.resourcemanager.ResourceManagerGateway;
-import org.apache.flink.runtime.rpc.RpcProtocol;
+import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.taskmanager.TaskExecutionState;
import scala.Tuple2;
@@ -56,7 +56,7 @@ import java.util.concurrent.TimeoutException;
* given task</li>
* </ul>
*/
-public class JobMaster extends RpcProtocol<JobMasterGateway> {
+public class JobMaster extends RpcEndpoint<JobMasterGateway> {
/** Execution context for future callbacks */
private final ExecutionContext executionContext;
http://git-wip-us.apache.org/repos/asf/flink/blob/ad6693ee/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManager.java
index da471fb..c7e8def 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManager.java
@@ -21,7 +21,7 @@ package org.apache.flink.runtime.rpc.resourcemanager;
import akka.dispatch.Mapper;
import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.rpc.RpcMethod;
-import org.apache.flink.runtime.rpc.RpcProtocol;
+import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.jobmaster.JobMaster;
import org.apache.flink.runtime.rpc.jobmaster.JobMasterGateway;
@@ -43,7 +43,7 @@ import java.util.concurrent.ExecutorService;
* <li>{@link #requestSlot(SlotRequest)} requests a slot from the resource manager</li>
* </ul>
*/
-public class ResourceManager extends RpcProtocol<ResourceManagerGateway> {
+public class ResourceManager extends RpcEndpoint<ResourceManagerGateway> {
private final ExecutionContext executionContext;
private final Map<JobMasterGateway, InstanceID> jobMasterGateways;
http://git-wip-us.apache.org/repos/asf/flink/blob/ad6693ee/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java
index 3849609..cdfc3bd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java
@@ -23,7 +23,7 @@ import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.rpc.RpcMethod;
-import org.apache.flink.runtime.rpc.RpcProtocol;
+import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;
import scala.concurrent.ExecutionContext;
@@ -41,7 +41,7 @@ import java.util.concurrent.ExecutorService;
* <li>{@link #cancelTask(ExecutionAttemptID)} cancels a given task identified by the {@link ExecutionAttemptID}</li>
* </ul>
*/
-public class TaskExecutor extends RpcProtocol<TaskExecutorGateway> {
+public class TaskExecutor extends RpcEndpoint<TaskExecutorGateway> {
private final ExecutionContext executionContext;
private final Set<ExecutionAttemptID> tasks = new HashSet<>();
http://git-wip-us.apache.org/repos/asf/flink/blob/ad6693ee/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcCompletenessTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcCompletenessTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcCompletenessTest.java
index 948ee85..0ded25e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcCompletenessTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcCompletenessTest.java
@@ -45,27 +45,27 @@ public class RpcCompletenessTest extends TestLogger {
public void testRpcCompleteness() {
Reflections reflections = new Reflections("org.apache.flink");
- Set<Class<? extends RpcProtocol>> classes = reflections.getSubTypesOf(RpcProtocol.class);
+ Set<Class<? extends RpcEndpoint>> classes = reflections.getSubTypesOf(RpcEndpoint.class);
- Class<? extends RpcProtocol> c;
+ Class<? extends RpcEndpoint> c;
- for (Class<? extends RpcProtocol> rpcProtocol :classes){
- c = rpcProtocol;
+ for (Class<? extends RpcEndpoint> rpcEndpoint :classes){
+ c = rpcEndpoint;
Type superClass = c.getGenericSuperclass();
Class<?> rpcGatewayType = extractTypeParameter(superClass, 0);
if (rpcGatewayType != null) {
- checkCompleteness(rpcProtocol, (Class<? extends RpcGateway>) rpcGatewayType);
+ checkCompleteness(rpcEndpoint, (Class<? extends RpcGateway>) rpcGatewayType);
} else {
- fail("Could not retrieve the rpc gateway class for the given rpc protocol class " + rpcProtocol.getName());
+ fail("Could not retrieve the rpc gateway class for the given rpc endpoint class " + rpcEndpoint.getName());
}
}
}
- private void checkCompleteness(Class<? extends RpcProtocol> rpcProtocol, Class<? extends RpcGateway> rpcGateway) {
+ private void checkCompleteness(Class<? extends RpcEndpoint> rpcEndpoint, Class<? extends RpcGateway> rpcGateway) {
Method[] gatewayMethods = rpcGateway.getDeclaredMethods();
- Method[] serverMethods = rpcProtocol.getDeclaredMethods();
+ Method[] serverMethods = rpcEndpoint.getDeclaredMethods();
Map<String, Set<Method>> rpcMethods = new HashMap<>();
Set<Method> unmatchedRpcMethods = new HashSet<>();
@@ -90,17 +90,17 @@ public class RpcCompletenessTest extends TestLogger {
for (Method gatewayMethod : gatewayMethods) {
assertTrue(
- "The rpc protocol " + rpcProtocol.getName() + " does not contain a RpcMethod " +
+ "The rpc endpoint " + rpcEndpoint.getName() + " does not contain a RpcMethod " +
"annotated method with the same name and signature " +
- generateProtocolMethodSignature(gatewayMethod) + ".",
+ generateEndpointMethodSignature(gatewayMethod) + ".",
rpcMethods.containsKey(gatewayMethod.getName()));
checkGatewayMethod(gatewayMethod);
- if (!matchGatewayMethodWithProtocol(gatewayMethod, rpcMethods.get(gatewayMethod.getName()), unmatchedRpcMethods)) {
- fail("Could not find a RpcMethod annotated method in rpc protocol " +
- rpcProtocol.getName() + " matching the rpc gateway method " +
- generateProtocolMethodSignature(gatewayMethod) + " defined in the rpc gateway " +
+ if (!matchGatewayMethodWithEndpoint(gatewayMethod, rpcMethods.get(gatewayMethod.getName()), unmatchedRpcMethods)) {
+ fail("Could not find a RpcMethod annotated method in rpc endpoint " +
+ rpcEndpoint.getName() + " matching the rpc gateway method " +
+ generateEndpointMethodSignature(gatewayMethod) + " defined in the rpc gateway " +
rpcGateway.getName() + ".");
}
}
@@ -112,7 +112,7 @@ public class RpcCompletenessTest extends TestLogger {
builder.append(unmatchedRpcMethod).append("\n");
}
- fail("The rpc protocol " + rpcProtocol.getName() + " contains rpc methods which " +
+ fail("The rpc endpoint " + rpcEndpoint.getName() + " contains rpc methods which " +
"are not matched to gateway methods of " + rpcGateway.getName() + ":\n" +
builder.toString());
}
@@ -153,17 +153,17 @@ public class RpcCompletenessTest extends TestLogger {
/**
* Checks whether we find a matching overloaded version for the gateway method among the methods
- * with the same name in the rpc protocol.
+ * with the same name in the rpc endpoint.
*
* @param gatewayMethod Gateway method
- * @param protocolMethods Set of rpc methods on the rpc server with the same name as the gateway
+ * @param endpointMethods Set of rpc methods on the rpc endpoint with the same name as the gateway
* method
- * @param unmatchedRpcMethods Set of unmatched rpc methods on the protocol side (so far)
+ * @param unmatchedRpcMethods Set of unmatched rpc methods on the endpoint side (so far)
*/
- private boolean matchGatewayMethodWithProtocol(Method gatewayMethod, Set<Method> protocolMethods, Set<Method> unmatchedRpcMethods) {
- for (Method protocolMethod : protocolMethods) {
- if (checkMethod(gatewayMethod, protocolMethod)) {
- unmatchedRpcMethods.remove(protocolMethod);
+ private boolean matchGatewayMethodWithEndpoint(Method gatewayMethod, Set<Method> endpointMethods, Set<Method> unmatchedRpcMethods) {
+ for (Method endpointMethod : endpointMethods) {
+ if (checkMethod(gatewayMethod, endpointMethod)) {
+ unmatchedRpcMethods.remove(endpointMethod);
return true;
}
}
@@ -171,11 +171,11 @@ public class RpcCompletenessTest extends TestLogger {
return false;
}
- private boolean checkMethod(Method gatewayMethod, Method protocolMethod) {
+ private boolean checkMethod(Method gatewayMethod, Method endpointMethod) {
Class<?>[] gatewayParameterTypes = gatewayMethod.getParameterTypes();
Annotation[][] gatewayParameterAnnotations = gatewayMethod.getParameterAnnotations();
- Class<?>[] protocolParameterTypes = protocolMethod.getParameterTypes();
+ Class<?>[] endpointParameterTypes = endpointMethod.getParameterTypes();
List<Class<?>> filteredGatewayParameterTypes = new ArrayList<>();
@@ -188,18 +188,18 @@ public class RpcCompletenessTest extends TestLogger {
}
}
- if (filteredGatewayParameterTypes.size() != protocolParameterTypes.length) {
+ if (filteredGatewayParameterTypes.size() != endpointParameterTypes.length) {
return false;
} else {
// check the parameter types
for (int i = 0; i < filteredGatewayParameterTypes.size(); i++) {
- if (!checkType(filteredGatewayParameterTypes.get(i), protocolParameterTypes[i])) {
+ if (!checkType(filteredGatewayParameterTypes.get(i), endpointParameterTypes[i])) {
return false;
}
}
// check the return types
- if (protocolMethod.getReturnType() == void.class) {
+ if (endpointMethod.getReturnType() == void.class) {
if (gatewayMethod.getReturnType() != void.class) {
return false;
}
@@ -213,22 +213,22 @@ public class RpcCompletenessTest extends TestLogger {
} else {
Class<?> valueClass = extractTypeParameter(futureClass, 0);
- if (protocolMethod.getReturnType().equals(futureClass)) {
- Class<?> rpcProtocolValueClass = extractTypeParameter(protocolMethod.getReturnType(), 0);
+ if (endpointMethod.getReturnType().equals(futureClass)) {
+ Class<?> rpcEndpointValueClass = extractTypeParameter(endpointMethod.getReturnType(), 0);
// check if we have the same future value types
- if (valueClass != null && rpcProtocolValueClass != null && !checkType(valueClass, rpcProtocolValueClass)) {
+ if (valueClass != null && rpcEndpointValueClass != null && !checkType(valueClass, rpcEndpointValueClass)) {
return false;
}
} else {
- if (valueClass != null && !checkType(valueClass, protocolMethod.getReturnType())) {
+ if (valueClass != null && !checkType(valueClass, endpointMethod.getReturnType())) {
return false;
}
}
}
}
- return gatewayMethod.getName().equals(protocolMethod.getName());
+ return gatewayMethod.getName().equals(endpointMethod.getName());
}
}
@@ -237,7 +237,7 @@ public class RpcCompletenessTest extends TestLogger {
}
/**
- * Generates from a gateway rpc method signature the corresponding rpc protocol signature.
+ * Generates from a gateway rpc method signature the corresponding rpc endpoint signature.
*
* For example the {@link RpcTimeout} annotation adds an additional parameter to the gateway
* signature which is not relevant on the server side.
@@ -245,7 +245,7 @@ public class RpcCompletenessTest extends TestLogger {
* @param method Method to generate the signature string for
* @return String of the respective server side rpc method signature
*/
- private String generateProtocolMethodSignature(Method method) {
+ private String generateEndpointMethodSignature(Method method) {
StringBuilder builder = new StringBuilder();
if (method.getReturnType().equals(Void.TYPE)) {