Rpc In Flink

Sep 09, 2019

Apache Flink uses Akka for communication between its Dispatcher, JobMaster, ResourceManager, and TaskExecutor components.

Translated from Chinese with AI · Read the original

Apache Flink uses Akka for communication between its Dispatcher, JobMaster, ResourceManager, and TaskExecutor components.

Dispatcher, JobMaster, ResourceManager, and TaskExecutor all extend the RpcEndpoint abstract class. Let’s look at the RpcEndpoint constructor.

1
/**
2
 * Initializes the RPC endpoint.
3
 *
4
 * @param rpcService The RPC server that dispatches calls to this RPC endpoint.
5
 * @param endpointId Unique identifier for this endpoint
6
 */
7
protected RpcEndpoint(final RpcService rpcService, final String endpointId) {
8
  this.rpcService = checkNotNull(rpcService, "rpcService");
9
  this.endpointId = checkNotNull(endpointId, "endpointId");
10

11
  this.rpcServer = rpcService.startServer(this);
12

13
  this.mainThreadExecutor = new MainThreadExecutor(rpcServer, this::validateRunsInMainThread);
14
}

One of the parameters is a RpcService instance, which is used to start or connect to a RpcEndpoint. For example, TaskExecutor uses its rpcService instance to start the Actor corresponding to its own TaskExecutor. To connect to JobMaster, it calls the connect method on rpcService to obtain the jobMasterGateway associated with JobMaster, allowing it to invoke JobMaster methods remotely through the gateway.

The RpcServer initialized in the constructor is a wrapper around the Actor, containing properties such as address and host.

How does the gateway invoke JobMaster methods remotely?

First, let’s look at how the gateway itself is generated. Starting from the connect method, we reach the end of connectInternal:

1
private <C extends RpcGateway> CompletableFuture<C> connectInternal(
2
    final String address,
3
    final Class<C> clazz,
4
    Function<ActorRef, InvocationHandler> invocationHandlerFactory) {
5
  checkState(!stopped, "RpcService is stopped");
6

7
  ........
8

9
  return actorRefFuture.thenCombineAsync(
10
    handshakeFuture,
11
    (ActorRef actorRef, HandshakeSuccessMessage ignored) -> {
12
      InvocationHandler invocationHandler = invocationHandlerFactory.apply(actorRef);
13

14
      // Rather than using the System ClassLoader directly, we derive the ClassLoader
15
      // from this class . That works better in cases where Flink runs embedded and all Flink
16
      // code is loaded dynamically (for example from an OSGI bundle) through a custom ClassLoader
17
      ClassLoader classLoader = getClass().getClassLoader();
18

19
      @SuppressWarnings("unchecked")
20
      C proxy = (C) Proxy.newProxyInstance(
21
        classLoader,
22
        new Class<?>[]{clazz},
23
        invocationHandler);
24

25
      return proxy;
26
    },
27
    actorSystem.dispatcher());
28
}

This uses the proxy pattern, with AkkaInvocationHandler as the invocation handler. Its invokeRpc method wraps the method being called and its arguments in a RpcInvocation (that is, a Message), then uses Akka’s own mechanisms to deliver it to the Actor corresponding to JobMaster. After receiving the message, JobMaster’s Actor invokes the corresponding method through reflection.