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

	@Override
	public CompletableFuture<RegistrationResponse> registerTaskExecutor(
			final String taskExecutorAddress,
			final ResourceID taskExecutorResourceId,
			final int dataPort,
			final HardwareDescription hardwareDescription,
			final Time timeout) {

		CompletableFuture<TaskExecutorGateway> taskExecutorGatewayFuture = getRpcService().connect(taskExecutorAddress, TaskExecutorGateway.class);
		taskExecutorGatewayFutures.put(taskExecutorResourceId, taskExecutorGatewayFuture);

		return taskExecutorGatewayFuture.handleAsync(
			(TaskExecutorGateway taskExecutorGateway, Throwable throwable) -> {
				if (taskExecutorGatewayFuture == taskExecutorGatewayFutures.get(taskExecutorResourceId)) {
					taskExecutorGatewayFutures.remove(taskExecutorResourceId);
					if (throwable != null) {
						return new RegistrationResponse.Decline(throwable.getMessage());
					} else {
						return registerTaskExecutorInternal(
							taskExecutorGateway,
							taskExecutorAddress,
							taskExecutorResourceId,
							dataPort,
							hardwareDescription);
					}
				} else {
					log.info("Ignoring outdated TaskExecutorGateway connection.");
					return new RegistrationResponse.Decline("Decline outdated task executor registration.");
				}
			},
			getMainThreadExecutor());
	}

	public  ResourceManager(
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			JobManagerMetricGroup jobManagerMetricGroup);

}