@Test public void testHeartbeatTimeoutWithResourceManager() throws Exception { final String rmAddress = "rm"; final ResourceID rmResourceId = new ResourceID(rmAddress); final long heartbeatInterval = 1L; final long heartbeatTimeout = 3L; final ResourceManagerId rmLeaderId = ResourceManagerId.generate(); TestingResourceManagerGateway rmGateway = new TestingResourceManagerGateway( rmLeaderId, rmResourceId, rmAddress, rmAddress); final TaskExecutorRegistrationSuccess registrationResponse = new TaskExecutorRegistrationSuccess( new InstanceID(), rmResourceId, new ClusterInformation("localhost", 1234)); final CompletableFuture<ResourceID> taskExecutorRegistrationFuture = new CompletableFuture<>(); final CountDownLatch registrationAttempts = new CountDownLatch(2); rmGateway.setRegisterTaskExecutorFunction( registration -> { taskExecutorRegistrationFuture.complete(registration.f1); registrationAttempts.countDown(); return CompletableFuture.completedFuture(registrationResponse); }); final CompletableFuture<ResourceID> taskExecutorDisconnectFuture = new CompletableFuture<>(); rmGateway.setDisconnectTaskExecutorConsumer( disconnectInfo -> taskExecutorDisconnectFuture.complete(disconnectInfo.f0)); rpc.registerGateway(rmAddress, rmGateway); final TaskSlotTable taskSlotTable = mock(TaskSlotTable.class); final SlotReport slotReport = new SlotReport(); when(taskSlotTable.createSlotReport(any(ResourceID.class))).thenReturn(slotReport); HeartbeatServices heartbeatServices = new HeartbeatServices(heartbeatInterval, heartbeatTimeout); final TaskExecutorLocalStateStoresManager localStateStoresManager = createTaskExecutorLocalStateStoresManager(); final TaskManagerServices taskManagerServices = new TaskManagerServicesBuilder() .setTaskManagerLocation(taskManagerLocation) .setTaskSlotTable(taskSlotTable) .setTaskStateManager(localStateStoresManager) .build(); final TaskExecutor taskManager = createTaskExecutor(taskManagerServices, heartbeatServices); try { taskManager.start(); resourceManagerLeaderRetriever.notifyListener(rmAddress, rmLeaderId.toUUID()); assertThat(taskExecutorRegistrationFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS), equalTo(taskManagerLocation.getResourceID())); assertThat(taskExecutorDisconnectFuture.get(heartbeatTimeout * 50L, TimeUnit.MILLISECONDS), equalTo(taskManagerLocation.getResourceID())); assertTrue( "The TaskExecutor should try to reconnect to the RM", registrationAttempts.await(timeout.toMilliseconds(), TimeUnit.SECONDS)); } finally { RpcUtils.terminateRpcEndpoint(taskManager, timeout); } }
public ResourceID getResourceID() { return taskManagerLocation.getResourceID(); }
TaskExecutor extends RpcEndpoint implements TaskExecutorGateway { public ResourceID getResourceID() { return taskManagerLocation.getResourceID(); } }
TaskExecutor extends RpcEndpoint implements TaskExecutorGateway { public ResourceID getResourceID() { return taskManagerLocation.getResourceID(); } TaskExecutor(
			RpcService rpcService,
			TaskManagerConfiguration taskManagerConfiguration,
			HighAvailabilityServices haServices,
			TaskManagerServices taskExecutorServices,
			HeartbeatServices heartbeatServices,
			TaskManagerMetricGroup taskManagerMetricGroup,
			@Nullable String metricQueryServiceAddress,
			BlobCacheService blobCacheService,
			FatalErrorHandler fatalErrorHandler,
			PartitionTable<JobID> partitionTable); }
TaskExecutor extends RpcEndpoint implements TaskExecutorGateway { public ResourceID getResourceID() { return taskManagerLocation.getResourceID(); } TaskExecutor(
			RpcService rpcService,
			TaskManagerConfiguration taskManagerConfiguration,
			HighAvailabilityServices haServices,
			TaskManagerServices taskExecutorServices,
			HeartbeatServices heartbeatServices,
			TaskManagerMetricGroup taskManagerMetricGroup,
			@Nullable String metricQueryServiceAddress,
			BlobCacheService blobCacheService,
			FatalErrorHandler fatalErrorHandler,
			PartitionTable<JobID> partitionTable); @Override CompletableFuture<Boolean> canBeReleased(); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<StackTraceSampleResponse> requestStackTraceSample(
			final ExecutionAttemptID executionAttemptId,
			final int sampleId,
			final int numSamples,
			final Time delayBetweenSamples,
			final int maxStackTraceDepth,
			final Time timeout); @Override CompletableFuture<Acknowledge> submitTask(
			TaskDeploymentDescriptor tdd,
			JobMasterId jobMasterId,
			Time timeout); @Override CompletableFuture<Acknowledge> cancelTask(ExecutionAttemptID executionAttemptID, Time timeout); @Override CompletableFuture<Acknowledge> updatePartitions(
			final ExecutionAttemptID executionAttemptID,
			Iterable<PartitionInfo> partitionInfos,
			Time timeout); @Override void releasePartitions(JobID jobId, Collection<ResultPartitionID> partitionIds); @Override void heartbeatFromJobManager(ResourceID resourceID, AllocatedSlotReport allocatedSlotReport); @Override void heartbeatFromResourceManager(ResourceID resourceID); @Override CompletableFuture<Acknowledge> triggerCheckpoint(
			ExecutionAttemptID executionAttemptID,
			long checkpointId,
			long checkpointTimestamp,
			CheckpointOptions checkpointOptions,
			boolean advanceToEndOfEventTime); @Override CompletableFuture<Acknowledge> confirmCheckpoint(
			ExecutionAttemptID executionAttemptID,
			long checkpointId,
			long checkpointTimestamp); @Override CompletableFuture<Acknowledge> requestSlot(
		final SlotID slotId,
		final JobID jobId,
		final AllocationID allocationId,
		final String targetAddress,
		final ResourceManagerId resourceManagerId,
		final Time timeout); @Override CompletableFuture<Acknowledge> freeSlot(AllocationID allocationId, Throwable cause, Time timeout); @Override CompletableFuture<TransientBlobKey> requestFileUpload(FileType fileType, Time timeout); @Override CompletableFuture<SerializableOptional<String>> requestMetricQueryServiceAddress(Time timeout); @Override void disconnectJobManager(JobID jobId, Exception cause); @Override void disconnectResourceManager(Exception cause); ResourceID getResourceID(); }
TaskExecutor extends RpcEndpoint implements TaskExecutorGateway { public ResourceID getResourceID() { return taskManagerLocation.getResourceID(); } TaskExecutor(
			RpcService rpcService,
			TaskManagerConfiguration taskManagerConfiguration,
			HighAvailabilityServices haServices,
			TaskManagerServices taskExecutorServices,
			HeartbeatServices heartbeatServices,
			TaskManagerMetricGroup taskManagerMetricGroup,
			@Nullable String metricQueryServiceAddress,
			BlobCacheService blobCacheService,
			FatalErrorHandler fatalErrorHandler,
			PartitionTable<JobID> partitionTable); @Override CompletableFuture<Boolean> canBeReleased(); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<StackTraceSampleResponse> requestStackTraceSample(
			final ExecutionAttemptID executionAttemptId,
			final int sampleId,
			final int numSamples,
			final Time delayBetweenSamples,
			final int maxStackTraceDepth,
			final Time timeout); @Override CompletableFuture<Acknowledge> submitTask(
			TaskDeploymentDescriptor tdd,
			JobMasterId jobMasterId,
			Time timeout); @Override CompletableFuture<Acknowledge> cancelTask(ExecutionAttemptID executionAttemptID, Time timeout); @Override CompletableFuture<Acknowledge> updatePartitions(
			final ExecutionAttemptID executionAttemptID,
			Iterable<PartitionInfo> partitionInfos,
			Time timeout); @Override void releasePartitions(JobID jobId, Collection<ResultPartitionID> partitionIds); @Override void heartbeatFromJobManager(ResourceID resourceID, AllocatedSlotReport allocatedSlotReport); @Override void heartbeatFromResourceManager(ResourceID resourceID); @Override CompletableFuture<Acknowledge> triggerCheckpoint(
			ExecutionAttemptID executionAttemptID,
			long checkpointId,
			long checkpointTimestamp,
			CheckpointOptions checkpointOptions,
			boolean advanceToEndOfEventTime); @Override CompletableFuture<Acknowledge> confirmCheckpoint(
			ExecutionAttemptID executionAttemptID,
			long checkpointId,
			long checkpointTimestamp); @Override CompletableFuture<Acknowledge> requestSlot(
		final SlotID slotId,
		final JobID jobId,
		final AllocationID allocationId,
		final String targetAddress,
		final ResourceManagerId resourceManagerId,
		final Time timeout); @Override CompletableFuture<Acknowledge> freeSlot(AllocationID allocationId, Throwable cause, Time timeout); @Override CompletableFuture<TransientBlobKey> requestFileUpload(FileType fileType, Time timeout); @Override CompletableFuture<SerializableOptional<String>> requestMetricQueryServiceAddress(Time timeout); @Override void disconnectJobManager(JobID jobId, Exception cause); @Override void disconnectResourceManager(Exception cause); ResourceID getResourceID(); static final String TASK_MANAGER_NAME; }