diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutorToResourceManagerConnection.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutorToResourceManagerConnection.java index f398b7d06d6dc..7ccc8794761e8 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutorToResourceManagerConnection.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutorToResourceManagerConnection.java @@ -55,7 +55,7 @@ public class TaskExecutorToResourceManagerConnection { private final String resourceManagerAddress; - private ResourceManagerRegistration pendingRegistration; + private TaskExecutorToResourceManagerConnection.ResourceManagerRegistration pendingRegistration; private ResourceManagerGateway registeredResourceManager; @@ -86,13 +86,13 @@ public void start() { checkState(!closed, "The connection is already closed"); checkState(!isRegistered() && pendingRegistration == null, "The connection is already started"); - ResourceManagerRegistration registration = new ResourceManagerRegistration( + pendingRegistration = new TaskExecutorToResourceManagerConnection.ResourceManagerRegistration( log, taskExecutor.getRpcService(), resourceManagerAddress, resourceManagerLeaderId, taskExecutor.getAddress(), taskExecutor.getResourceID()); - registration.startRegistration(); + pendingRegistration.startRegistration(); - Future> future = registration.getFuture(); + Future> future = pendingRegistration.getFuture(); future.onSuccess(new OnSuccess>() { @Override @@ -167,14 +167,14 @@ public String toString() { // Utilities // ------------------------------------------------------------------------ - static class ResourceManagerRegistration + private static class ResourceManagerRegistration extends RetryingRegistration { private final String taskExecutorAddress; private final ResourceID resourceID; - public ResourceManagerRegistration( + ResourceManagerRegistration( Logger log, RpcService rpcService, String targetAddress,