InstanceManager { public boolean reportHeartBeat(InstanceID instanceId, byte[] lastMetricsReport) { if (instanceId == null) { throw new IllegalArgumentException("InstanceID may not be null."); } synchronized (this.lock) { if (this.isShutdown) { return false; } Instance host = registeredHostsById.get(instanceId); if (host == null){ if (LOG.isDebugEnabled()) { LOG.debug("Received heartbeat from unknown TaskManager with instance ID " + instanceId.toString() + " Possibly TaskManager was marked as dead (timed-out) earlier. " + "Reporting back that task manager is no longer known."); } return false; } host.reportHeartBeat(); host.setMetricsReport(lastMetricsReport); if (LOG.isDebugEnabled()) { LOG.debug("Received heartbeat from TaskManager " + host); } return true; } } InstanceManager(); void shutdown(); boolean reportHeartBeat(InstanceID instanceId, byte[] lastMetricsReport); InstanceID registerTaskManager(
			ActorRef taskManager,
			InstanceConnectionInfo connectionInfo,
			HardwareDescription resources,
			int numberOfSlots,
			UUID leaderSessionID); void unregisterTaskManager(ActorRef taskManager, boolean terminated); void unregisterAllTaskManagers(); boolean isRegistered(ActorRef taskManager); int getNumberOfRegisteredTaskManagers(); int getTotalNumberOfSlots(); int getNumberOfAvailableSlots(); Collection<Instance> getAllRegisteredInstances(); Instance getRegisteredInstanceById(InstanceID instanceID); Instance getRegisteredInstance(ActorRef ref); void addInstanceListener(InstanceListener listener); void removeInstanceListener(InstanceListener listener);  }
@Test public void testReportHeartbeat() { try { InstanceManager cm = new InstanceManager(); final int dataPort = 20000; HardwareDescription hardwareDescription = HardwareDescription.extractFromSystem(4096); InetAddress address = InetAddress.getByName("127.0.0.1"); InstanceConnectionInfo ici1 = new InstanceConnectionInfo(address, dataPort); InstanceConnectionInfo ici2 = new InstanceConnectionInfo(address, dataPort + 1); InstanceConnectionInfo ici3 = new InstanceConnectionInfo(address, dataPort + 2); JavaTestKit probe1 = new JavaTestKit(system); JavaTestKit probe2 = new JavaTestKit(system); JavaTestKit probe3 = new JavaTestKit(system); InstanceID i1 = cm.registerTaskManager(probe1.getRef(), ici1, hardwareDescription, 1, leaderSessionID); InstanceID i2 = cm.registerTaskManager(probe2.getRef(), ici2, hardwareDescription, 1, leaderSessionID); InstanceID i3 = cm.registerTaskManager(probe3.getRef(), ici3, hardwareDescription, 1, leaderSessionID); assertTrue(cm.reportHeartBeat(i1, new byte[] {})); assertTrue(cm.reportHeartBeat(i2, new byte[] {})); assertTrue(cm.reportHeartBeat(i3, new byte[] {})); assertFalse(cm.reportHeartBeat(new InstanceID(), new byte[] {})); final long WAIT = 200; CommonTestUtils.sleepUninterruptibly(WAIT); Iterator<Instance> it = cm.getAllRegisteredInstances().iterator(); Instance instance1 = it.next(); long h1 = instance1.getLastHeartBeat(); long h2 = it.next().getLastHeartBeat(); long h3 = it.next().getLastHeartBeat(); assertTrue(cm.reportHeartBeat(instance1.getId(), new byte[] {})); long newH1 = instance1.getLastHeartBeat(); long now = System.currentTimeMillis(); assertTrue(now - h1 >= WAIT); assertTrue(now - h2 >= WAIT); assertTrue(now - h3 >= WAIT); assertTrue(now - newH1 <= WAIT); cm.shutdown(); } catch (Exception e) { System.err.println(e.getMessage()); e.printStackTrace(); Assert.fail("Test erroneous: " + e.getMessage()); } }