HeartBeatHandler { public HeartBeatResponse handleHeartBeat(HeartBeat heartbeat) throws AmbariException { long now = System.currentTimeMillis(); if (heartbeat.getAgentEnv() != null && heartbeat.getAgentEnv().getHostHealth() != null) { heartbeat.getAgentEnv().getHostHealth().setServerTimeStampAtReporting(now); } String hostname = heartbeat.getHostname(); Long currentResponseId = hostResponseIds.get(hostname); HeartBeatResponse response; if (currentResponseId == null) { LOG.error("CurrentResponseId unknown for " + hostname + " - send register command"); return createRegisterCommand(); } LOG.debug("Received heartbeat from host" + ", hostname=" + hostname + ", currentResponseId=" + currentResponseId + ", receivedResponseId=" + heartbeat.getResponseId()); if (heartbeat.getResponseId() == currentResponseId - 1) { LOG.warn("Old responseId received - response was lost - returning cached response"); return hostResponses.get(hostname); } else if (heartbeat.getResponseId() != currentResponseId) { LOG.error("Error in responseId sequence - sending agent restart command"); return createRestartCommand(currentResponseId); } response = new HeartBeatResponse(); response.setResponseId(++currentResponseId); Host hostObject; try { hostObject = clusterFsm.getHost(hostname); } catch (HostNotFoundException e) { LOG.error("Host: {} not found. Agent is still heartbeating.", hostname); if (LOG.isDebugEnabled()) { LOG.debug("Host associated with the agent heratbeat might have been " + "deleted", e); } return response; } if (hostObject.getState().equals(HostState.HEARTBEAT_LOST)) { LOG.warn("Host is in HEARTBEAT_LOST state - sending register command"); return createRegisterCommand(); } hostResponseIds.put(hostname, currentResponseId); hostResponses.put(hostname, response); if (heartbeat.componentStatus.size() > 0 && hostObject.getState().equals(HostState.WAITING_FOR_HOST_STATUS_UPDATES)) { try { LOG.debug("Got component status updates"); hostObject.handleEvent(new HostStatusUpdatesReceivedEvent(hostname, now)); } catch (InvalidStateTransitionException e) { LOG.warn("Failed to notify the host about component status updates", e); } } if (heartbeat.getRecoveryReport() != null) { RecoveryReport rr = heartbeat.getRecoveryReport(); processRecoveryReport(rr, hostname); } try { if (heartbeat.getNodeStatus().getStatus().equals(HostStatus.Status.HEALTHY)) { hostObject.handleEvent(new HostHealthyHeartbeatEvent(hostname, now, heartbeat.getAgentEnv(), heartbeat.getMounts())); } else { hostObject.handleEvent(new HostUnhealthyHeartbeatEvent(hostname, now, null)); } } catch (InvalidStateTransitionException ex) { LOG.warn("Asking agent to re-register due to " + ex.getMessage(), ex); hostObject.setState(HostState.INIT); return createRegisterCommand(); } Set<Cluster> clusters = clusterFsm.getClustersForHost(hostname); if (clusters.size() > 0) { String clusterName = clusters.iterator().next().getClusterName(); if (recoveryConfigHelper.isConfigStale(clusterName, hostname, heartbeat.getRecoveryTimestamp())) { RecoveryConfig rc = recoveryConfigHelper.getRecoveryConfig(clusterName, hostname); response.setRecoveryConfig(rc); if (response.getRecoveryConfig() != null) { LOG.info("Recovery configuration set to {}", response.getRecoveryConfig().toString()); } } } heartbeatProcessor.addHeartbeat(heartbeat); if (hostObject.getState().equals(HostState.HEALTHY)) { sendCommands(hostname, response); annotateResponse(hostname, response); } return response; } @Inject  HeartBeatHandler(Clusters fsm, ActionQueue aq, ActionManager am,
                          Injector injector); void start(); void setHeartbeatProcessor(HeartbeatProcessor heartbeatProcessor); HeartbeatProcessor getHeartbeatProcessor(); HeartBeatResponse handleHeartBeat(HeartBeat heartbeat); String getOsType(String os, String osRelease); RegistrationResponse handleRegistration(Register register); ComponentsResponse handleComponents(String clusterName); void stop();  }
@Test @SuppressWarnings("unchecked") public void testStatusHeartbeatWithAnnotation() throws Exception { Cluster cluster = heartbeatTestHelper.getDummyCluster(); Service hdfs = cluster.addService(HDFS); hdfs.addServiceComponent(DATANODE); hdfs.addServiceComponent(NAMENODE); hdfs.addServiceComponent(SECONDARY_NAMENODE); ActionQueue aq = new ActionQueue(); HeartBeat hb = new HeartBeat(); hb.setTimestamp(System.currentTimeMillis()); hb.setResponseId(0); hb.setHostname(DummyHostname1); hb.setNodeStatus(new HostStatus(Status.HEALTHY, DummyHostStatus)); hb.setReports(new ArrayList<CommandReport>()); ArrayList<ComponentStatus> componentStatuses = new ArrayList<ComponentStatus>(); hb.setComponentStatus(componentStatuses); final HostRoleCommand command = hostRoleCommandFactory.create(DummyHostname1, Role.DATANODE, null, null); ActionManager am = actionManagerTestHelper.getMockActionManager(); expect(am.getTasks(EasyMock.<List<Long>>anyObject())).andReturn( new ArrayList<HostRoleCommand>() {{ add(command); }}).anyTimes(); replay(am); HeartBeatHandler handler = heartbeatTestHelper.getHeartBeatHandler(am, aq); HeartBeatResponse resp = handler.handleHeartBeat(hb); Assert.assertFalse(resp.hasMappedComponents()); hdfs.getServiceComponent(DATANODE).addServiceComponentHost(DummyHostname1); ServiceComponentHost serviceComponentHost1 = clusters.getCluster(DummyCluster).getService(HDFS). getServiceComponent(DATANODE).getServiceComponentHost(DummyHostname1); serviceComponentHost1.setState(State.INIT); hb = new HeartBeat(); hb.setTimestamp(System.currentTimeMillis()); hb.setResponseId(1); hb.setHostname(DummyHostname1); hb.setNodeStatus(new HostStatus(Status.HEALTHY, DummyHostStatus)); hb.setReports(new ArrayList<CommandReport>()); hb.setComponentStatus(componentStatuses); resp = handler.handleHeartBeat(hb); Assert.assertTrue(resp.hasMappedComponents()); }