@Test
public void testSlotAcceptance() throws Exception {

	final JobID jobId = new JobID();
	final Configuration configuration = new Configuration();
	final TaskManagerConfiguration taskManagerConfiguration = TaskManagerConfiguration.fromConfiguration(configuration);
	final ResourceID resourceId = new ResourceID("foobar");
	final TaskManagerLocation taskManagerLocation = new TaskManagerLocation(resourceId, InetAddress.getLoopbackAddress(), 1234);
	final TestingHighAvailabilityServices haServices = new TestingHighAvailabilityServices();
	final TimerService<AllocationID> timerService = mock(TimerService.class);
	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);
	final TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler();
	final String resourceManagerAddress = "rm";
	final UUID resourceManagerLeaderId = UUID.randomUUID();
	final ResourceID resourceManagerResourceId = new ResourceID(resourceManagerAddress);
	final String jobManagerAddress = "jm";
	final UUID jobManagerLeaderId = UUID.randomUUID();
	final LeaderRetrievalService resourceManagerLeaderRetrievalService = new TestingLeaderRetrievalService(resourceManagerAddress, resourceManagerLeaderId);
	final LeaderRetrievalService jobManagerLeaderRetrievalService = new TestingLeaderRetrievalService(jobManagerAddress, jobManagerLeaderId);
	haServices.setResourceManagerLeaderRetriever(resourceManagerLeaderRetrievalService);
	haServices.setJobMasterLeaderRetriever(jobId, jobManagerLeaderRetrievalService);
	final ResourceManagerGateway resourceManagerGateway = mock(ResourceManagerGateway.class);
	final InstanceID registrationId = new InstanceID();
	when(resourceManagerGateway.registerTaskExecutor(
		any(String.class),
		eq(resourceId),
		any(SlotReport.class),
		any(Time.class))).thenReturn(CompletableFuture.completedFuture(new TaskExecutorRegistrationSuccess(registrationId, resourceManagerResourceId, 1000L)));
	final ResourceID jmResourceId = new ResourceID(jobManagerAddress);
	final int blobPort = 42;
	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, blobPort)));
	when(jobMasterGateway.getHostname()).thenReturn(jobManagerAddress);
	when(jobMasterGateway.offerSlots(
			any(ResourceID.class), any(Iterable.class), any(Time.class)))
		.thenReturn(CompletableFuture.completedFuture((Collection<SlotOffer>)Collections.singleton(offer1)));
	rpc.registerGateway(resourceManagerAddress, resourceManagerGateway);
	rpc.registerGateway(jobManagerAddress, jobMasterGateway);
	TaskExecutor taskManager = new TaskExecutor(
		rpc,
		taskManagerConfiguration,
		taskManagerLocation,
		mock(MemoryManager.class),
		mock(IOManager.class),
		mock(NetworkEnvironment.class),
		haServices,
		mock(HeartbeatServices.class, RETURNS_MOCKS),
		mock(TaskManagerMetricGroup.class),
		mock(BroadcastVariableManager.class),
		mock(FileCache.class),
		taskSlotTable,
		jobManagerTable,
		jobLeaderService,
		testingFatalErrorHandler);
	try {
		taskManager.start();
		// wait for the registration at the ResourceManager
		verify(resourceManagerGateway, Mockito.timeout(timeout.toMilliseconds())).registerTaskExecutor(
			eq(taskManager.getAddress()),
			eq(resourceId),
			any(SlotReport.class),
			any(Time.class));
		taskSlotTable.allocateSlot(0, jobId, allocationId1, Time.milliseconds(10000L));
		taskSlotTable.allocateSlot(1, jobId, allocationId2, Time.milliseconds(10000L));
		// we have to add the job after the TaskExecutor, because otherwise the service has not
		// been properly started.
		jobLeaderService.addJob(jobId, jobManagerAddress);
		verify(resourceManagerGateway, Mockito.timeout(timeout.toMilliseconds())).notifySlotAvailable(
			eq(registrationId),
			eq(new SlotID(resourceId, 1)),
			eq(allocationId2));
		assertTrue(taskSlotTable.existsActiveSlot(jobId, allocationId1));
		assertFalse(taskSlotTable.existsActiveSlot(jobId, allocationId2));
		assertTrue(taskSlotTable.isSlotFree(1));
		// check if a concurrent error occurred
		testingFatalErrorHandler.rethrowError();
	} finally {
		taskManager.shutDown();
		taskManager.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
	}
}