@Test public void testSlotAcceptance() throws Exception { final TaskSlotTable taskSlotTable = new TaskSlotTable(Arrays.asList(mock(ResourceProfile.class), mock(ResourceProfile.class)), timerService); final JobManagerTable jobManagerTable = new JobManagerTable(); final JobLeaderService jobLeaderService = new JobLeaderService(taskManagerLocation, RetryingRegistrationConfiguration.defaultConfiguration()); final String resourceManagerAddress = "rm"; final UUID resourceManagerLeaderId = UUID.randomUUID(); final String jobManagerAddress = "jm"; final UUID jobManagerLeaderId = UUID.randomUUID(); resourceManagerLeaderRetriever.notifyListener(resourceManagerAddress, resourceManagerLeaderId); jobManagerLeaderRetriever.notifyListener(jobManagerAddress, jobManagerLeaderId); final TestingResourceManagerGateway resourceManagerGateway = new TestingResourceManagerGateway(); final ResourceID resourceManagerResourceId = resourceManagerGateway.getOwnResourceId(); final InstanceID registrationId = new InstanceID(); final CompletableFuture<ResourceID> registrationFuture = new CompletableFuture<>(); resourceManagerGateway.setRegisterTaskExecutorFunction( stringResourceIDIntegerHardwareDescriptionTuple4 -> { registrationFuture.complete(stringResourceIDIntegerHardwareDescriptionTuple4.f1); return CompletableFuture.completedFuture(new TaskExecutorRegistrationSuccess(registrationId, resourceManagerResourceId, new ClusterInformation("localhost", 1234))); }); final CompletableFuture<Tuple3<InstanceID, SlotID, AllocationID>> availableSlotFuture = new CompletableFuture<>(); resourceManagerGateway.setNotifySlotAvailableConsumer(availableSlotFuture::complete); final ResourceID jmResourceId = new ResourceID(jobManagerAddress); final AllocationID allocationId1 = new AllocationID(); final AllocationID allocationId2 = new AllocationID(); final SlotOffer offer1 = new SlotOffer(allocationId1, 0, ResourceProfile.UNKNOWN); final JobMasterGateway jobMasterGateway = mock(JobMasterGateway.class); when(jobMasterGateway.registerTaskManager( any(String.class), eq(taskManagerLocation), any(Time.class) )).thenReturn(CompletableFuture.completedFuture(new JMTMRegistrationSuccess(jmResourceId))); when(jobMasterGateway.getHostname()).thenReturn(jobManagerAddress); when(jobMasterGateway.offerSlots( any(ResourceID.class), any(Collection.class), any(Time.class))) .thenReturn(CompletableFuture.completedFuture((Collection<SlotOffer>) Collections.singleton(offer1))); rpc.registerGateway(resourceManagerAddress, resourceManagerGateway); rpc.registerGateway(jobManagerAddress, jobMasterGateway); final TaskExecutorLocalStateStoresManager localStateStoresManager = createTaskExecutorLocalStateStoresManager(); final TaskManagerServices taskManagerServices = new TaskManagerServicesBuilder() .setTaskManagerLocation(taskManagerLocation) .setTaskSlotTable(taskSlotTable) .setJobManagerTable(jobManagerTable) .setJobLeaderService(jobLeaderService) .setTaskStateManager(localStateStoresManager) .build(); TaskExecutor taskManager = new TaskExecutor( rpc, taskManagerConfiguration, haServices, taskManagerServices, HEARTBEAT_SERVICES, UnregisteredMetricGroups.createUnregisteredTaskManagerMetricGroup(), null, dummyBlobCacheService, testingFatalErrorHandler); try { taskManager.start(); assertThat(registrationFuture.get(), equalTo(taskManagerLocation.getResourceID())); taskSlotTable.allocateSlot(0, jobId, allocationId1, Time.milliseconds(10000L)); taskSlotTable.allocateSlot(1, jobId, allocationId2, Time.milliseconds(10000L)); jobLeaderService.addJob(jobId, jobManagerAddress); final Tuple3<InstanceID, SlotID, AllocationID> instanceIDSlotIDAllocationIDTuple3 = availableSlotFuture.get(); final Tuple3<InstanceID, SlotID, AllocationID> expectedResult = Tuple3.of(registrationId, new SlotID(taskManagerLocation.getResourceID(), 1), allocationId2); assertThat(instanceIDSlotIDAllocationIDTuple3, equalTo(expectedResult)); assertTrue(taskSlotTable.tryMarkSlotActive(jobId, allocationId1)); assertFalse(taskSlotTable.tryMarkSlotActive(jobId, allocationId2)); assertTrue(taskSlotTable.isSlotFree(1)); } 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 metricQueryServicePath,
			BlobCacheService blobCacheService,
			FatalErrorHandler fatalErrorHandler); }
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 metricQueryServicePath,
			BlobCacheService blobCacheService,
			FatalErrorHandler fatalErrorHandler); @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> stopTask(ExecutionAttemptID executionAttemptID, Time timeout); @Override CompletableFuture<Acknowledge> updatePartitions(
			final ExecutionAttemptID executionAttemptID,
			Iterable<PartitionInfo> partitionInfos,
			Time timeout); @Override void failPartition(ExecutionAttemptID executionAttemptID); @Override void heartbeatFromJobManager(ResourceID resourceID); @Override void heartbeatFromResourceManager(ResourceID resourceID); @Override CompletableFuture<Acknowledge> triggerCheckpoint(
			ExecutionAttemptID executionAttemptID,
			long checkpointId,
			long checkpointTimestamp,
			CheckpointOptions checkpointOptions); @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 metricQueryServicePath,
			BlobCacheService blobCacheService,
			FatalErrorHandler fatalErrorHandler); @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> stopTask(ExecutionAttemptID executionAttemptID, Time timeout); @Override CompletableFuture<Acknowledge> updatePartitions(
			final ExecutionAttemptID executionAttemptID,
			Iterable<PartitionInfo> partitionInfos,
			Time timeout); @Override void failPartition(ExecutionAttemptID executionAttemptID); @Override void heartbeatFromJobManager(ResourceID resourceID); @Override void heartbeatFromResourceManager(ResourceID resourceID); @Override CompletableFuture<Acknowledge> triggerCheckpoint(
			ExecutionAttemptID executionAttemptID,
			long checkpointId,
			long checkpointTimestamp,
			CheckpointOptions checkpointOptions); @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; }