flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
/**
	 * Handle rpc invocations by looking up the rpc method on the rpc endpoint and calling this
	 * method with the provided method arguments. If the method has a return value, it is returned
	 * to the sender of the call.
	 *
	 * @param rpcInvocation Rpc invocation message
	 */
Handle rpc invocations by looking up the rpc method on the rpc endpoint and calling this
method with the provided method arguments. If the method has a return value, it is returned
to the sender of the call.

@param rpcInvocation Rpc invocation message
Handle rpc invocations by looking up the rpc method on the rpc endpoint and calling this method with the provided method arguments . If the method has a return value it is returned to the sender of the call . @param rpcInvocation Rpc invocation message
private void handleRpcInvocation(RpcInvocation rpcInvocation) {
		Method rpcMethod = null;
		try {
			String methodName = rpcInvocation.getMethodName();
			Class<?>[] parameterTypes = rpcInvocation.getParameterTypes();
			rpcMethod = lookupRpcMethod(methodName, parameterTypes);
		} catch (ClassNotFoundException e) {
			log.error("Could not load method arguments.", e);
			RpcConnectionException rpcException = new RpcConnectionException("Could not load method arguments.", e);
			getSender().tell(new Status.Failure(rpcException), getSelf());
		} catch (IOException e) {
			log.error("Could not deserialize rpc invocation message.", e);
			RpcConnectionException rpcException = new RpcConnectionException("Could not deserialize rpc invocation message.", e);
			getSender().tell(new Status.Failure(rpcException), getSelf());
		} catch (final NoSuchMethodException e) {
			log.error("Could not find rpc method for rpc invocation.", e);
			RpcConnectionException rpcException = new RpcConnectionException("Could not find rpc method for rpc invocation.", e);
			getSender().tell(new Status.Failure(rpcException), getSelf());
		}
		if (rpcMethod != null) {
			try {
				rpcMethod.setAccessible(true);
				if (rpcMethod.getReturnType().equals(Void.TYPE)) {
					rpcMethod.invoke(rpcEndpoint, rpcInvocation.getArgs());
				}
				else {
					final Object result;
					try {
						result = rpcMethod.invoke(rpcEndpoint, rpcInvocation.getArgs());
					}
					catch (InvocationTargetException e) {
						log.trace("Reporting back error thrown in remote procedure {}", rpcMethod, e);
						getSender().tell(new Status.Failure(e.getTargetException()), getSelf());
						return;
					}
					if (result instanceof CompletableFuture) {
						final CompletableFuture<?> future = (CompletableFuture<?>) result;
						Promise.DefaultPromise<Object> promise = new Promise.DefaultPromise<>();
						future.whenComplete(
							(value, throwable) -> {
								if (throwable != null) {
									promise.failure(throwable);
								} else {
									promise.success(value);
								}
							});
						Patterns.pipe(promise.future(), getContext().dispatcher()).to(getSender());
					} else {
						getSender().tell(new Status.Success(result), getSelf());
					}
				}
			} catch (Throwable e) {
				log.error("Error while executing remote procedure call {}.", rpcMethod, e);
				getSender().tell(new Status.Failure(e), getSelf());
			}
		}
	}
private void handleRpcInvocation ( RpcInvocation rpcInvocation ) { Method rpcMethod = null ; try { String methodName = rpcInvocation . getMethodName ( ) ; Class < ? > [ ] parameterTypes = rpcInvocation . getParameterTypes ( ) ; rpcMethod = lookupRpcMethod ( methodName , parameterTypes ) ; } catch ( ClassNotFoundException e ) { log . error ( "Could not load method arguments." , e ) ; RpcConnectionException rpcException = new RpcConnectionException ( "Could not load method arguments." , e ) ; getSender ( ) . tell ( new Status . Failure ( rpcException ) , getSelf ( ) ) ; } catch ( IOException e ) { log . error ( "Could not deserialize rpc invocation message." , e ) ; RpcConnectionException rpcException = new RpcConnectionException ( "Could not deserialize rpc invocation message." , e ) ; getSender ( ) . tell ( new Status . Failure ( rpcException ) , getSelf ( ) ) ; } catch ( final NoSuchMethodException e ) { log . error ( "Could not find rpc method for rpc invocation." , e ) ; RpcConnectionException rpcException = new RpcConnectionException ( "Could not find rpc method for rpc invocation." , e ) ; getSender ( ) . tell ( new Status . Failure ( rpcException ) , getSelf ( ) ) ; } if ( rpcMethod != null ) { try { rpcMethod . setAccessible ( true ) ; if ( rpcMethod . getReturnType ( ) . equals ( Void . TYPE ) ) { rpcMethod . invoke ( rpcEndpoint , rpcInvocation . getArgs ( ) ) ; } else { final Object result ; try { result = rpcMethod . invoke ( rpcEndpoint , rpcInvocation . getArgs ( ) ) ; } catch ( InvocationTargetException e ) { log . trace ( "Reporting back error thrown in remote procedure {}" , rpcMethod , e ) ; getSender ( ) . tell ( new Status . Failure ( e . getTargetException ( ) ) , getSelf ( ) ) ; return ; } if ( result instanceof CompletableFuture ) { final CompletableFuture < ? > future = ( CompletableFuture < ? > ) result ; Promise . DefaultPromise < Object > promise = new Promise . DefaultPromise < > ( ) ; future . whenComplete ( ( value , throwable ) -> { if ( throwable != null ) { promise . failure ( throwable ) ; } else { promise . success ( value ) ; } } ) ; Patterns . pipe ( promise . future ( ) , getContext ( ) . dispatcher ( ) ) . to ( getSender ( ) ) ; } else { getSender ( ) . tell ( new Status . Success ( result ) , getSelf ( ) ) ; } } } catch ( Throwable e ) { log . error ( "Error while executing remote procedure call {}." , rpcMethod , e ) ; getSender ( ) . tell ( new Status . Failure ( e ) , getSelf ( ) ) ; } } }
Handle rpc invocations by looking up the rpc method on the rpc endpoint and calling this
 method with the provided method arguments.
Handle rpc invocations by looking up the rpc method on the rpc endpoint and calling this method with the provided method arguments .
// this supports declaration of anonymous classes // No return value to send back // tell the sender about the failure // tell the sender the result of the computation // tell the sender about the failure