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 a heartbeat timeout is signaled if the heartbeat is not reported in time.
	 *
	 * @throws Exception
	 */
Tests that a heartbeat timeout is signaled if the heartbeat is not reported in time.
@throws Exception
Tests that a heartbeat timeout is signaled if the heartbeat is not reported in time . @throws Exception
@Test
	public void testHeartbeatTimeout() throws Exception {
		long heartbeatTimeout = 100L;
		int numHeartbeats = 10;
		long heartbeatInterval = 20L;
		Object payload = new Object();
		ResourceID ownResourceID = new ResourceID("foobar");
		ResourceID targetResourceID = new ResourceID("barfoo");
		TestingHeartbeatListener heartbeatListener = new TestingHeartbeatListener(payload);
		ScheduledExecutorService scheduledExecutorService = mock(ScheduledExecutorService.class);
		ScheduledFuture<?> scheduledFuture = mock(ScheduledFuture.class);
		doReturn(scheduledFuture).when(scheduledExecutorService).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class));
		Object expectedObject = new Object();
		HeartbeatManagerImpl<Object, Object> heartbeatManager = new HeartbeatManagerImpl<>(
			heartbeatTimeout,
			ownResourceID,
			heartbeatListener,
			new DirectExecutorService(),
			new ScheduledExecutorServiceAdapter(new ScheduledThreadPoolExecutor(1)),
			LOG);
		@SuppressWarnings("unchecked")
		HeartbeatTarget<Object> heartbeatTarget = mock(HeartbeatTarget.class);
		CompletableFuture<ResourceID> timeoutFuture = heartbeatListener.getTimeoutFuture();
		heartbeatManager.monitorTarget(targetResourceID, heartbeatTarget);
		for (int i = 0; i < numHeartbeats; i++) {
			heartbeatManager.receiveHeartbeat(targetResourceID, expectedObject);
			Thread.sleep(heartbeatInterval);
		}
		assertFalse(timeoutFuture.isDone());
		ResourceID timeoutResourceID = timeoutFuture.get(2 * heartbeatTimeout, TimeUnit.MILLISECONDS);
		assertEquals(targetResourceID, timeoutResourceID);
	}
@ Test public void testHeartbeatTimeout ( ) throws Exception { long heartbeatTimeout = 100L ; int numHeartbeats = 10 ; long heartbeatInterval = 20L ; Object payload = new Object ( ) ; ResourceID ownResourceID = new ResourceID ( "foobar" ) ; ResourceID targetResourceID = new ResourceID ( "barfoo" ) ; TestingHeartbeatListener heartbeatListener = new TestingHeartbeatListener ( payload ) ; ScheduledExecutorService scheduledExecutorService = mock ( ScheduledExecutorService . class ) ; ScheduledFuture < ? > scheduledFuture = mock ( ScheduledFuture . class ) ; doReturn ( scheduledFuture ) . when ( scheduledExecutorService ) . schedule ( any ( Runnable . class ) , anyLong ( ) , any ( TimeUnit . class ) ) ; Object expectedObject = new Object ( ) ; HeartbeatManagerImpl < Object , Object > heartbeatManager = new HeartbeatManagerImpl < > ( heartbeatTimeout , ownResourceID , heartbeatListener , new DirectExecutorService ( ) , new ScheduledExecutorServiceAdapter ( new ScheduledThreadPoolExecutor ( 1 ) ) , LOG ) ; @ SuppressWarnings ( "unchecked" ) HeartbeatTarget < Object > heartbeatTarget = mock ( HeartbeatTarget . class ) ; CompletableFuture < ResourceID > timeoutFuture = heartbeatListener . getTimeoutFuture ( ) ; heartbeatManager . monitorTarget ( targetResourceID , heartbeatTarget ) ; for ( int i = 0 ; i < numHeartbeats ; i ++ ) { heartbeatManager . receiveHeartbeat ( targetResourceID , expectedObject ) ; Thread . sleep ( heartbeatInterval ) ; } assertFalse ( timeoutFuture . isDone ( ) ) ; ResourceID timeoutResourceID = timeoutFuture . get ( 2 * heartbeatTimeout , TimeUnit . MILLISECONDS ) ; assertEquals ( targetResourceID , timeoutResourceID ) ; }
Tests that a heartbeat timeout is signaled if the heartbeat is not reported in time.
Tests that a heartbeat timeout is signaled if the heartbeat is not reported in time .