@Test public void testHeartbeatTimeoutWithTaskExecutor() throws Exception { final ResourceID taskExecutorId = ResourceID.generate(); final CompletableFuture<ResourceID> heartbeatRequestFuture = new CompletableFuture<>(); final CompletableFuture<Exception> disconnectFuture = new CompletableFuture<>(); final TaskExecutorGateway taskExecutorGateway = new TestingTaskExecutorGatewayBuilder() .setDisconnectResourceManagerConsumer(disconnectFuture::complete) .setHeartbeatResourceManagerConsumer(heartbeatRequestFuture::complete) .createTestingTaskExecutorGateway(); rpcService.registerGateway(taskExecutorGateway.getAddress(), taskExecutorGateway); runHeartbeatTimeoutTest( resourceManagerGateway -> { registerTaskExecutor(resourceManagerGateway, taskExecutorId, taskExecutorGateway.getAddress()); }, resourceManagerResourceId -> { assertThat(heartbeatRequestFuture.get(), is(equalTo(resourceManagerResourceId))); assertThat(disconnectFuture.get(), instanceOf(TimeoutException.class)); } ); }
@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()); }
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()); } }
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()); } ResourceManager(
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			JobManagerMetricGroup jobManagerMetricGroup); }
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()); } ResourceManager(
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			JobManagerMetricGroup jobManagerMetricGroup); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<RegistrationResponse> registerJobManager(
			final JobMasterId jobMasterId,
			final ResourceID jobManagerResourceId,
			final String jobManagerAddress,
			final JobID jobId,
			final Time timeout); @Override CompletableFuture<RegistrationResponse> registerTaskExecutor(
			final String taskExecutorAddress,
			final ResourceID taskExecutorResourceId,
			final int dataPort,
			final HardwareDescription hardwareDescription,
			final Time timeout); @Override CompletableFuture<Acknowledge> sendSlotReport(ResourceID taskManagerResourceId, InstanceID taskManagerRegistrationId, SlotReport slotReport, Time timeout); @Override void heartbeatFromTaskManager(final ResourceID resourceID, final SlotReport slotReport); @Override void heartbeatFromJobManager(final ResourceID resourceID); @Override void disconnectTaskManager(final ResourceID resourceId, final Exception cause); @Override void disconnectJobManager(final JobID jobId, final Exception cause); @Override CompletableFuture<Acknowledge> requestSlot(
			JobMasterId jobMasterId,
			SlotRequest slotRequest,
			final Time timeout); @Override void cancelSlotRequest(AllocationID allocationID); @Override void notifySlotAvailable(
			final InstanceID instanceID,
			final SlotID slotId,
			final AllocationID allocationId); @Override void registerInfoMessageListener(final String address); @Override void unRegisterInfoMessageListener(final String address); @Override CompletableFuture<Acknowledge> deregisterApplication(
			final ApplicationStatus finalStatus,
			@Nullable final String diagnostics); @Override CompletableFuture<Integer> getNumberOfRegisteredTaskManagers(); @Override CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout); @Override CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(ResourceID resourceId, Time timeout); @Override CompletableFuture<ResourceOverview> requestResourceOverview(Time timeout); @Override CompletableFuture<Collection<Tuple2<ResourceID, String>>> requestTaskManagerMetricQueryServicePaths(Time timeout); @Override CompletableFuture<TransientBlobKey> requestTaskManagerFileUpload(ResourceID taskManagerId, FileType fileType, Time timeout); void sendInfoMessage(final String message); @Override void grantLeadership(final UUID newLeaderSessionID); @Override void revokeLeadership(); @Override void handleError(final Exception exception); @VisibleForTesting abstract Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); abstract boolean stopWorker(WorkerType worker); }
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()); } ResourceManager(
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			JobManagerMetricGroup jobManagerMetricGroup); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<RegistrationResponse> registerJobManager(
			final JobMasterId jobMasterId,
			final ResourceID jobManagerResourceId,
			final String jobManagerAddress,
			final JobID jobId,
			final Time timeout); @Override CompletableFuture<RegistrationResponse> registerTaskExecutor(
			final String taskExecutorAddress,
			final ResourceID taskExecutorResourceId,
			final int dataPort,
			final HardwareDescription hardwareDescription,
			final Time timeout); @Override CompletableFuture<Acknowledge> sendSlotReport(ResourceID taskManagerResourceId, InstanceID taskManagerRegistrationId, SlotReport slotReport, Time timeout); @Override void heartbeatFromTaskManager(final ResourceID resourceID, final SlotReport slotReport); @Override void heartbeatFromJobManager(final ResourceID resourceID); @Override void disconnectTaskManager(final ResourceID resourceId, final Exception cause); @Override void disconnectJobManager(final JobID jobId, final Exception cause); @Override CompletableFuture<Acknowledge> requestSlot(
			JobMasterId jobMasterId,
			SlotRequest slotRequest,
			final Time timeout); @Override void cancelSlotRequest(AllocationID allocationID); @Override void notifySlotAvailable(
			final InstanceID instanceID,
			final SlotID slotId,
			final AllocationID allocationId); @Override void registerInfoMessageListener(final String address); @Override void unRegisterInfoMessageListener(final String address); @Override CompletableFuture<Acknowledge> deregisterApplication(
			final ApplicationStatus finalStatus,
			@Nullable final String diagnostics); @Override CompletableFuture<Integer> getNumberOfRegisteredTaskManagers(); @Override CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout); @Override CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(ResourceID resourceId, Time timeout); @Override CompletableFuture<ResourceOverview> requestResourceOverview(Time timeout); @Override CompletableFuture<Collection<Tuple2<ResourceID, String>>> requestTaskManagerMetricQueryServicePaths(Time timeout); @Override CompletableFuture<TransientBlobKey> requestTaskManagerFileUpload(ResourceID taskManagerId, FileType fileType, Time timeout); void sendInfoMessage(final String message); @Override void grantLeadership(final UUID newLeaderSessionID); @Override void revokeLeadership(); @Override void handleError(final Exception exception); @VisibleForTesting abstract Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); abstract boolean stopWorker(WorkerType worker); static final String RESOURCE_MANAGER_NAME; }