SlotPool extends RpcEndpoint implements SlotPoolGateway { @Override public CompletableFuture<Boolean> offerSlot(final AllocatedSlot slot) { validateRunsInMainThread(); final ResourceID resourceID = slot.getTaskManagerId(); final AllocationID allocationID = slot.getSlotAllocationId(); if (!registeredTaskManagers.contains(resourceID)) { LOG.debug("Received outdated slot offering [{}] from unregistered TaskManager: {}", slot.getSlotAllocationId(), slot); return CompletableFuture.completedFuture(false); } if (allocatedSlots.contains(allocationID) || availableSlots.contains(allocationID)) { LOG.debug("Received repeated offer for slot [{}]. Ignoring.", allocationID); return CompletableFuture.completedFuture(true); } PendingRequest pendingRequest = pendingRequests.remove(allocationID); if (pendingRequest != null) { SimpleSlot resultSlot = createSimpleSlot(slot, Locality.UNKNOWN); pendingRequest.getFuture().complete(resultSlot); allocatedSlots.add(resultSlot); } else { availableSlots.add(slot, clock.relativeTimeMillis()); } return CompletableFuture.completedFuture(true); } SlotPool(RpcService rpcService, JobID jobId);  SlotPool(
			RpcService rpcService,
			JobID jobId,
			Clock clock,
			Time slotRequestTimeout,
			Time resourceManagerAllocationTimeout,
			Time resourceManagerRequestTimeout); @Override void start(); void start(JobMasterId jobMasterId, String newJobManagerAddress); @Override void suspend(); SlotOwner getSlotOwner(); SlotProvider getSlotProvider(); @Override void connectToResourceManager(ResourceManagerGateway resourceManagerGateway); @Override void disconnectResourceManager(); @Override CompletableFuture<SimpleSlot> allocateSlot(
			ScheduledUnit task,
			ResourceProfile resources,
			Iterable<TaskManagerLocation> locationPreferences,
			Time timeout); @Override void returnAllocatedSlot(Slot slot); @Override CompletableFuture<Collection<SlotOffer>> offerSlots(Collection<Tuple2<AllocatedSlot, SlotOffer>> offers); @Override CompletableFuture<Boolean> offerSlot(final AllocatedSlot slot); @Override void failAllocation(final AllocationID allocationID, final Exception cause); @Override void registerTaskManager(final ResourceID resourceID); @Override CompletableFuture<Acknowledge> releaseTaskManager(final ResourceID resourceID);  }
@Test public void testOfferSlot() throws Exception { ResourceManagerGateway resourceManagerGateway = createResourceManagerGatewayMock(); final SlotPool slotPool = new SlotPool(rpcService, jobId); try { SlotPoolGateway slotPoolGateway = setupSlotPool(slotPool, resourceManagerGateway); ResourceID resourceID = new ResourceID("resource"); slotPoolGateway.registerTaskManager(resourceID); CompletableFuture<SimpleSlot> future = slotPoolGateway.allocateSlot(mock(ScheduledUnit.class), DEFAULT_TESTING_PROFILE, null, timeout); assertFalse(future.isDone()); ArgumentCaptor<SlotRequest> slotRequestArgumentCaptor = ArgumentCaptor.forClass(SlotRequest.class); verify(resourceManagerGateway, Mockito.timeout(timeout.toMilliseconds())).requestSlot(any(JobMasterId.class), slotRequestArgumentCaptor.capture(), any(Time.class)); final SlotRequest slotRequest = slotRequestArgumentCaptor.getValue(); AllocatedSlot invalid = createAllocatedSlot(new ResourceID("unregistered"), slotRequest.getAllocationId(), jobId, DEFAULT_TESTING_PROFILE); assertFalse(slotPoolGateway.offerSlot(invalid).get()); AllocatedSlot notRequested = createAllocatedSlot(resourceID, new AllocationID(), jobId, DEFAULT_TESTING_PROFILE); assertTrue(slotPoolGateway.offerSlot(notRequested).get()); AllocatedSlot allocatedSlot = createAllocatedSlot(resourceID, slotRequest.getAllocationId(), jobId, DEFAULT_TESTING_PROFILE); assertTrue(slotPoolGateway.offerSlot(allocatedSlot).get()); SimpleSlot slot = future.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS); assertTrue(slot.isAlive()); assertTrue(slotPoolGateway.offerSlot(allocatedSlot).get()); assertTrue(slot.isAlive()); slot.releaseSlot(); assertTrue(slotPoolGateway.offerSlot(allocatedSlot).get()); } finally { slotPool.shutDown(); } }