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 the heartbeat interplay between the {@link HeartbeatManagerImpl} and the
	 * {@link HeartbeatManagerSenderImpl}. The sender should regularly trigger heartbeat requests
	 * which are fulfilled by the receiver. Upon stopping the receiver, the sender should notify
	 * the heartbeat listener about the heartbeat timeout.
	 *
	 * @throws Exception
	 */
Tests the heartbeat interplay between the  HeartbeatManagerImpl and the
HeartbeatManagerSenderImpl. The sender should regularly trigger heartbeat requests
which are fulfilled by the receiver. Upon stopping the receiver, the sender should notify
the heartbeat listener about the heartbeat timeout.

@throws Exception
Tests the heartbeat interplay between the HeartbeatManagerImpl and the HeartbeatManagerSenderImpl . The sender should regularly trigger heartbeat requests which are fulfilled by the receiver . Upon stopping the receiver the sender should notify the heartbeat listener about the heartbeat timeout . @throws Exception
@Test
	public void testHeartbeatCluster() throws Exception {
		long heartbeatTimeout = 100L;
		long heartbeatPeriod = 20L;
		Object object = new Object();
		Object object2 = new Object();
		ResourceID resourceID = new ResourceID("foobar");
		ResourceID resourceID2 = new ResourceID("barfoo");
		@SuppressWarnings("unchecked")
		HeartbeatListener<Object, Object> heartbeatListener = mock(HeartbeatListener.class);
		when(heartbeatListener.retrievePayload()).thenReturn(CompletableFuture.completedFuture(object));
		TestingHeartbeatListener heartbeatListener2 = new TestingHeartbeatListener(object2);
		CompletableFuture<ResourceID> futureTimeout = heartbeatListener2.getTimeoutFuture();
		HeartbeatManagerImpl<Object, Object> heartbeatManager = new HeartbeatManagerImpl<>(
			heartbeatTimeout,
			resourceID,
			heartbeatListener,
			new DirectExecutorService(),
			new ScheduledExecutorServiceAdapter(new ScheduledThreadPoolExecutor(1)),
			LOG);
		HeartbeatManagerSenderImpl<Object, Object> heartbeatManager2 = new HeartbeatManagerSenderImpl<>(
			heartbeatPeriod,
			heartbeatTimeout,
			resourceID2,
			heartbeatListener2,
			new DirectExecutorService(),
			new ScheduledExecutorServiceAdapter(new ScheduledThreadPoolExecutor(1)),
			LOG);
		heartbeatManager.monitorTarget(resourceID2, heartbeatManager2);
		heartbeatManager2.monitorTarget(resourceID, heartbeatManager);
		Thread.sleep(2 * heartbeatTimeout);
		assertFalse(futureTimeout.isDone());
		heartbeatManager.stop();
		ResourceID timeoutResourceID = futureTimeout.get(2 * heartbeatTimeout, TimeUnit.MILLISECONDS);
		assertEquals(resourceID, timeoutResourceID);
		int numberHeartbeats = (int) (2 * heartbeatTimeout / heartbeatPeriod);
		verify(heartbeatListener, atLeast(numberHeartbeats / 2)).reportPayload(resourceID2, object2);
		assertTrue(heartbeatListener2.getNumberHeartbeatReports() >= numberHeartbeats / 2);
	}
@ Test public void testHeartbeatCluster ( ) throws Exception { long heartbeatTimeout = 100L ; long heartbeatPeriod = 20L ; Object object = new Object ( ) ; Object object2 = new Object ( ) ; ResourceID resourceID = new ResourceID ( "foobar" ) ; ResourceID resourceID2 = new ResourceID ( "barfoo" ) ; @ SuppressWarnings ( "unchecked" ) HeartbeatListener < Object , Object > heartbeatListener = mock ( HeartbeatListener . class ) ; when ( heartbeatListener . retrievePayload ( ) ) . thenReturn ( CompletableFuture . completedFuture ( object ) ) ; TestingHeartbeatListener heartbeatListener2 = new TestingHeartbeatListener ( object2 ) ; CompletableFuture < ResourceID > futureTimeout = heartbeatListener2 . getTimeoutFuture ( ) ; HeartbeatManagerImpl < Object , Object > heartbeatManager = new HeartbeatManagerImpl < > ( heartbeatTimeout , resourceID , heartbeatListener , new DirectExecutorService ( ) , new ScheduledExecutorServiceAdapter ( new ScheduledThreadPoolExecutor ( 1 ) ) , LOG ) ; HeartbeatManagerSenderImpl < Object , Object > heartbeatManager2 = new HeartbeatManagerSenderImpl < > ( heartbeatPeriod , heartbeatTimeout , resourceID2 , heartbeatListener2 , new DirectExecutorService ( ) , new ScheduledExecutorServiceAdapter ( new ScheduledThreadPoolExecutor ( 1 ) ) , LOG ) ; heartbeatManager . monitorTarget ( resourceID2 , heartbeatManager2 ) ; heartbeatManager2 . monitorTarget ( resourceID , heartbeatManager ) ; Thread . sleep ( 2 * heartbeatTimeout ) ; assertFalse ( futureTimeout . isDone ( ) ) ; heartbeatManager . stop ( ) ; ResourceID timeoutResourceID = futureTimeout . get ( 2 * heartbeatTimeout , TimeUnit . MILLISECONDS ) ; assertEquals ( resourceID , timeoutResourceID ) ; int numberHeartbeats = ( int ) ( 2 * heartbeatTimeout / heartbeatPeriod ) ; verify ( heartbeatListener , atLeast ( numberHeartbeats / 2 ) ) . reportPayload ( resourceID2 , object2 ) ; assertTrue ( heartbeatListener2 . getNumberHeartbeatReports ( ) >= numberHeartbeats / 2 ) ; }
Tests the heartbeat interplay between the {@link HeartbeatManagerImpl} and the
 {@link HeartbeatManagerSenderImpl}.
Tests the heartbeat interplay between the { @link HeartbeatManagerImpl } and the { @link HeartbeatManagerSenderImpl } .