flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/heartbeat/HeartbeatManagerTest.java
/**
	 * Tests that the heartbeat monitors are updated when receiving a new heartbeat signal.
	 */
Tests that the heartbeat monitors are updated when receiving a new heartbeat signal.
Tests that the heartbeat monitors are updated when receiving a new heartbeat signal .
@Test
	public void testHeartbeatMonitorUpdate() {
		long heartbeatTimeout = 1000L;
		ResourceID ownResourceID = new ResourceID("foobar");
		ResourceID targetResourceID = new ResourceID("barfoo");
		@SuppressWarnings("unchecked")
		HeartbeatListener<Object, Object> heartbeatListener = mock(HeartbeatListener.class);
		ScheduledExecutor scheduledExecutor = mock(ScheduledExecutor.class);
		ScheduledFuture<?> scheduledFuture = mock(ScheduledFuture.class);
		doReturn(scheduledFuture).when(scheduledExecutor).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class));
		Object expectedObject = new Object();
		when(heartbeatListener.retrievePayload()).thenReturn(CompletableFuture.completedFuture(expectedObject));
		HeartbeatManagerImpl<Object, Object> heartbeatManager = new HeartbeatManagerImpl<>(
			heartbeatTimeout,
			ownResourceID,
			heartbeatListener,
			new DirectExecutorService(),
			scheduledExecutor,
			LOG);
		@SuppressWarnings("unchecked")
		HeartbeatTarget<Object> heartbeatTarget = mock(HeartbeatTarget.class);
		heartbeatManager.monitorTarget(targetResourceID, heartbeatTarget);
		heartbeatManager.receiveHeartbeat(targetResourceID, expectedObject);
		verify(scheduledFuture, times(1)).cancel(true);
		verify(scheduledExecutor, times(2)).schedule(any(Runnable.class), eq(heartbeatTimeout), eq(TimeUnit.MILLISECONDS));
	}
@ Test public void testHeartbeatMonitorUpdate ( ) { long heartbeatTimeout = 1000L ; ResourceID ownResourceID = new ResourceID ( "foobar" ) ; ResourceID targetResourceID = new ResourceID ( "barfoo" ) ; @ SuppressWarnings ( "unchecked" ) HeartbeatListener < Object , Object > heartbeatListener = mock ( HeartbeatListener . class ) ; ScheduledExecutor scheduledExecutor = mock ( ScheduledExecutor . class ) ; ScheduledFuture < ? > scheduledFuture = mock ( ScheduledFuture . class ) ; doReturn ( scheduledFuture ) . when ( scheduledExecutor ) . schedule ( any ( Runnable . class ) , anyLong ( ) , any ( TimeUnit . class ) ) ; Object expectedObject = new Object ( ) ; when ( heartbeatListener . retrievePayload ( ) ) . thenReturn ( CompletableFuture . completedFuture ( expectedObject ) ) ; HeartbeatManagerImpl < Object , Object > heartbeatManager = new HeartbeatManagerImpl < > ( heartbeatTimeout , ownResourceID , heartbeatListener , new DirectExecutorService ( ) , scheduledExecutor , LOG ) ; @ SuppressWarnings ( "unchecked" ) HeartbeatTarget < Object > heartbeatTarget = mock ( HeartbeatTarget . class ) ; heartbeatManager . monitorTarget ( targetResourceID , heartbeatTarget ) ; heartbeatManager . receiveHeartbeat ( targetResourceID , expectedObject ) ; verify ( scheduledFuture , times ( 1 ) ) . cancel ( true ) ; verify ( scheduledExecutor , times ( 2 ) ) . schedule ( any ( Runnable . class ) , eq ( heartbeatTimeout ) , eq ( TimeUnit . MILLISECONDS ) ) ; }
Tests that the heartbeat monitors are updated when receiving a new heartbeat signal.
Tests that the heartbeat monitors are updated when receiving a new heartbeat signal .