@Test
public void testFilterOutDuplicateJobMasterRegistrations() throws Exception {

	final long verificationTimeout = 500L;
	final Configuration configuration = new Configuration();
	final TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler();
	final JobLeaderService jobLeaderService = mock(JobLeaderService.class);
	final TaskManagerConfiguration taskManagerConfiguration = TaskManagerConfiguration.fromConfiguration(configuration);
	final TaskManagerLocation taskManagerLocation = new TaskManagerLocation(ResourceID.generate(), InetAddress.getLocalHost(), 1234);
	final HighAvailabilityServices haServicesMock = mock(HighAvailabilityServices.class, Mockito.RETURNS_MOCKS);
	final HeartbeatServices heartbeatServicesMock = mock(HeartbeatServices.class, Mockito.RETURNS_MOCKS);
	final JobID jobId = new JobID();
	final JobMasterGateway jobMasterGateway = mock(JobMasterGateway.class);
	when(jobMasterGateway.getHostname()).thenReturn("localhost");
	final JMTMRegistrationSuccess registrationMessage = new JMTMRegistrationSuccess(ResourceID.generate(), 1);
	final JobManagerTable jobManagerTableMock = spy(new JobManagerTable());
	final TaskExecutor taskExecutor = new TaskExecutor(
		rpc,
		taskManagerConfiguration,
		taskManagerLocation,
		mock(MemoryManager.class),
		mock(IOManager.class),
		mock(NetworkEnvironment.class),
		haServicesMock,
		heartbeatServicesMock,
		mock(TaskManagerMetricGroup.class),
		mock(BroadcastVariableManager.class),
		mock(FileCache.class),
		mock(TaskSlotTable.class),
		jobManagerTableMock,
		jobLeaderService,
		testingFatalErrorHandler);
	try {
		taskExecutor.start();
		ArgumentCaptor<JobLeaderListener> jobLeaderListenerArgumentCaptor = ArgumentCaptor.forClass(JobLeaderListener.class);
		verify(jobLeaderService).start(anyString(), any(RpcService.class), any(HighAvailabilityServices.class), jobLeaderListenerArgumentCaptor.capture());
		JobLeaderListener taskExecutorListener = jobLeaderListenerArgumentCaptor.getValue();
		taskExecutorListener.jobManagerGainedLeadership(jobId, jobMasterGateway, registrationMessage);
		// duplicate job manager gained leadership message
		taskExecutorListener.jobManagerGainedLeadership(jobId, jobMasterGateway, registrationMessage);
		ArgumentCaptor<JobManagerConnection> jobManagerConnectionArgumentCaptor = ArgumentCaptor.forClass(JobManagerConnection.class);
		verify(jobManagerTableMock, Mockito.timeout(verificationTimeout).times(1)).put(eq(jobId), jobManagerConnectionArgumentCaptor.capture());
		JobManagerConnection jobManagerConnection = jobManagerConnectionArgumentCaptor.getValue();
		assertEquals(jobMasterGateway, jobManagerConnection.getJobManagerGateway());
		testingFatalErrorHandler.rethrowError();
	} finally {
		taskExecutor.shutDown();
		taskExecutor.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
	}
}