class ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender {

	@Override
	public CompletableFuture<RegistrationResponse> registerTaskExecutor(
			final TaskExecutorRegistration taskExecutorRegistration,
			final Time timeout) {

		CompletableFuture<TaskExecutorGateway> taskExecutorGatewayFuture = getRpcService().connect(taskExecutorRegistration.getTaskExecutorAddress(), TaskExecutorGateway.class);
		taskExecutorGatewayFutures.put(taskExecutorRegistration.getResourceId(), taskExecutorGatewayFuture);

		return taskExecutorGatewayFuture.handleAsync(
			(TaskExecutorGateway taskExecutorGateway, Throwable throwable) -> {
				final ResourceID resourceId = taskExecutorRegistration.getResourceId();
				if (taskExecutorGatewayFuture == taskExecutorGatewayFutures.get(resourceId)) {
					taskExecutorGatewayFutures.remove(resourceId);
					if (throwable != null) {
						return new RegistrationResponse.Decline(throwable.getMessage());
					} else {
						return registerTaskExecutorInternal(taskExecutorGateway, taskExecutorRegistration);
					}
				} else {
					log.debug("Ignoring outdated TaskExecutorGateway connection for {}.", resourceId.getStringWithMetadata());
					return new RegistrationResponse.Decline("Decline outdated task executor registration.");
				}
			},
			getMainThreadExecutor());
	}

	public  ResourceManager(
			RpcService rpcService,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			ResourceManagerPartitionTrackerFactory clusterPartitionTrackerFactory,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			ResourceManagerMetricGroup resourceManagerMetricGroup,
			Time rpcTimeout,
			Executor ioExecutor);

}