InstanceManager { public boolean reportHeartBeat(InstanceID instanceId) { 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(); LOG.trace("Received heartbeat from TaskManager {}", host); return true; } } InstanceManager(); void shutdown(); boolean reportHeartBeat(InstanceID instanceId); InstanceID registerTaskManager(
			TaskManagerGateway taskManagerGateway,
			TaskManagerLocation taskManagerLocation,
			HardwareDescription resources,
			int numberOfSlots); void unregisterTaskManager(InstanceID instanceId, boolean terminated); void unregisterAllTaskManagers(); boolean isRegistered(InstanceID instanceId); boolean isRegistered(ResourceID resourceId); int getNumberOfRegisteredTaskManagers(); int getTotalNumberOfSlots(); int getNumberOfAvailableSlots(); Collection<Instance> getAllRegisteredInstances(); Instance getRegisteredInstanceById(InstanceID instanceID); Instance getRegisteredInstance(ResourceID ref); void addInstanceListener(InstanceListener listener); void removeInstanceListener(InstanceListener listener);  }
@Test public void testReportHeartbeat() { try { InstanceManager cm = new InstanceManager(); final int dataPort = 20000; ResourceID resID1 = ResourceID.generate(); ResourceID resID2 = ResourceID.generate(); ResourceID resID3 = ResourceID.generate(); HardwareDescription hardwareDescription = HardwareDescription.extractFromSystem(4096); InetAddress address = InetAddress.getByName("127.0.0.1"); TaskManagerLocation ici1 = new TaskManagerLocation(resID1, address, dataPort); TaskManagerLocation ici2 = new TaskManagerLocation(resID2, address, dataPort + 1); TaskManagerLocation ici3 = new TaskManagerLocation(resID3, address, dataPort + 2); JavaTestKit probe1 = new JavaTestKit(system); JavaTestKit probe2 = new JavaTestKit(system); JavaTestKit probe3 = new JavaTestKit(system); InstanceID instanceID1 = cm.registerTaskManager( new ActorTaskManagerGateway(new AkkaActorGateway(probe1.getRef(), leaderSessionID)), ici1, hardwareDescription, 1); InstanceID instanceID2 = cm.registerTaskManager( new ActorTaskManagerGateway(new AkkaActorGateway(probe2.getRef(), leaderSessionID)), ici2, hardwareDescription, 1); InstanceID instanceID3 = cm.registerTaskManager( new ActorTaskManagerGateway(new AkkaActorGateway(probe3.getRef(), leaderSessionID)), ici3, hardwareDescription, 1); assertTrue(cm.reportHeartBeat(instanceID1)); assertTrue(cm.reportHeartBeat(instanceID2)); assertTrue(cm.reportHeartBeat(instanceID3)); assertFalse(cm.reportHeartBeat(new InstanceID())); 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())); 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()); } }