ambari-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mpapirkovs...@apache.org
Subject [3/3] ambari git commit: AMBARI-15141. Start all services request aborts in the middle and hosts go into heartbeat-lost state. (mpapirkovskyy)
Date Tue, 23 Feb 2016 19:04:59 GMT
AMBARI-15141. Start all services request aborts in the middle and hosts go into heartbeat-lost state. (mpapirkovskyy)


Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/083ac6da
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/083ac6da
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/083ac6da

Branch: refs/heads/trunk
Commit: 083ac6dab5cf59c01da054eb656507c089a54620
Parents: 9d7ff5f
Author: Myroslav Papirkovskyi <mpapyrkovskyy@hortonworks.com>
Authored: Tue Feb 23 13:01:13 2016 +0200
Committer: Myroslav Papirkovskyi <mpapyrkovskyy@hortonworks.com>
Committed: Tue Feb 23 21:04:22 2016 +0200

----------------------------------------------------------------------
 .../ambari/server/agent/HeartBeatHandler.java   |  582 +------
 .../ambari/server/agent/HeartbeatMonitor.java   |    6 +
 .../ambari/server/agent/HeartbeatProcessor.java |  773 +++++++++
 .../ambari/server/orm/dao/HostVersionDAO.java   |   78 +-
 .../server/orm/entities/HostVersionEntity.java  |    9 +
 .../server/state/cluster/ClusterImpl.java       |    6 +-
 .../svccomphost/ServiceComponentHostImpl.java   |   72 +-
 .../server/agent/HeartbeatProcessorTest.java    | 1290 +++++++++++++++
 .../server/agent/HeartbeatTestHelper.java       |  229 +++
 .../server/agent/TestHeartbeatHandler.java      | 1489 ++----------------
 10 files changed, 2559 insertions(+), 1975 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ambari/blob/083ac6da/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
index 248ce4b..ba14446 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
@@ -126,6 +126,7 @@ public class HeartBeatHandler {
   private final ActionQueue actionQueue;
   private final ActionManager actionManager;
   private HeartbeatMonitor heartbeatMonitor;
+  private HeartbeatProcessor heartbeatProcessor;
 
   @Inject
   private Injector injector;
@@ -137,38 +138,11 @@ public class HeartBeatHandler {
   private AmbariMetaInfo ambariMetaInfo;
 
   @Inject
-  private ActionMetadata actionMetadata;
-
-  @Inject
-  private Gson gson;
-
-  @Inject
   private ConfigHelper configHelper;
 
   @Inject
-  private HostDAO hostDAO;
-
-  @Inject
   private AlertDefinitionHash alertDefinitionHash;
 
-  /**
-   * Publishes {@link AlertEvent} instances.
-   */
-  @Inject
-  private AlertEventPublisher alertEventPublisher;
-
-  @Inject
-  private AmbariEventPublisher ambariEventPublisher;
-
-  @Inject
-  private VersionEventPublisher versionEventPublisher;
-
-
-  /**
-   * KerberosPrincipalHostDAO used to set and get Kerberos principal details
-   */
-  @Inject
-  private KerberosPrincipalHostDAO kerberosPrincipalHostDAO;
 
   /**
    * KerberosIdentityDataFileReaderFactory used to create KerberosIdentityDataFileReader instances
@@ -187,10 +161,12 @@ public class HeartBeatHandler {
     actionQueue = aq;
     actionManager = am;
     heartbeatMonitor = new HeartbeatMonitor(fsm, aq, am, 60000, injector);
+    heartbeatProcessor = new HeartbeatProcessor(fsm, am, heartbeatMonitor, injector); //TODO modify to match pattern
     injector.injectMembers(this);
   }
 
   public void start() {
+    heartbeatProcessor.startAsync();
     heartbeatMonitor.start();
   }
 
@@ -198,6 +174,14 @@ public class HeartBeatHandler {
     this.heartbeatMonitor = heartbeatMonitor;
   }
 
+  public void setHeartbeatProcessor(HeartbeatProcessor heartbeatProcessor) {
+    this.heartbeatProcessor = heartbeatProcessor;
+  }
+
+  public HeartbeatProcessor getHeartbeatProcessor() {
+    return heartbeatProcessor;
+  }
+
   public HeartBeatResponse handleHeartBeat(HeartBeat heartbeat)
       throws AmbariException {
     long now = System.currentTimeMillis();
@@ -283,18 +267,7 @@ public class HeartBeatHandler {
       return createRegisterCommand();
     }
 
-    // Examine heartbeat for command reports
-    processCommandReports(heartbeat, hostname, clusterFsm, now);
-
-    // Examine heartbeat for component live status reports
-    processStatusReports(heartbeat, hostname, clusterFsm);
-
-    // Calculate host status
-    // NOTE: This step must be after processing command/status reports
-    processHostStatus(heartbeat, hostname);
-
-    // Example heartbeat for alerts from the host or its components
-    processAlerts(heartbeat, hostname);
+    heartbeatProcessor.addHeartbeat(heartbeat);
 
     // Send commands if node is active
     if (hostObject.getState().equals(HostState.HEALTHY)) {
@@ -305,33 +278,7 @@ public class HeartBeatHandler {
     return response;
   }
 
-  /**
-   * Extracts all of the {@link Alert}s from the heartbeat and fires
-   * {@link AlertEvent}s for each one. If there is a problem looking up the
-   * cluster, then alerts will not be processed.
-   *
-   * @param heartbeat
-   *          the heartbeat to process.
-   * @param hostname
-   *          the host that the heartbeat is for.
-   */
-  protected void processAlerts(HeartBeat heartbeat, String hostname) {
-
-    if (null == hostname || null == heartbeat) {
-      return;
-    }
 
-    if (null != heartbeat.getAlerts()) {
-      AlertEvent event = new AlertReceivedEvent(heartbeat.getAlerts());
-      for (Alert alert : event.getAlerts()) {
-        if (alert.getHostName() == null) {
-          alert.setHostName(hostname);
-        }
-      }
-      alertEventPublisher.publish(event);
-
-    }
-  }
 
   protected void processRecoveryReport(RecoveryReport recoveryReport, String hostname) throws AmbariException {
     LOG.debug("Received recovery report: " + recoveryReport.toString());
@@ -339,480 +286,6 @@ public class HeartBeatHandler {
     host.setRecoveryReport(recoveryReport);
   }
 
-  protected void processHostStatus(HeartBeat heartbeat, String hostname) throws AmbariException {
-
-    Host host = clusterFsm.getHost(hostname);
-    HealthStatus healthStatus = host.getHealthStatus().getHealthStatus();
-
-    if (!healthStatus.equals(HostHealthStatus.HealthStatus.UNKNOWN)) {
-
-      List<ComponentStatus> componentStatuses = heartbeat.getComponentStatus();
-      //Host status info could be calculated only if agent returned statuses in heartbeat
-      //Or, if a command is executed that can change component status
-      boolean calculateHostStatus = false;
-      String clusterName = null;
-      if (componentStatuses.size() > 0) {
-        calculateHostStatus = true;
-        for (ComponentStatus componentStatus : componentStatuses) {
-          clusterName = componentStatus.getClusterName();
-          break;
-        }
-      }
-
-      if (!calculateHostStatus) {
-        List<CommandReport> reports = heartbeat.getReports();
-        for (CommandReport report : reports) {
-          if (RoleCommand.ACTIONEXECUTE.toString().equals(report.getRoleCommand())) {
-            continue;
-          }
-
-          String service = report.getServiceName();
-          if (actionMetadata.getActions(service.toLowerCase()).contains(report.getRole())) {
-            continue;
-          }
-          if (report.getStatus().equals("COMPLETED")) {
-            calculateHostStatus = true;
-            clusterName = report.getClusterName();
-            break;
-          }
-        }
-      }
-
-      if (calculateHostStatus) {
-        //Use actual component status to compute the host status
-        int masterCount = 0;
-        int mastersRunning = 0;
-        int slaveCount = 0;
-        int slavesRunning = 0;
-
-        StackId stackId;
-        Cluster cluster = clusterFsm.getCluster(clusterName);
-        stackId = cluster.getDesiredStackVersion();
-
-        MaintenanceStateHelper psh = injector.getInstance(MaintenanceStateHelper.class);
-
-        List<ServiceComponentHost> scHosts = cluster.getServiceComponentHosts(heartbeat.getHostname());
-        for (ServiceComponentHost scHost : scHosts) {
-          ComponentInfo componentInfo =
-              ambariMetaInfo.getComponent(stackId.getStackName(),
-                  stackId.getStackVersion(), scHost.getServiceName(),
-                  scHost.getServiceComponentName());
-
-          String status = scHost.getState().name();
-
-          String category = componentInfo.getCategory();
-
-          if (MaintenanceState.OFF == psh.getEffectiveState(scHost, host)) {
-            if (category.equals("MASTER")) {
-              ++masterCount;
-              if (status.equals("STARTED")) {
-                ++mastersRunning;
-              }
-            } else if (category.equals("SLAVE")) {
-              ++slaveCount;
-              if (status.equals("STARTED")) {
-                ++slavesRunning;
-              }
-            }
-          }
-        }
-
-        if (masterCount == mastersRunning && slaveCount == slavesRunning) {
-          healthStatus = HealthStatus.HEALTHY;
-        } else if (masterCount > 0 && mastersRunning < masterCount) {
-          healthStatus = HealthStatus.UNHEALTHY;
-        } else {
-          healthStatus = HealthStatus.ALERT;
-        }
-
-        host.setStatus(healthStatus.name());
-        host.persist();
-      }
-
-      //If host doesn't belong to any cluster
-      if ((clusterFsm.getClustersForHost(host.getHostName())).size() == 0) {
-        healthStatus = HealthStatus.HEALTHY;
-        host.setStatus(healthStatus.name());
-        host.persist();
-      }
-    }
-  }
-
-  protected void processCommandReports(
-      HeartBeat heartbeat, String hostname, Clusters clusterFsm, long now)
-      throws AmbariException {
-    List<CommandReport> reports = heartbeat.getReports();
-
-    // Cache HostRoleCommand entities because we will need them few times
-    List<Long> taskIds = new ArrayList<Long>();
-    for (CommandReport report : reports) {
-      taskIds.add(report.getTaskId());
-    }
-    Collection<HostRoleCommand> commands = actionManager.getTasks(taskIds);
-
-    Iterator<HostRoleCommand> hostRoleCommandIterator = commands.iterator();
-    for (CommandReport report : reports) {
-
-      Long clusterId = null;
-      if (report.getClusterName() != null) {
-        try {
-          Cluster cluster = clusterFsm.getCluster(report.getClusterName());
-          clusterId = Long.valueOf(cluster.getClusterId());
-        } catch (AmbariException e) {
-        }
-      }
-
-      LOG.debug("Received command report: " + report);
-      // Fetch HostRoleCommand that corresponds to a given task ID
-      HostRoleCommand hostRoleCommand = hostRoleCommandIterator.next();
-      HostEntity hostEntity = hostDAO.findByName(hostname);
-      if (hostEntity == null) {
-        LOG.error("Received a command report and was unable to retrieve HostEntity for hostname = " + hostname);
-        continue;
-      }
-
-      // Send event for final command reports for actions
-      if (RoleCommand.valueOf(report.getRoleCommand()) == RoleCommand.ACTIONEXECUTE &&
-          HostRoleStatus.valueOf(report.getStatus()).isCompletedState()) {
-        ActionFinalReportReceivedEvent event = new ActionFinalReportReceivedEvent(
-                clusterId, hostname, report, false);
-        ambariEventPublisher.publish(event);
-      }
-
-      // Skip sending events for command reports for ABORTed commands
-      if (hostRoleCommand.getStatus() == HostRoleStatus.ABORTED) {
-        continue;
-      }
-      if (hostRoleCommand.getStatus() == HostRoleStatus.QUEUED &&
-              report.getStatus().equals("IN_PROGRESS")) {
-        hostRoleCommand.setStartTime(now);
-      }
-
-      // If the report indicates the keytab file was successfully transferred to a host or removed
-      // from a host, record this for future reference
-      if (Service.Type.KERBEROS.name().equalsIgnoreCase(report.getServiceName()) &&
-          Role.KERBEROS_CLIENT.name().equalsIgnoreCase(report.getRole()) &&
-          RoleCommand.CUSTOM_COMMAND.name().equalsIgnoreCase(report.getRoleCommand()) &&
-          RequestExecution.Status.COMPLETED.name().equalsIgnoreCase(report.getStatus())) {
-
-        String customCommand = report.getCustomCommand();
-
-        boolean adding = "SET_KEYTAB".equalsIgnoreCase(customCommand);
-        if (adding || "REMOVE_KEYTAB".equalsIgnoreCase(customCommand)) {
-          WriteKeytabsStructuredOut writeKeytabsStructuredOut;
-          try {
-            writeKeytabsStructuredOut = gson.fromJson(report.getStructuredOut(), WriteKeytabsStructuredOut.class);
-          } catch (JsonSyntaxException ex) {
-            //Json structure was incorrect do nothing, pass this data further for processing
-            writeKeytabsStructuredOut = null;
-          }
-
-          if (writeKeytabsStructuredOut != null) {
-            Map<String, String> keytabs = writeKeytabsStructuredOut.getKeytabs();
-            if (keytabs != null) {
-              for (Map.Entry<String, String> entry : keytabs.entrySet()) {
-                String principal = entry.getKey();
-                if (!kerberosPrincipalHostDAO.exists(principal, hostEntity.getHostId())) {
-                  if (adding) {
-                    kerberosPrincipalHostDAO.create(principal, hostEntity.getHostId());
-                  } else if ("_REMOVED_".equalsIgnoreCase(entry.getValue())) {
-                    kerberosPrincipalHostDAO.remove(principal, hostEntity.getHostId());
-                  }
-                }
-              }
-            }
-          }
-        }
-      }
-
-      //pass custom START, STOP and RESTART
-      if (RoleCommand.ACTIONEXECUTE.toString().equals(report.getRoleCommand()) ||
-         (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
-         !("RESTART".equals(report.getCustomCommand()) ||
-         "START".equals(report.getCustomCommand()) ||
-         "STOP".equals(report.getCustomCommand())))) {
-        continue;
-      }
-
-      Cluster cl = clusterFsm.getCluster(report.getClusterName());
-      String service = report.getServiceName();
-      if (service == null || service.isEmpty()) {
-        throw new AmbariException("Invalid command report, service: " + service);
-      }
-      if (actionMetadata.getActions(service.toLowerCase()).contains(report.getRole())) {
-        LOG.debug(report.getRole() + " is an action - skip component lookup");
-      } else {
-        try {
-          Service svc = cl.getService(service);
-          ServiceComponent svcComp = svc.getServiceComponent(report.getRole());
-          ServiceComponentHost scHost = svcComp.getServiceComponentHost(hostname);
-          String schName = scHost.getServiceComponentName();
-
-          if (report.getStatus().equals(HostRoleStatus.COMPLETED.toString())) {
-
-            // Reading component version if it is present
-            if (StringUtils.isNotBlank(report.getStructuredOut())) {
-              ComponentVersionStructuredOut structuredOutput = null;
-              try {
-                structuredOutput = gson.fromJson(report.getStructuredOut(), ComponentVersionStructuredOut.class);
-              } catch (JsonSyntaxException ex) {
-                //Json structure for component version was incorrect
-                //do nothing, pass this data further for processing
-              }
-
-              String newVersion = structuredOutput == null ? null : structuredOutput.version;
-
-              // Pass true to always publish a version event.  It is safer to recalculate the version even if we don't
-              // detect a difference in the value.  This is useful in case that a manual database edit is done while
-              // ambari-server is stopped.
-              handleComponentVersionReceived(cl, scHost, newVersion, true);
-            }
-
-            // Updating stack version, if needed (this is not actually for express/rolling upgrades!)
-            if (scHost.getState().equals(State.UPGRADING)) {
-              scHost.setStackVersion(scHost.getDesiredStackVersion());
-            } else if ((report.getRoleCommand().equals(RoleCommand.START.toString()) ||
-                (report.getRoleCommand().equals(RoleCommand.CUSTOM_COMMAND.toString()) &&
-                    ("START".equals(report.getCustomCommand()) ||
-                    "RESTART".equals(report.getCustomCommand()))))
-                && null != report.getConfigurationTags()
-                && !report.getConfigurationTags().isEmpty()) {
-              LOG.info("Updating applied config on service " + scHost.getServiceName() +
-                ", component " + scHost.getServiceComponentName() + ", host " + scHost.getHostName());
-              scHost.updateActualConfigs(report.getConfigurationTags());
-              scHost.setRestartRequired(false);
-            }
-            // Necessary for resetting clients stale configs after starting service
-            if ((RoleCommand.INSTALL.toString().equals(report.getRoleCommand()) ||
-                (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
-                "INSTALL".equals(report.getCustomCommand()))) && svcComp.isClientComponent()){
-              scHost.updateActualConfigs(report.getConfigurationTags());
-              scHost.setRestartRequired(false);
-            }
-            if (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
-                !("START".equals(report.getCustomCommand()) ||
-                 "STOP".equals(report.getCustomCommand()))) {
-              //do not affect states for custom commands except START and STOP
-              //lets status commands to be responsible for this
-              continue;
-            }
-
-            if (RoleCommand.START.toString().equals(report.getRoleCommand()) ||
-                (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
-                    "START".equals(report.getCustomCommand()))) {
-              scHost.handleEvent(new ServiceComponentHostStartedEvent(schName,
-                  hostname, now));
-              scHost.setRestartRequired(false);
-            } else if (RoleCommand.STOP.toString().equals(report.getRoleCommand()) ||
-                (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
-                    "STOP".equals(report.getCustomCommand()))) {
-              scHost.handleEvent(new ServiceComponentHostStoppedEvent(schName,
-                  hostname, now));
-            } else {
-              scHost.handleEvent(new ServiceComponentHostOpSucceededEvent(schName,
-                  hostname, now));
-            }
-          } else if (report.getStatus().equals("FAILED")) {
-
-            if (StringUtils.isNotBlank(report.getStructuredOut())) {
-              try {
-                ComponentVersionStructuredOut structuredOutput = gson.fromJson(report.getStructuredOut(), ComponentVersionStructuredOut.class);
-
-                if (null != structuredOutput.upgradeDirection && structuredOutput.upgradeDirection.isUpgrade()) {
-                  scHost.setUpgradeState(UpgradeState.FAILED);
-                }
-              } catch (JsonSyntaxException ex) {
-                LOG.warn("Structured output was found, but not parseable: {}", report.getStructuredOut());
-              }
-            }
-
-            LOG.warn("Operation failed - may be retried. Service component host: "
-                + schName + ", host: " + hostname + " Action id" + report.getActionId());
-            if (actionManager.isInProgressCommand(report)) {
-              scHost.handleEvent(new ServiceComponentHostOpFailedEvent
-                (schName, hostname, now));
-            } else {
-              LOG.info("Received report for a command that is no longer active. " + report);
-            }
-          } else if (report.getStatus().equals("IN_PROGRESS")) {
-            scHost.handleEvent(new ServiceComponentHostOpInProgressEvent(schName,
-                hostname, now));
-          }
-        } catch (ServiceComponentNotFoundException scnex) {
-          LOG.warn("Service component not found ", scnex);
-        } catch (InvalidStateTransitionException ex) {
-          if (LOG.isDebugEnabled()) {
-            LOG.warn("State machine exception.", ex);
-          } else {
-            LOG.warn("State machine exception. " + ex.getMessage());
-          }
-        }
-      }
-    }
-
-    //Update state machines from reports
-    actionManager.processTaskResponse(hostname, reports, commands);
-  }
-
-  protected void processStatusReports(HeartBeat heartbeat,
-                                      String hostname,
-                                      Clusters clusterFsm)
-      throws AmbariException {
-    Set<Cluster> clusters = clusterFsm.getClustersForHost(hostname);
-    for (Cluster cl : clusters) {
-      for (ComponentStatus status : heartbeat.componentStatus) {
-        if (status.getClusterName().equals(cl.getClusterName())) {
-          try {
-            Service svc = cl.getService(status.getServiceName());
-
-            String componentName = status.getComponentName();
-            if (svc.getServiceComponents().containsKey(componentName)) {
-              ServiceComponent svcComp = svc.getServiceComponent(
-                  componentName);
-              ServiceComponentHost scHost = svcComp.getServiceComponentHost(
-                  hostname);
-              State prevState = scHost.getState();
-              State liveState = State.valueOf(State.class, status.getStatus());
-              if (prevState.equals(State.INSTALLED)
-                  || prevState.equals(State.STARTED)
-                  || prevState.equals(State.STARTING)
-                  || prevState.equals(State.STOPPING)
-                  || prevState.equals(State.UNKNOWN)) {
-                scHost.setState(liveState); //TODO direct status set breaks state machine sometimes !!!
-                if (!prevState.equals(liveState)) {
-                  LOG.info("State of service component " + componentName
-                      + " of service " + status.getServiceName()
-                      + " of cluster " + status.getClusterName()
-                      + " has changed from " + prevState + " to " + liveState
-                      + " at host " + hostname);
-                }
-              }
-
-              SecurityState prevSecurityState = scHost.getSecurityState();
-              SecurityState currentSecurityState = SecurityState.valueOf(status.getSecurityState());
-              if((prevSecurityState != currentSecurityState)) {
-                if(prevSecurityState.isEndpoint()) {
-                  scHost.setSecurityState(currentSecurityState);
-                  LOG.info(String.format("Security of service component %s of service %s of cluster %s " +
-                          "has changed from %s to %s on host %s",
-                      componentName, status.getServiceName(), status.getClusterName(), prevSecurityState,
-                      currentSecurityState, hostname));
-                }
-                else {
-                  LOG.debug(String.format("Security of service component %s of service %s of cluster %s " +
-                          "has changed from %s to %s on host %s but will be ignored since %s is a " +
-                          "transitional state",
-                      componentName, status.getServiceName(), status.getClusterName(),
-                      prevSecurityState, currentSecurityState, hostname, prevSecurityState));
-                }
-              }
-
-              if (null != status.getStackVersion() && !status.getStackVersion().isEmpty()) {
-                scHost.setStackVersion(gson.fromJson(status.getStackVersion(), StackId.class));
-              }
-
-              if (null != status.getConfigTags()) {
-                scHost.updateActualConfigs(status.getConfigTags());
-              }
-
-              Map<String, Object> extra = status.getExtra();
-              if (null != extra && !extra.isEmpty()) {
-                try {
-                  if (extra.containsKey("processes")) {
-                    @SuppressWarnings("unchecked")
-                    List<Map<String, String>> list = (List<Map<String, String>>) extra.get("processes");
-                    scHost.setProcesses(list);
-                  }
-                  if (extra.containsKey("version")) {
-                    String version = extra.get("version").toString();
-
-                    handleComponentVersionReceived(cl, scHost, version, false);
-                  }
-
-                } catch (Exception e) {
-                  LOG.error("Could not access extra JSON for " +
-                      scHost.getServiceComponentName() + " from " +
-                      scHost.getHostName() + ": " + status.getExtra() +
-                      " (" + e.getMessage() + ")");
-                }
-              }
-
-              this.heartbeatMonitor.getAgentRequests()
-                  .setExecutionDetailsRequest(hostname, componentName, status.getSendExecCmdDet());
-            } else {
-              // TODO: What should be done otherwise?
-            }
-          } catch (ServiceNotFoundException e) {
-            LOG.warn("Received a live status update for a non-initialized"
-                + " service"
-                + ", clusterName=" + status.getClusterName()
-                + ", serviceName=" + status.getServiceName());
-            // FIXME ignore invalid live update and continue for now?
-            continue;
-          } catch (ServiceComponentNotFoundException e) {
-            LOG.warn("Received a live status update for a non-initialized"
-                + " servicecomponent"
-                + ", clusterName=" + status.getClusterName()
-                + ", serviceName=" + status.getServiceName()
-                + ", componentName=" + status.getComponentName());
-            // FIXME ignore invalid live update and continue for now?
-            continue;
-          } catch (ServiceComponentHostNotFoundException e) {
-            LOG.warn("Received a live status update for a non-initialized"
-                + " service"
-                + ", clusterName=" + status.getClusterName()
-                + ", serviceName=" + status.getServiceName()
-                + ", componentName=" + status.getComponentName()
-                + ", hostname=" + hostname);
-            // FIXME ignore invalid live update and continue for now?
-            continue;
-          } catch (RuntimeException e) {
-            LOG.warn("Received a live status with invalid payload"
-                + " service"
-                + ", clusterName=" + status.getClusterName()
-                + ", serviceName=" + status.getServiceName()
-                + ", componentName=" + status.getComponentName()
-                + ", hostname=" + hostname
-                + ", error=" + e.getMessage());
-            continue;
-          }
-        }
-      }
-    }
-  }
-
-  /**
-   * Updates the version of the given service component, sets the upgrade state (if needed)
-   * and publishes a version event through the version event publisher.
-   *
-   * @param cluster        the cluster
-   * @param scHost         service component host
-   * @param newVersion     new version of service component
-   * @param alwaysPublish  if true, always publish a version event; if false,
-   *                       only publish if the component version was updated
-   */
-  private void handleComponentVersionReceived(Cluster cluster, ServiceComponentHost scHost,
-                                              String newVersion, boolean alwaysPublish) {
-
-    boolean updated = false;
-
-    if (StringUtils.isNotBlank(newVersion)) {
-      final String previousVersion = scHost.getVersion();
-      if (!StringUtils.equals(previousVersion, newVersion)) {
-        scHost.setVersion(newVersion);
-        scHost.setStackVersion(cluster.getDesiredStackVersion());
-        if (previousVersion != null && !previousVersion.equalsIgnoreCase(State.UNKNOWN.toString())) {
-          scHost.setUpgradeState(UpgradeState.COMPLETE);
-        }
-        updated = true;
-      }
-    }
-
-    if (updated || alwaysPublish) {
-      HostComponentVersionEvent event = new HostComponentVersionEvent(cluster, scHost);
-      versionEventPublisher.publish(event);
-    }
-  }
 
   /**
    * Adds commands from action queue to a heartbeat response.
@@ -1229,35 +702,4 @@ public class HeartBeatHandler {
     }
   }
 
-  /**
-   * This class is used for mapping json of structured output for component START action.
-   */
-  private static class ComponentVersionStructuredOut {
-    @SerializedName("version")
-    private String version;
-
-    @SerializedName("upgrade_type")
-    private UpgradeType upgradeType = null;
-
-    @SerializedName("direction")
-    private Direction upgradeDirection = null;
-
-  }
-
-  /**
-   * This class is used for mapping json of structured output for keytab distribution actions.
-   */
-  private static class WriteKeytabsStructuredOut {
-    @SerializedName("keytabs")
-    private Map<String,String> keytabs;
-
-    public Map<String, String> getKeytabs() {
-      return keytabs;
-    }
-
-    public void setKeytabs(Map<String, String> keytabs) {
-      this.keytabs = keytabs;
-    }
-  }
-
 }

http://git-wip-us.apache.org/repos/asf/ambari/blob/083ac6da/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatMonitor.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatMonitor.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatMonitor.java
index efc717d..378e123 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatMonitor.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatMonitor.java
@@ -140,6 +140,10 @@ public class HeartbeatMonitor implements Runnable {
     List<Host> allHosts = clusters.getHosts();
     long now = System.currentTimeMillis();
     for (Host hostObj : allHosts) {
+      if (hostObj.getState() == HostState.HEARTBEAT_LOST) {
+        //do not check if host already known be lost
+        continue;
+      }
       String host = hostObj.getHostName();
       HostState hostState = hostObj.getState();
       String hostname = hostObj.getHostName();
@@ -212,6 +216,8 @@ public class HeartbeatMonitor implements Runnable {
         switch (sch.getState()) {
           case INIT:
           case INSTALLING:
+          case STARTING:
+          case STOPPING:
             //don't send commands until component is installed at least
             continue;
           default:

http://git-wip-us.apache.org/repos/asf/ambari/blob/083ac6da/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatProcessor.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatProcessor.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatProcessor.java
new file mode 100644
index 0000000..2188a77
--- /dev/null
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatProcessor.java
@@ -0,0 +1,773 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.agent;
+
+
+import com.google.common.util.concurrent.AbstractScheduledService;
+import com.google.common.util.concurrent.AbstractService;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.gson.Gson;
+import com.google.gson.JsonSyntaxException;
+import com.google.gson.annotations.SerializedName;
+import com.google.inject.Inject;
+import com.google.inject.Injector;
+import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.Role;
+import org.apache.ambari.server.RoleCommand;
+import org.apache.ambari.server.ServiceComponentHostNotFoundException;
+import org.apache.ambari.server.ServiceComponentNotFoundException;
+import org.apache.ambari.server.ServiceNotFoundException;
+import org.apache.ambari.server.actionmanager.ActionManager;
+import org.apache.ambari.server.actionmanager.HostRoleCommand;
+import org.apache.ambari.server.actionmanager.HostRoleStatus;
+import org.apache.ambari.server.api.services.AmbariMetaInfo;
+import org.apache.ambari.server.controller.MaintenanceStateHelper;
+import org.apache.ambari.server.events.ActionFinalReportReceivedEvent;
+import org.apache.ambari.server.events.AlertEvent;
+import org.apache.ambari.server.events.AlertReceivedEvent;
+import org.apache.ambari.server.events.HostComponentVersionEvent;
+import org.apache.ambari.server.events.publishers.AlertEventPublisher;
+import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
+import org.apache.ambari.server.events.publishers.VersionEventPublisher;
+import org.apache.ambari.server.metadata.ActionMetadata;
+import org.apache.ambari.server.orm.dao.KerberosPrincipalHostDAO;
+import org.apache.ambari.server.state.Alert;
+import org.apache.ambari.server.state.Cluster;
+import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.ComponentInfo;
+import org.apache.ambari.server.state.Host;
+import org.apache.ambari.server.state.HostHealthStatus;
+import org.apache.ambari.server.state.MaintenanceState;
+import org.apache.ambari.server.state.SecurityState;
+import org.apache.ambari.server.state.Service;
+import org.apache.ambari.server.state.ServiceComponent;
+import org.apache.ambari.server.state.ServiceComponentHost;
+import org.apache.ambari.server.state.StackId;
+import org.apache.ambari.server.state.State;
+import org.apache.ambari.server.state.UpgradeState;
+import org.apache.ambari.server.state.fsm.InvalidStateTransitionException;
+import org.apache.ambari.server.state.scheduler.RequestExecution;
+import org.apache.ambari.server.state.stack.upgrade.Direction;
+import org.apache.ambari.server.state.stack.upgrade.UpgradeType;
+import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpFailedEvent;
+import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpInProgressEvent;
+import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpSucceededEvent;
+import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStartedEvent;
+import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStoppedEvent;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * HeartbeatProcessor class is used for bulk processing data retrieved from agents in background
+ *
+ */
+public class HeartbeatProcessor extends AbstractService{
+  private static final Logger LOG = LoggerFactory.getLogger(HeartbeatProcessor.class);
+
+  private ScheduledExecutorService executor;
+
+  private ConcurrentLinkedQueue<HeartBeat> heartBeatsQueue = new ConcurrentLinkedQueue<>();
+
+  private volatile boolean shouldRun = true;
+
+  //TODO rewrite to correlate with heartbeat frequency, hardcoded in agent as of now
+  private long delay = 5000;
+  private long period = 1000;
+
+  private int poolSize = 1;
+
+  private Clusters clusterFsm;
+  private HeartbeatMonitor heartbeatMonitor;
+  private Injector injector;
+  private ActionManager actionManager;
+
+  /**
+   * Publishes {@link AlertEvent} instances.
+   */
+  @Inject
+  AlertEventPublisher alertEventPublisher;
+
+  @Inject
+  AmbariEventPublisher ambariEventPublisher;
+
+  @Inject
+  VersionEventPublisher versionEventPublisher;
+
+  @Inject
+  ActionMetadata actionMetadata;
+
+  @Inject
+  MaintenanceStateHelper maintenanceStateHelper;
+
+  @Inject
+  AmbariMetaInfo ambariMetaInfo;
+
+  @Inject
+  KerberosPrincipalHostDAO kerberosPrincipalHostDAO;
+
+  @Inject
+  Gson gson;
+
+  @Inject
+  public HeartbeatProcessor(Clusters clusterFsm, ActionManager am, HeartbeatMonitor heartbeatMonitor,
+                            Injector injector) {
+    injector.injectMembers(this);
+
+    this.injector = injector;
+    this.heartbeatMonitor = heartbeatMonitor;
+    this.clusterFsm = clusterFsm;
+    actionManager = am;
+    ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("ambari-heartbeat-processor-%d").build();
+    executor = Executors.newScheduledThreadPool(poolSize, threadFactory);
+  }
+
+  @Override
+  protected void doStart() {
+    LOG.info("**** Starting heartbeats processing threads ****");
+    for (int i=0; i< poolSize; i++) {
+      executor.scheduleAtFixedRate(new HeartbeatProcessingTask(), delay, period, TimeUnit.MILLISECONDS);
+    }
+  }
+
+  @Override
+  protected void doStop() {
+    LOG.info("**** Stopping heartbeats processing threads ****");
+    shouldRun = false;
+    executor.shutdown();
+  }
+
+  public void addHeartbeat(HeartBeat heartBeat) {
+    heartBeatsQueue.add(heartBeat);
+  }
+
+  private HeartBeat pollHeartbeat() {
+    return heartBeatsQueue.poll();
+  }
+
+  /**
+   * Processing task to be scheduled for execution
+   */
+  private class HeartbeatProcessingTask implements Runnable {
+
+    @Override
+    public void run() {
+      while (shouldRun) {
+        try {
+          HeartBeat heartbeat = pollHeartbeat();
+          if (heartbeat == null) {
+            break;
+          }
+          processHeartbeat(heartbeat);
+        } catch (Exception e) {
+          LOG.error("Exception received while processing heartbeat", e);
+        } catch (Throwable throwable) {
+          //catch everything to prevent task suppression
+          LOG.error("ERROR: ", throwable);
+        }
+
+
+      }
+    }
+  }
+
+  /**
+   * Incapsulates logic for processing data from agent heartbeat
+   * @param heartbeat Agent heartbeat object
+   * @throws AmbariException
+   */
+  public void processHeartbeat(HeartBeat heartbeat) throws AmbariException {
+    long now = System.currentTimeMillis();
+
+    processAlerts(heartbeat);
+
+    processCommandReports(heartbeat, now);
+    processStatusReports(heartbeat);
+    //host status calculation are based on task and status reports, should be performed last
+    processHostStatus(heartbeat);
+  }
+
+
+
+  /**
+   * Extracts all of the {@link Alert}s from the heartbeat and fires
+   * {@link AlertEvent}s for each one. If there is a problem looking up the
+   * cluster, then alerts will not be processed.
+   *
+   * @param heartbeat
+   *          the heartbeat to process.
+   */
+  protected void processAlerts(HeartBeat heartbeat) {
+    if (heartbeat == null) {
+      return;
+    }
+
+    String hostname = heartbeat.getHostname();
+
+    if (null != heartbeat.getAlerts()) {
+      AlertEvent event = new AlertReceivedEvent(heartbeat.getAlerts());
+      for (Alert alert : event.getAlerts()) {
+        if (alert.getHostName() == null) {
+          alert.setHostName(hostname);
+        }
+      }
+      alertEventPublisher.publish(event);
+
+    }
+  }
+
+  /**
+   * Update host status basing on components statuses
+   * @param heartbeat heartbeat to process
+   * @throws AmbariException
+   */
+  protected void processHostStatus(HeartBeat heartbeat) throws AmbariException {
+
+    String hostname = heartbeat.getHostname();
+    Host host = clusterFsm.getHost(hostname);
+    HostHealthStatus.HealthStatus healthStatus = host.getHealthStatus().getHealthStatus();
+
+    if (!healthStatus.equals(HostHealthStatus.HealthStatus.UNKNOWN)) {
+
+      List<ComponentStatus> componentStatuses = heartbeat.getComponentStatus();
+      //Host status info could be calculated only if agent returned statuses in heartbeat
+      //Or, if a command is executed that can change component status
+      boolean calculateHostStatus = false;
+      String clusterName = null;
+      if (componentStatuses.size() > 0) {
+        calculateHostStatus = true;
+        for (ComponentStatus componentStatus : componentStatuses) {
+          clusterName = componentStatus.getClusterName();
+          break;
+        }
+      }
+
+      if (!calculateHostStatus) {
+        List<CommandReport> reports = heartbeat.getReports();
+        for (CommandReport report : reports) {
+          if (RoleCommand.ACTIONEXECUTE.toString().equals(report.getRoleCommand())) {
+            continue;
+          }
+
+          String service = report.getServiceName();
+          if (actionMetadata.getActions(service.toLowerCase()).contains(report.getRole())) {
+            continue;
+          }
+          if (report.getStatus().equals("COMPLETED")) {
+            calculateHostStatus = true;
+            clusterName = report.getClusterName();
+            break;
+          }
+        }
+      }
+
+      if (calculateHostStatus) {
+        //Use actual component status to compute the host status
+        int masterCount = 0;
+        int mastersRunning = 0;
+        int slaveCount = 0;
+        int slavesRunning = 0;
+
+        StackId stackId;
+        Cluster cluster = clusterFsm.getCluster(clusterName);
+        stackId = cluster.getDesiredStackVersion();
+
+
+        List<ServiceComponentHost> scHosts = cluster.getServiceComponentHosts(heartbeat.getHostname());
+        for (ServiceComponentHost scHost : scHosts) {
+          ComponentInfo componentInfo =
+              ambariMetaInfo.getComponent(stackId.getStackName(),
+                  stackId.getStackVersion(), scHost.getServiceName(),
+                  scHost.getServiceComponentName());
+
+          String status = scHost.getState().name();
+
+          String category = componentInfo.getCategory();
+
+          if (MaintenanceState.OFF == maintenanceStateHelper.getEffectiveState(scHost, host)) {
+            if (category.equals("MASTER")) {
+              ++masterCount;
+              if (status.equals("STARTED")) {
+                ++mastersRunning;
+              }
+            } else if (category.equals("SLAVE")) {
+              ++slaveCount;
+              if (status.equals("STARTED")) {
+                ++slavesRunning;
+              }
+            }
+          }
+        }
+
+        if (masterCount == mastersRunning && slaveCount == slavesRunning) {
+          healthStatus = HostHealthStatus.HealthStatus.HEALTHY;
+        } else if (masterCount > 0 && mastersRunning < masterCount) {
+          healthStatus = HostHealthStatus.HealthStatus.UNHEALTHY;
+        } else {
+          healthStatus = HostHealthStatus.HealthStatus.ALERT;
+        }
+
+        host.setStatus(healthStatus.name());
+        host.persist();
+      }
+
+      //If host doesn't belong to any cluster
+      if ((clusterFsm.getClustersForHost(host.getHostName())).size() == 0) {
+        healthStatus = HostHealthStatus.HealthStatus.HEALTHY;
+        host.setStatus(healthStatus.name());
+        host.persist();
+      }
+    }
+  }
+
+  /**
+   * Process reports of tasks executed on agents
+   * @param heartbeat heartbeat to process
+   * @param now cached current time
+   * @throws AmbariException
+   */
+  protected void processCommandReports(
+      HeartBeat heartbeat, long now)
+      throws AmbariException {
+    String hostname = heartbeat.getHostname();
+    List<CommandReport> reports = heartbeat.getReports();
+
+    // Cache HostRoleCommand entities because we will need them few times
+    List<Long> taskIds = new ArrayList<Long>();
+    for (CommandReport report : reports) {
+      taskIds.add(report.getTaskId());
+    }
+    Collection<HostRoleCommand> commands = actionManager.getTasks(taskIds);
+
+    Iterator<HostRoleCommand> hostRoleCommandIterator = commands.iterator();
+    for (CommandReport report : reports) {
+
+      Long clusterId = null;
+      if (report.getClusterName() != null) {
+        try {
+          Cluster cluster = clusterFsm.getCluster(report.getClusterName());
+          clusterId = cluster.getClusterId();
+        } catch (AmbariException e) {
+        }
+      }
+
+      LOG.debug("Received command report: " + report);
+      // Fetch HostRoleCommand that corresponds to a given task ID
+      HostRoleCommand hostRoleCommand = hostRoleCommandIterator.next();
+      Host host = clusterFsm.getHost(hostname);
+//      HostEntity hostEntity = hostDAO.findByName(hostname); //don't touch database
+      if (host == null) {
+        LOG.error("Received a command report and was unable to retrieve Host for hostname = " + hostname);
+        continue;
+      }
+
+      // Send event for final command reports for actions
+      if (RoleCommand.valueOf(report.getRoleCommand()) == RoleCommand.ACTIONEXECUTE &&
+          HostRoleStatus.valueOf(report.getStatus()).isCompletedState()) {
+        ActionFinalReportReceivedEvent event = new ActionFinalReportReceivedEvent(
+            clusterId, hostname, report, false);
+        ambariEventPublisher.publish(event);
+      }
+
+      // Skip sending events for command reports for ABORTed commands
+      if (hostRoleCommand.getStatus() == HostRoleStatus.ABORTED) {
+        continue;
+      }
+      if (hostRoleCommand.getStatus() == HostRoleStatus.QUEUED &&
+          report.getStatus().equals("IN_PROGRESS")) {
+        hostRoleCommand.setStartTime(now);
+      }
+
+      // If the report indicates the keytab file was successfully transferred to a host or removed
+      // from a host, record this for future reference
+      if (Service.Type.KERBEROS.name().equalsIgnoreCase(report.getServiceName()) &&
+          Role.KERBEROS_CLIENT.name().equalsIgnoreCase(report.getRole()) &&
+          RoleCommand.CUSTOM_COMMAND.name().equalsIgnoreCase(report.getRoleCommand()) &&
+          RequestExecution.Status.COMPLETED.name().equalsIgnoreCase(report.getStatus())) {
+
+        String customCommand = report.getCustomCommand();
+
+        boolean adding = "SET_KEYTAB".equalsIgnoreCase(customCommand);
+        if (adding || "REMOVE_KEYTAB".equalsIgnoreCase(customCommand)) {
+          WriteKeytabsStructuredOut writeKeytabsStructuredOut;
+          try {
+            writeKeytabsStructuredOut = gson.fromJson(report.getStructuredOut(), WriteKeytabsStructuredOut.class);
+          } catch (JsonSyntaxException ex) {
+            //Json structure was incorrect do nothing, pass this data further for processing
+            writeKeytabsStructuredOut = null;
+          }
+
+          if (writeKeytabsStructuredOut != null) {
+            Map<String, String> keytabs = writeKeytabsStructuredOut.getKeytabs();
+            if (keytabs != null) {
+              for (Map.Entry<String, String> entry : keytabs.entrySet()) {
+                String principal = entry.getKey();
+                if (!kerberosPrincipalHostDAO.exists(principal, host.getHostId())) {
+                  if (adding) {
+                    kerberosPrincipalHostDAO.create(principal, host.getHostId());
+                  } else if ("_REMOVED_".equalsIgnoreCase(entry.getValue())) {
+                    kerberosPrincipalHostDAO.remove(principal, host.getHostId());
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+
+      //pass custom START, STOP and RESTART
+      if (RoleCommand.ACTIONEXECUTE.toString().equals(report.getRoleCommand()) ||
+          (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
+              !("RESTART".equals(report.getCustomCommand()) ||
+                  "START".equals(report.getCustomCommand()) ||
+                  "STOP".equals(report.getCustomCommand())))) {
+        continue;
+      }
+
+      Cluster cl = clusterFsm.getCluster(report.getClusterName());
+      String service = report.getServiceName();
+      if (service == null || service.isEmpty()) {
+        throw new AmbariException("Invalid command report, service: " + service);
+      }
+      if (actionMetadata.getActions(service.toLowerCase()).contains(report.getRole())) {
+        LOG.debug(report.getRole() + " is an action - skip component lookup");
+      } else {
+        try {
+          Service svc = cl.getService(service);
+          ServiceComponent svcComp = svc.getServiceComponent(report.getRole());
+          ServiceComponentHost scHost = svcComp.getServiceComponentHost(hostname);
+          String schName = scHost.getServiceComponentName();
+
+          if (report.getStatus().equals(HostRoleStatus.COMPLETED.toString())) {
+
+            // Reading component version if it is present
+            if (StringUtils.isNotBlank(report.getStructuredOut())) {
+              ComponentVersionStructuredOut structuredOutput = null;
+              try {
+                structuredOutput = gson.fromJson(report.getStructuredOut(), ComponentVersionStructuredOut.class);
+              } catch (JsonSyntaxException ex) {
+                //Json structure for component version was incorrect
+                //do nothing, pass this data further for processing
+              }
+
+              String newVersion = structuredOutput == null ? null : structuredOutput.version;
+
+              // Pass true to always publish a version event.  It is safer to recalculate the version even if we don't
+              // detect a difference in the value.  This is useful in case that a manual database edit is done while
+              // ambari-server is stopped.
+              handleComponentVersionReceived(cl, scHost, newVersion, true);
+            }
+
+            // Updating stack version, if needed (this is not actually for express/rolling upgrades!)
+            if (scHost.getState().equals(org.apache.ambari.server.state.State.UPGRADING)) {
+              scHost.setStackVersion(scHost.getDesiredStackVersion());
+            } else if ((report.getRoleCommand().equals(RoleCommand.START.toString()) ||
+                (report.getRoleCommand().equals(RoleCommand.CUSTOM_COMMAND.toString()) &&
+                    ("START".equals(report.getCustomCommand()) ||
+                        "RESTART".equals(report.getCustomCommand()))))
+                && null != report.getConfigurationTags()
+                && !report.getConfigurationTags().isEmpty()) {
+              LOG.info("Updating applied config on service " + scHost.getServiceName() +
+                  ", component " + scHost.getServiceComponentName() + ", host " + scHost.getHostName());
+              scHost.updateActualConfigs(report.getConfigurationTags());
+              scHost.setRestartRequired(false);
+            }
+            // Necessary for resetting clients stale configs after starting service
+            if ((RoleCommand.INSTALL.toString().equals(report.getRoleCommand()) ||
+                (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
+                    "INSTALL".equals(report.getCustomCommand()))) && svcComp.isClientComponent()){
+              scHost.updateActualConfigs(report.getConfigurationTags());
+              scHost.setRestartRequired(false);
+            }
+            if (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
+                !("START".equals(report.getCustomCommand()) ||
+                    "STOP".equals(report.getCustomCommand()))) {
+              //do not affect states for custom commands except START and STOP
+              //lets status commands to be responsible for this
+              continue;
+            }
+
+            if (RoleCommand.START.toString().equals(report.getRoleCommand()) ||
+                (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
+                    "START".equals(report.getCustomCommand()))) {
+              scHost.handleEvent(new ServiceComponentHostStartedEvent(schName,
+                  hostname, now));
+              scHost.setRestartRequired(false);
+            } else if (RoleCommand.STOP.toString().equals(report.getRoleCommand()) ||
+                (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
+                    "STOP".equals(report.getCustomCommand()))) {
+              scHost.handleEvent(new ServiceComponentHostStoppedEvent(schName,
+                  hostname, now));
+            } else {
+              scHost.handleEvent(new ServiceComponentHostOpSucceededEvent(schName,
+                  hostname, now));
+            }
+          } else if (report.getStatus().equals("FAILED")) {
+
+            if (StringUtils.isNotBlank(report.getStructuredOut())) {
+              try {
+                ComponentVersionStructuredOut structuredOutput = gson.fromJson(report.getStructuredOut(), ComponentVersionStructuredOut.class);
+
+                if (null != structuredOutput.upgradeDirection && structuredOutput.upgradeDirection.isUpgrade()) {
+                  scHost.setUpgradeState(UpgradeState.FAILED);
+                }
+              } catch (JsonSyntaxException ex) {
+                LOG.warn("Structured output was found, but not parseable: {}", report.getStructuredOut());
+              }
+            }
+
+            LOG.warn("Operation failed - may be retried. Service component host: "
+                + schName + ", host: " + hostname + " Action id" + report.getActionId());
+            if (actionManager.isInProgressCommand(report)) {
+              scHost.handleEvent(new ServiceComponentHostOpFailedEvent
+                  (schName, hostname, now));
+            } else {
+              LOG.info("Received report for a command that is no longer active. " + report);
+            }
+          } else if (report.getStatus().equals("IN_PROGRESS")) {
+            scHost.handleEvent(new ServiceComponentHostOpInProgressEvent(schName,
+                hostname, now));
+          }
+        } catch (ServiceComponentNotFoundException scnex) {
+          LOG.warn("Service component not found ", scnex);
+        } catch (InvalidStateTransitionException ex) {
+          if (LOG.isDebugEnabled()) {
+            LOG.warn("State machine exception.", ex);
+          } else {
+            LOG.warn("State machine exception. " + ex.getMessage());
+          }
+        }
+      }
+    }
+
+    //Update state machines from reports
+    actionManager.processTaskResponse(hostname, reports, commands);
+  }
+
+  /**
+   * Process reports of status commands
+   * @param heartbeat heartbeat to process
+   * @throws AmbariException
+   */
+  protected void processStatusReports(HeartBeat heartbeat) throws AmbariException {
+    String hostname = heartbeat.getHostname();
+    Set<Cluster> clusters = clusterFsm.getClustersForHost(hostname);
+    for (Cluster cl : clusters) {
+      for (ComponentStatus status : heartbeat.componentStatus) {
+        if (status.getClusterName().equals(cl.getClusterName())) {
+          try {
+            Service svc = cl.getService(status.getServiceName());
+
+            String componentName = status.getComponentName();
+            if (svc.getServiceComponents().containsKey(componentName)) {
+              ServiceComponent svcComp = svc.getServiceComponent(
+                  componentName);
+              ServiceComponentHost scHost = svcComp.getServiceComponentHost(
+                  hostname);
+              org.apache.ambari.server.state.State prevState = scHost.getState();
+              org.apache.ambari.server.state.State liveState =
+                  org.apache.ambari.server.state.State.valueOf(org.apache.ambari.server.state.State.class,
+                      status.getStatus());
+              if (prevState.equals(org.apache.ambari.server.state.State.INSTALLED)
+                  || prevState.equals(org.apache.ambari.server.state.State.STARTED)
+                  || prevState.equals(org.apache.ambari.server.state.State.STARTING)
+                  || prevState.equals(org.apache.ambari.server.state.State.STOPPING)
+                  || prevState.equals(org.apache.ambari.server.state.State.UNKNOWN)) {
+                scHost.setState(liveState); //TODO direct status set breaks state machine sometimes !!!
+                if (!prevState.equals(liveState)) {
+                  LOG.info("State of service component " + componentName
+                      + " of service " + status.getServiceName()
+                      + " of cluster " + status.getClusterName()
+                      + " has changed from " + prevState + " to " + liveState
+                      + " at host " + hostname);
+                }
+              }
+
+              SecurityState prevSecurityState = scHost.getSecurityState();
+              SecurityState currentSecurityState = SecurityState.valueOf(status.getSecurityState());
+              if((prevSecurityState != currentSecurityState)) {
+                if(prevSecurityState.isEndpoint()) {
+                  scHost.setSecurityState(currentSecurityState);
+                  LOG.info(String.format("Security of service component %s of service %s of cluster %s " +
+                          "has changed from %s to %s on host %s",
+                      componentName, status.getServiceName(), status.getClusterName(), prevSecurityState,
+                      currentSecurityState, hostname));
+                }
+                else {
+                  LOG.debug(String.format("Security of service component %s of service %s of cluster %s " +
+                          "has changed from %s to %s on host %s but will be ignored since %s is a " +
+                          "transitional state",
+                      componentName, status.getServiceName(), status.getClusterName(),
+                      prevSecurityState, currentSecurityState, hostname, prevSecurityState));
+                }
+              }
+
+              if (null != status.getStackVersion() && !status.getStackVersion().isEmpty()) {
+                scHost.setStackVersion(gson.fromJson(status.getStackVersion(), StackId.class));
+              }
+
+              if (null != status.getConfigTags()) {
+                scHost.updateActualConfigs(status.getConfigTags());
+              }
+
+              Map<String, Object> extra = status.getExtra();
+              if (null != extra && !extra.isEmpty()) {
+                try {
+                  if (extra.containsKey("processes")) {
+                    @SuppressWarnings("unchecked")
+                    List<Map<String, String>> list = (List<Map<String, String>>) extra.get("processes");
+                    scHost.setProcesses(list);
+                  }
+                  if (extra.containsKey("version")) {
+                    String version = extra.get("version").toString();
+
+                    handleComponentVersionReceived(cl, scHost, version, false);
+                  }
+
+                } catch (Exception e) {
+                  LOG.error("Could not access extra JSON for " +
+                      scHost.getServiceComponentName() + " from " +
+                      scHost.getHostName() + ": " + status.getExtra() +
+                      " (" + e.getMessage() + ")");
+                }
+              }
+
+              this.heartbeatMonitor.getAgentRequests()
+                  .setExecutionDetailsRequest(hostname, componentName, status.getSendExecCmdDet());
+            } else {
+              // TODO: What should be done otherwise?
+            }
+          } catch (ServiceNotFoundException e) {
+            LOG.warn("Received a live status update for a non-initialized"
+                + " service"
+                + ", clusterName=" + status.getClusterName()
+                + ", serviceName=" + status.getServiceName());
+            // FIXME ignore invalid live update and continue for now?
+            continue;
+          } catch (ServiceComponentNotFoundException e) {
+            LOG.warn("Received a live status update for a non-initialized"
+                + " servicecomponent"
+                + ", clusterName=" + status.getClusterName()
+                + ", serviceName=" + status.getServiceName()
+                + ", componentName=" + status.getComponentName());
+            // FIXME ignore invalid live update and continue for now?
+            continue;
+          } catch (ServiceComponentHostNotFoundException e) {
+            LOG.warn("Received a live status update for a non-initialized"
+                + " service"
+                + ", clusterName=" + status.getClusterName()
+                + ", serviceName=" + status.getServiceName()
+                + ", componentName=" + status.getComponentName()
+                + ", hostname=" + hostname);
+            // FIXME ignore invalid live update and continue for now?
+            continue;
+          } catch (RuntimeException e) {
+            LOG.warn("Received a live status with invalid payload"
+                + " service"
+                + ", clusterName=" + status.getClusterName()
+                + ", serviceName=" + status.getServiceName()
+                + ", componentName=" + status.getComponentName()
+                + ", hostname=" + hostname
+                + ", error=" + e.getMessage());
+            continue;
+          }
+        }
+      }
+    }
+  }
+
+
+
+  /**
+   * Updates the version of the given service component, sets the upgrade state (if needed)
+   * and publishes a version event through the version event publisher.
+   *
+   * @param cluster        the cluster
+   * @param scHost         service component host
+   * @param newVersion     new version of service component
+   * @param alwaysPublish  if true, always publish a version event; if false,
+   *                       only publish if the component version was updated
+   */
+  private void handleComponentVersionReceived(Cluster cluster, ServiceComponentHost scHost,
+                                              String newVersion, boolean alwaysPublish) {
+
+    boolean updated = false;
+
+    if (StringUtils.isNotBlank(newVersion)) {
+      final String previousVersion = scHost.getVersion();
+      if (!StringUtils.equals(previousVersion, newVersion)) {
+        scHost.setVersion(newVersion);
+        scHost.setStackVersion(cluster.getDesiredStackVersion());
+        if (previousVersion != null && !previousVersion.equalsIgnoreCase(
+            org.apache.ambari.server.state.State.UNKNOWN.toString())) {
+          scHost.setUpgradeState(UpgradeState.COMPLETE);
+        }
+        updated = true;
+      }
+    }
+
+    if (updated || alwaysPublish) {
+      HostComponentVersionEvent event = new HostComponentVersionEvent(cluster, scHost);
+      versionEventPublisher.publish(event);
+    }
+  }
+
+  /**
+   * This class is used for mapping json of structured output for keytab distribution actions.
+   */
+  private static class WriteKeytabsStructuredOut {
+    @SerializedName("keytabs")
+    private Map<String,String> keytabs;
+
+    public Map<String, String> getKeytabs() {
+      return keytabs;
+    }
+
+    public void setKeytabs(Map<String, String> keytabs) {
+      this.keytabs = keytabs;
+    }
+  }
+
+
+  /**
+   * This class is used for mapping json of structured output for component START action.
+   */
+  private static class ComponentVersionStructuredOut {
+    @SerializedName("version")
+    private String version;
+
+    @SerializedName("upgrade_type")
+    private UpgradeType upgradeType = null;
+
+    @SerializedName("direction")
+    private Direction upgradeDirection = null;
+
+  }
+}

http://git-wip-us.apache.org/repos/asf/ambari/blob/083ac6da/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostVersionDAO.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostVersionDAO.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostVersionDAO.java
index 040876a..e28f9ef 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostVersionDAO.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostVersionDAO.java
@@ -131,8 +131,8 @@ public class HostVersionDAO extends CrudDAO<HostVersionEntity, Long> {
   }
 
   /**
-   * Retrieve all of the host versions for the given cluster name, host name, and state.
-   *
+   * Retrieve all of the host versions for the given cluster name, host name, and state. <br/>
+   * Consider using faster method: {@link HostVersionDAO#findByClusterHostAndState(long, long, org.apache.ambari.server.state.RepositoryVersionState)}
    * @param clusterName Cluster name
    * @param hostName FQDN of host
    * @param state repository version state
@@ -150,8 +150,29 @@ public class HostVersionDAO extends CrudDAO<HostVersionEntity, Long> {
   }
 
   /**
+   * Faster version of {@link HostVersionDAO#findByClusterHostAndState(java.lang.String, java.lang.String, org.apache.ambari.server.state.RepositoryVersionState)}
+   *
+   * @param clusterId Cluster ID
+   * @param hostId Host ID
+   * @param state repository version state
+   * @return Return all of the host versions that match the criteria.
+   */
+  @RequiresSession
+  public List<HostVersionEntity> findByClusterHostAndState(long clusterId, long hostId, RepositoryVersionState state) {
+    TypedQuery<HostVersionEntity> query =
+        entityManagerProvider.get().createNamedQuery("hostVersionByClusterHostIdAndState", HostVersionEntity.class);
+
+    query.setParameter("clusterId", clusterId);
+    query.setParameter("hostId", hostId);
+    query.setParameter("state", state);
+
+    return daoUtils.selectList(query);
+  }
+
+  /**
    * Retrieve the single host version whose state is {@link org.apache.ambari.server.state.RepositoryVersionState#CURRENT}, of which there should be exactly one at all times
    * for the given host.
+   * Consider using faster method {@link HostVersionDAO#findByHostAndStateCurrent(long, long)}
    *
    * @param clusterName Cluster name
    * @param hostName Host name
@@ -175,8 +196,36 @@ public class HostVersionDAO extends CrudDAO<HostVersionEntity, Long> {
   }
 
   /**
+   * Retrieve the single host version whose state is {@link org.apache.ambari.server.state.RepositoryVersionState#CURRENT}, of which there should be exactly one at all times
+   * for the given host.
+   * Faster version of {@link HostVersionDAO#findByHostAndStateCurrent(java.lang.String, java.lang.String)}
+   * @param clusterId Cluster ID
+   * @param hostId host ID
+   * @return Returns the single host version for this host whose state is {@link org.apache.ambari.server.state.RepositoryVersionState#CURRENT}, or {@code null} otherwise.
+   */
+  @RequiresSession
+  public HostVersionEntity findByHostAndStateCurrent(long clusterId, long hostId) {
+    try {
+      List<?> results = findByClusterHostAndState(clusterId, hostId, RepositoryVersionState.CURRENT);
+      if (results.isEmpty()) {
+        return null;
+      } else {
+        if (results.size() == 1) {
+          return (HostVersionEntity) results.get(0);
+        }
+      }
+      throw new NonUniqueResultException();
+    } catch (NoResultException ignored) {
+      return null;
+    }
+  }
+
+  /**
    * Retrieve the single host version for the given cluster, stack name, stack
-   * version, and host name.
+   * version, and host name. <br/>
+   * This query is slow and not suitable for frequent use. <br/>
+   * Please, use {@link HostVersionDAO#findByClusterStackVersionAndHost(long, org.apache.ambari.server.state.StackId, java.lang.String, long)} <br/>
+   * It is ~50 times faster
    *
    * @param clusterName
    *          Cluster name
@@ -203,6 +252,29 @@ public class HostVersionDAO extends CrudDAO<HostVersionEntity, Long> {
     return daoUtils.selectSingle(query);
   }
 
+  /**
+   * Optimized version of {@link HostVersionDAO#findByClusterStackVersionAndHost(java.lang.String, org.apache.ambari.server.state.StackId, java.lang.String, java.lang.String)}
+   * @param clusterId Id of cluster
+   * @param stackId Stack ID (e.g., HDP-2.2)
+   * @param version Stack version (e.g., 2.2.0.1-995)
+   * @param hostId Host Id
+   * @return Returns the single host version that matches the criteria.
+   */
+  @RequiresSession
+  public HostVersionEntity findByClusterStackVersionAndHost(long clusterId, StackId stackId, String version,
+                                                            long hostId) {
+    TypedQuery<HostVersionEntity> query = entityManagerProvider.get()
+        .createNamedQuery("hostVersionByClusterStackVersionAndHostId", HostVersionEntity.class);
+
+    query.setParameter("clusterId", clusterId);
+    query.setParameter("stackName", stackId.getStackName());
+    query.setParameter("stackVersion", stackId.getStackVersion());
+    query.setParameter("version", version);
+    query.setParameter("hostId", hostId);
+
+    return daoUtils.selectSingle(query);
+  }
+
   @Transactional
   public void removeByHostName(String hostName) {
     Collection<HostVersionEntity> hostVersions = this.findByHost(hostName);

http://git-wip-us.apache.org/repos/asf/ambari/blob/083ac6da/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/HostVersionEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/HostVersionEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/HostVersionEntity.java
index b69518b..6be4b50 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/HostVersionEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/HostVersionEntity.java
@@ -62,6 +62,15 @@ import org.apache.ambari.server.state.RepositoryVersionState;
         "SELECT hostVersion FROM HostVersionEntity hostVersion JOIN hostVersion.hostEntity host JOIN host.clusterEntities clusters " +
             "WHERE clusters.clusterName=:clusterName AND hostVersion.repositoryVersion.stack.stackName=:stackName AND hostVersion.repositoryVersion.stack.stackVersion=:stackVersion AND hostVersion.repositoryVersion.version=:version AND " +
             "hostVersion.hostEntity.hostName=:hostName"),
+
+    @NamedQuery(name = "hostVersionByClusterHostIdAndState", query =
+        "SELECT hostVersion FROM HostVersionEntity hostVersion JOIN hostVersion.hostEntity host JOIN host.clusterEntities clusters " +
+            "WHERE clusters.clusterId=:clusterId AND hostVersion.hostId=:hostId AND hostVersion.state=:state"),
+
+    @NamedQuery(name = "hostVersionByClusterStackVersionAndHostId", query =
+        "SELECT hostVersion FROM HostVersionEntity hostVersion JOIN hostVersion.hostEntity host JOIN host.clusterEntities clusters " +
+        "WHERE hostVersion.hostId=:hostId AND clusters.clusterId=:clusterId AND hostVersion.repositoryVersion.stack.stackName=:stackName " +
+        "AND hostVersion.repositoryVersion.stack.stackVersion=:stackVersion AND hostVersion.repositoryVersion.version=:version")
 })
 public class HostVersionEntity {
 

http://git-wip-us.apache.org/repos/asf/ambari/blob/083ac6da/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
index 4212975..c6d01e8 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
@@ -1560,8 +1560,8 @@ public class ClusterImpl implements Cluster {
     StackId repoVersionStackId = new StackId(repoVersionStackEntity);
 
     HostVersionEntity hostVersionEntity = hostVersionDAO.findByClusterStackVersionAndHost(
-      getClusterName(), repoVersionStackId, repositoryVersion.getVersion(),
-      host.getHostName());
+      getClusterId(), repoVersionStackId, repositoryVersion.getVersion(),
+      host.getHostId());
 
     hostTransitionStateWriteLock.lock();
     try {
@@ -1576,7 +1576,7 @@ public class ClusterImpl implements Cluster {
         hostVersionDAO.create(hostVersionEntity);
       }
 
-      HostVersionEntity currentVersionEntity = hostVersionDAO.findByHostAndStateCurrent(getClusterName(), host.getHostName());
+      HostVersionEntity currentVersionEntity = hostVersionDAO.findByHostAndStateCurrent(getClusterId(), host.getHostId());
       boolean isCurrentPresent = (currentVersionEntity != null);
       final ServiceComponentHostSummary hostSummary = new ServiceComponentHostSummary(ambariMetaInfo, host, stack);
 

http://git-wip-us.apache.org/repos/asf/ambari/blob/083ac6da/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
index bfb6214..1bd60a8 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
@@ -1319,43 +1319,48 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
 
   @Override
   public ServiceComponentHostResponse convertToResponse() {
-    readLock.lock();
+    clusterGlobalLock.readLock().lock();
     try {
-      HostComponentStateEntity hostComponentStateEntity = getStateEntity();
-      if (null == hostComponentStateEntity) {
-        LOG.warn("Could not convert ServiceComponentHostResponse to a response. It's possible that Host " + getHostName() + " was deleted.");
-        return null;
-      }
+      readLock.lock();
+      try {
+        HostComponentStateEntity hostComponentStateEntity = getStateEntity();
+        if (null == hostComponentStateEntity) {
+          LOG.warn("Could not convert ServiceComponentHostResponse to a response. It's possible that Host " + getHostName() + " was deleted.");
+          return null;
+        }
 
-      String clusterName = serviceComponent.getClusterName();
-      String serviceName = serviceComponent.getServiceName();
-      String serviceComponentName = serviceComponent.getName();
-      String hostName = getHostName();
-      String state = getState().toString();
-      String stackId = getStackVersion().getStackId();
-      String desiredState = getDesiredState().toString();
-      String desiredStackId = getDesiredStackVersion().getStackId();
-      HostComponentAdminState componentAdminState = getComponentAdminState();
-      UpgradeState upgradeState = hostComponentStateEntity.getUpgradeState();
-
-      ServiceComponentHostResponse r = new ServiceComponentHostResponse(
-          clusterName, serviceName,
-          serviceComponentName, hostName, state,
-          stackId, desiredState,
-          desiredStackId, componentAdminState);
-
-      r.setActualConfigs(actualConfigs);
-      r.setUpgradeState(upgradeState);
+        String clusterName = serviceComponent.getClusterName();
+        String serviceName = serviceComponent.getServiceName();
+        String serviceComponentName = serviceComponent.getName();
+        String hostName = getHostName();
+        String state = getState().toString();
+        String stackId = getStackVersion().getStackId();
+        String desiredState = getDesiredState().toString();
+        String desiredStackId = getDesiredStackVersion().getStackId();
+        HostComponentAdminState componentAdminState = getComponentAdminState();
+        UpgradeState upgradeState = hostComponentStateEntity.getUpgradeState();
+
+        ServiceComponentHostResponse r = new ServiceComponentHostResponse(
+            clusterName, serviceName,
+            serviceComponentName, hostName, state,
+            stackId, desiredState,
+            desiredStackId, componentAdminState);
+
+        r.setActualConfigs(actualConfigs);
+        r.setUpgradeState(upgradeState);
 
-      try {
-        r.setStaleConfig(helper.isStaleConfigs(this));
-      } catch (Exception e) {
-        LOG.error("Could not determine stale config", e);
-      }
+        try {
+          r.setStaleConfig(helper.isStaleConfigs(this));
+        } catch (Exception e) {
+          LOG.error("Could not determine stale config", e);
+        }
 
-      return r;
+        return r;
+      } finally {
+        readLock.unlock();
+      }
     } finally {
-      readLock.unlock();
+      clusterGlobalLock.readLock().unlock();
     }
   }
 
@@ -1797,6 +1802,7 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
     }
 
     final String hostName = getHostName();
+    final long hostId = getHost().getHostId();
     final Set<Cluster> clustersForHost = clusters.getClustersForHost(hostName);
     if (clustersForHost.size() != 1) {
       throw new AmbariException("Host " + hostName + " should be assigned only to one cluster");
@@ -1815,7 +1821,7 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
         repositoryVersion = createRepositoryVersion(version, stackId, stackInfo);
       }
 
-      final HostEntity host = hostDAO.findByName(hostName);
+      final HostEntity host = hostDAO.findById(hostId);
       cluster.transitionHostVersionState(host, repositoryVersion, stackId);
     } finally {
       writeLock.unlock();


Mime
View raw message