@Test public void testIgnoringSlotRequestsIfNotRegistered() throws Exception { final TaskSlotTable taskSlotTable = new TaskSlotTable(Collections.singleton(ResourceProfile.UNKNOWN), timerService); final TaskManagerServices taskManagerServices = new TaskManagerServicesBuilder().setTaskSlotTable(taskSlotTable).build(); final TaskExecutor taskExecutor = createTaskExecutor(taskManagerServices); taskExecutor.start(); try { final TestingResourceManagerGateway testingResourceManagerGateway = new TestingResourceManagerGateway(); final CompletableFuture<RegistrationResponse> registrationFuture = new CompletableFuture<>(); final CompletableFuture<ResourceID> taskExecutorResourceIdFuture = new CompletableFuture<>(); testingResourceManagerGateway.setRegisterTaskExecutorFunction(stringResourceIDSlotReportIntegerHardwareDescriptionTuple5 -> { taskExecutorResourceIdFuture.complete(stringResourceIDSlotReportIntegerHardwareDescriptionTuple5.f1); return registrationFuture; }); rpc.registerGateway(testingResourceManagerGateway.getAddress(), testingResourceManagerGateway); resourceManagerLeaderRetriever.notifyListener(testingResourceManagerGateway.getAddress(), testingResourceManagerGateway.getFencingToken().toUUID()); final TaskExecutorGateway taskExecutorGateway = taskExecutor.getSelfGateway(TaskExecutorGateway.class); final ResourceID resourceId = taskExecutorResourceIdFuture.get(); final SlotID slotId = new SlotID(resourceId, 0); final CompletableFuture<Acknowledge> slotRequestResponse = taskExecutorGateway.requestSlot(slotId, jobId, new AllocationID(), "foobar", testingResourceManagerGateway.getFencingToken(), timeout); try { slotRequestResponse.get(); fail("We should not be able to request slots before the TaskExecutor is registered at the ResourceManager."); } catch (ExecutionException ee) { assertThat(ExceptionUtils.stripExecutionException(ee), instanceOf(TaskManagerException.class)); } } finally { RpcUtils.terminateRpcEndpoint(taskExecutor, timeout); } }
@Override public CompletableFuture<Acknowledge> requestSlot( final SlotID slotId, final JobID jobId, final AllocationID allocationId, final String targetAddress, final ResourceManagerId resourceManagerId, final Time timeout) { log.info("Receive slot request {} for job {} from resource manager with leader id {}.", allocationId, jobId, resourceManagerId); try { if (!isConnectedToResourceManager(resourceManagerId)) { final String message = String.format("TaskManager is not connected to the resource manager %s.", resourceManagerId); log.debug(message); throw new TaskManagerException(message); } if (taskSlotTable.isSlotFree(slotId.getSlotNumber())) { if (taskSlotTable.allocateSlot(slotId.getSlotNumber(), jobId, allocationId, taskManagerConfiguration.getTimeout())) { log.info("Allocated slot for {}.", allocationId); } else { log.info("Could not allocate slot for {}.", allocationId); throw new SlotAllocationException("Could not allocate slot."); } } else if (!taskSlotTable.isAllocated(slotId.getSlotNumber(), jobId, allocationId)) { final String message = "The slot " + slotId + " has already been allocated for a different job."; log.info(message); final AllocationID allocationID = taskSlotTable.getCurrentAllocation(slotId.getSlotNumber()); throw new SlotOccupiedException(message, allocationID, taskSlotTable.getOwningJob(allocationID)); } if (jobManagerTable.contains(jobId)) { offerSlotsToJobManager(jobId); } else { try { jobLeaderService.addJob(jobId, targetAddress); } catch (Exception e) { try { taskSlotTable.freeSlot(allocationId); } catch (SlotNotFoundException slotNotFoundException) { onFatalError(slotNotFoundException); } localStateStoresManager.releaseLocalStateForAllocationId(allocationId); if (!taskSlotTable.isSlotFree(slotId.getSlotNumber())) { onFatalError(new Exception("Could not free slot " + slotId)); } throw new SlotAllocationException("Could not add job to job leader service.", e); } } } catch (TaskManagerException taskManagerException) { return FutureUtils.completedExceptionally(taskManagerException); } return CompletableFuture.completedFuture(Acknowledge.get()); }
TaskExecutor extends RpcEndpoint implements TaskExecutorGateway { @Override public CompletableFuture<Acknowledge> requestSlot( final SlotID slotId, final JobID jobId, final AllocationID allocationId, final String targetAddress, final ResourceManagerId resourceManagerId, final Time timeout) { log.info("Receive slot request {} for job {} from resource manager with leader id {}.", allocationId, jobId, resourceManagerId); try { if (!isConnectedToResourceManager(resourceManagerId)) { final String message = String.format("TaskManager is not connected to the resource manager %s.", resourceManagerId); log.debug(message); throw new TaskManagerException(message); } if (taskSlotTable.isSlotFree(slotId.getSlotNumber())) { if (taskSlotTable.allocateSlot(slotId.getSlotNumber(), jobId, allocationId, taskManagerConfiguration.getTimeout())) { log.info("Allocated slot for {}.", allocationId); } else { log.info("Could not allocate slot for {}.", allocationId); throw new SlotAllocationException("Could not allocate slot."); } } else if (!taskSlotTable.isAllocated(slotId.getSlotNumber(), jobId, allocationId)) { final String message = "The slot " + slotId + " has already been allocated for a different job."; log.info(message); final AllocationID allocationID = taskSlotTable.getCurrentAllocation(slotId.getSlotNumber()); throw new SlotOccupiedException(message, allocationID, taskSlotTable.getOwningJob(allocationID)); } if (jobManagerTable.contains(jobId)) { offerSlotsToJobManager(jobId); } else { try { jobLeaderService.addJob(jobId, targetAddress); } catch (Exception e) { try { taskSlotTable.freeSlot(allocationId); } catch (SlotNotFoundException slotNotFoundException) { onFatalError(slotNotFoundException); } localStateStoresManager.releaseLocalStateForAllocationId(allocationId); if (!taskSlotTable.isSlotFree(slotId.getSlotNumber())) { onFatalError(new Exception("Could not free slot " + slotId)); } throw new SlotAllocationException("Could not add job to job leader service.", e); } } } catch (TaskManagerException taskManagerException) { return FutureUtils.completedExceptionally(taskManagerException); } return CompletableFuture.completedFuture(Acknowledge.get()); } }
TaskExecutor extends RpcEndpoint implements TaskExecutorGateway { @Override public CompletableFuture<Acknowledge> requestSlot( final SlotID slotId, final JobID jobId, final AllocationID allocationId, final String targetAddress, final ResourceManagerId resourceManagerId, final Time timeout) { log.info("Receive slot request {} for job {} from resource manager with leader id {}.", allocationId, jobId, resourceManagerId); try { if (!isConnectedToResourceManager(resourceManagerId)) { final String message = String.format("TaskManager is not connected to the resource manager %s.", resourceManagerId); log.debug(message); throw new TaskManagerException(message); } if (taskSlotTable.isSlotFree(slotId.getSlotNumber())) { if (taskSlotTable.allocateSlot(slotId.getSlotNumber(), jobId, allocationId, taskManagerConfiguration.getTimeout())) { log.info("Allocated slot for {}.", allocationId); } else { log.info("Could not allocate slot for {}.", allocationId); throw new SlotAllocationException("Could not allocate slot."); } } else if (!taskSlotTable.isAllocated(slotId.getSlotNumber(), jobId, allocationId)) { final String message = "The slot " + slotId + " has already been allocated for a different job."; log.info(message); final AllocationID allocationID = taskSlotTable.getCurrentAllocation(slotId.getSlotNumber()); throw new SlotOccupiedException(message, allocationID, taskSlotTable.getOwningJob(allocationID)); } if (jobManagerTable.contains(jobId)) { offerSlotsToJobManager(jobId); } else { try { jobLeaderService.addJob(jobId, targetAddress); } catch (Exception e) { try { taskSlotTable.freeSlot(allocationId); } catch (SlotNotFoundException slotNotFoundException) { onFatalError(slotNotFoundException); } localStateStoresManager.releaseLocalStateForAllocationId(allocationId); if (!taskSlotTable.isSlotFree(slotId.getSlotNumber())) { onFatalError(new Exception("Could not free slot " + slotId)); } throw new SlotAllocationException("Could not add job to job leader service.", e); } } } catch (TaskManagerException taskManagerException) { return FutureUtils.completedExceptionally(taskManagerException); } return CompletableFuture.completedFuture(Acknowledge.get()); } 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 { @Override public CompletableFuture<Acknowledge> requestSlot( final SlotID slotId, final JobID jobId, final AllocationID allocationId, final String targetAddress, final ResourceManagerId resourceManagerId, final Time timeout) { log.info("Receive slot request {} for job {} from resource manager with leader id {}.", allocationId, jobId, resourceManagerId); try { if (!isConnectedToResourceManager(resourceManagerId)) { final String message = String.format("TaskManager is not connected to the resource manager %s.", resourceManagerId); log.debug(message); throw new TaskManagerException(message); } if (taskSlotTable.isSlotFree(slotId.getSlotNumber())) { if (taskSlotTable.allocateSlot(slotId.getSlotNumber(), jobId, allocationId, taskManagerConfiguration.getTimeout())) { log.info("Allocated slot for {}.", allocationId); } else { log.info("Could not allocate slot for {}.", allocationId); throw new SlotAllocationException("Could not allocate slot."); } } else if (!taskSlotTable.isAllocated(slotId.getSlotNumber(), jobId, allocationId)) { final String message = "The slot " + slotId + " has already been allocated for a different job."; log.info(message); final AllocationID allocationID = taskSlotTable.getCurrentAllocation(slotId.getSlotNumber()); throw new SlotOccupiedException(message, allocationID, taskSlotTable.getOwningJob(allocationID)); } if (jobManagerTable.contains(jobId)) { offerSlotsToJobManager(jobId); } else { try { jobLeaderService.addJob(jobId, targetAddress); } catch (Exception e) { try { taskSlotTable.freeSlot(allocationId); } catch (SlotNotFoundException slotNotFoundException) { onFatalError(slotNotFoundException); } localStateStoresManager.releaseLocalStateForAllocationId(allocationId); if (!taskSlotTable.isSlotFree(slotId.getSlotNumber())) { onFatalError(new Exception("Could not free slot " + slotId)); } throw new SlotAllocationException("Could not add job to job leader service.", e); } } } catch (TaskManagerException taskManagerException) { return FutureUtils.completedExceptionally(taskManagerException); } return CompletableFuture.completedFuture(Acknowledge.get()); } 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 { @Override public CompletableFuture<Acknowledge> requestSlot( final SlotID slotId, final JobID jobId, final AllocationID allocationId, final String targetAddress, final ResourceManagerId resourceManagerId, final Time timeout) { log.info("Receive slot request {} for job {} from resource manager with leader id {}.", allocationId, jobId, resourceManagerId); try { if (!isConnectedToResourceManager(resourceManagerId)) { final String message = String.format("TaskManager is not connected to the resource manager %s.", resourceManagerId); log.debug(message); throw new TaskManagerException(message); } if (taskSlotTable.isSlotFree(slotId.getSlotNumber())) { if (taskSlotTable.allocateSlot(slotId.getSlotNumber(), jobId, allocationId, taskManagerConfiguration.getTimeout())) { log.info("Allocated slot for {}.", allocationId); } else { log.info("Could not allocate slot for {}.", allocationId); throw new SlotAllocationException("Could not allocate slot."); } } else if (!taskSlotTable.isAllocated(slotId.getSlotNumber(), jobId, allocationId)) { final String message = "The slot " + slotId + " has already been allocated for a different job."; log.info(message); final AllocationID allocationID = taskSlotTable.getCurrentAllocation(slotId.getSlotNumber()); throw new SlotOccupiedException(message, allocationID, taskSlotTable.getOwningJob(allocationID)); } if (jobManagerTable.contains(jobId)) { offerSlotsToJobManager(jobId); } else { try { jobLeaderService.addJob(jobId, targetAddress); } catch (Exception e) { try { taskSlotTable.freeSlot(allocationId); } catch (SlotNotFoundException slotNotFoundException) { onFatalError(slotNotFoundException); } localStateStoresManager.releaseLocalStateForAllocationId(allocationId); if (!taskSlotTable.isSlotFree(slotId.getSlotNumber())) { onFatalError(new Exception("Could not free slot " + slotId)); } throw new SlotAllocationException("Could not add job to job leader service.", e); } } } catch (TaskManagerException taskManagerException) { return FutureUtils.completedExceptionally(taskManagerException); } return CompletableFuture.completedFuture(Acknowledge.get()); } 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; }