@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();
		// slot from unregistered resource
		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);
		// we'll also accept non requested slots
		assertTrue(slotPoolGateway.offerSlot(notRequested).get());
		AllocatedSlot allocatedSlot = createAllocatedSlot(resourceID, slotRequest.getAllocationId(), jobId, DEFAULT_TESTING_PROFILE);
		// accepted slot
		assertTrue(slotPoolGateway.offerSlot(allocatedSlot).get());
		SimpleSlot slot = future.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
		assertTrue(slot.isAlive());
		// duplicated offer with using slot
		assertTrue(slotPoolGateway.offerSlot(allocatedSlot).get());
		assertTrue(slot.isAlive());
		// duplicated offer with free slot
		slot.releaseSlot();
		assertTrue(slotPoolGateway.offerSlot(allocatedSlot).get());
	} finally {
		slotPool.shutDown();
	}
}