AMBARI-1447. Report current Stack version for all host components. (Sumit Mohanty via mahadev)

git-svn-id: https://svn.apache.org/repos/asf/incubator/ambari/trunk@1449806 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -56,6 +56,9 @@ Trunk (unreleased changes):
  AMBARI-1267. Store example Hive Queries somewhere in Ambari that's easily
  accessible for demo/test purposes. (mahadev)
 
+ AMBARI-1447. Report current Stack version for all host components.
+ (Sumit Mohanty via mahadev)
+
  IMPROVEMENTS
 
  AMBARI-1477. Improve performance for App.statusMapper. (yusaku)

ambari-server/src/main/java/org/apache/ambari/server/agent/ComponentStatus.java

@@ -25,6 +25,7 @@ public class ComponentStatus {
   String status;
   String serviceName;
   String clusterName;
+  String stackVersion;
 
   public String getComponentName() {
     return this.componentName;
@@ -50,6 +51,14 @@ public class ComponentStatus {
     this.status = status;
   }
 
+  public String getStackVersion() {
+    return this.stackVersion;
+  }
+
+  public void setStackVersion(String stackVersion) {
+    this.stackVersion = stackVersion;
+  }
+
   public String getMsg() {
     return msg;
   }
@@ -82,6 +91,7 @@ public class ComponentStatus {
             ", status='" + status + '\'' +
             ", serviceName='" + serviceName + '\'' +
             ", clusterName='" + clusterName + '\'' +
+            ", stackVersion='" + stackVersion + '\'' +
             '}';
   }
 }

ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java

@@ -233,6 +233,12 @@ public class HeartBeatHandler {
                       + " at host " + hostname);
                 }
               }
+
+              if(null != status.getStackVersion() && !status.getStackVersion().isEmpty())
+              {
+                scHost.setStackVersion(new StackId(status.getStackVersion()));
+              }
+
               // TODO need to get config version and stack version from live state
             } else {
               // TODO: What should be done otherwise?
@@ -265,6 +271,16 @@ public class HeartBeatHandler {
             // 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;
+          }
         }
       }
     }

ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java

@@ -195,8 +195,9 @@ public class AmbariManagementControllerImpl implements
     clusters.addCluster(request.getClusterName());
     Cluster c = clusters.getCluster(request.getClusterName());
     if (request.getStackVersion() != null) {
-      c.setDesiredStackVersion(
-          new StackId(request.getStackVersion()));
+      StackId newStackId = new StackId(request.getStackVersion());
+      c.setDesiredStackVersion(newStackId);
+      clusters.setCurrentStackVersion(request.getClusterName(), newStackId);
     }
 
     if (request.getHostNames() != null) {

ambari-server/src/main/java/org/apache/ambari/server/orm/dao/ClusterStateDAO.java

@@ -30,8 +30,8 @@ public class ClusterStateDAO {
   Provider<EntityManager> entityManagerProvider;
 
   @Transactional
-  public ClusterStateEntity findByPK(String clusterName) {
-    return entityManagerProvider.get().find(ClusterStateEntity.class, clusterName);
+  public ClusterStateEntity findByPK(long clusterId) {
+    return entityManagerProvider.get().find(ClusterStateEntity.class, clusterId);
   }
 
   @Transactional
@@ -55,8 +55,8 @@ public class ClusterStateDAO {
   }
 
   @Transactional
-  public void removeByPK(String clusterName) {
-    remove(findByPK(clusterName));
+  public void removeByPK(long clusterId) {
+    remove(findByPK(clusterId));
   }
 
 }

ambari-server/src/main/java/org/apache/ambari/server/orm/dao/ExecutionCommandDAO.java

@@ -51,7 +51,7 @@ public class ExecutionCommandDAO {
   }
 
   @Transactional
-  public void removeByPK(int taskId) {
+  public void removeByPK(long taskId) {
     remove(findByPK(taskId));
   }
 }

ambari-server/src/main/java/org/apache/ambari/server/orm/entities/ClusterStateEntity.java

@@ -27,6 +27,9 @@ import javax.persistence.OneToOne;
 @Entity
 public class ClusterStateEntity {
   private Long clusterId;
+  private String currentClusterState = "";
+  private String currentStackVersion = "";
+  private ClusterEntity clusterEntity;
 
   @javax.persistence.Column(name = "cluster_id", nullable = false, insertable = false, updatable = false, length = 10)
   @Id
@@ -38,8 +41,6 @@ public class ClusterStateEntity {
     this.clusterId = clusterId;
   }
 
-  private String currentClusterState = "";
-
   @javax.persistence.Column(name = "current_cluster_state", nullable = false, insertable = true, updatable = true)
   @Basic
   public String getCurrentClusterState() {
@@ -50,7 +51,17 @@ public class ClusterStateEntity {
     this.currentClusterState = currentClusterState;
   }
 
-  @Override
+  @javax.persistence.Column(name = "current_stack_version", nullable = false, insertable = true, updatable = true)
+  @Basic
+  public String getCurrentStackVersion() {
+    return currentStackVersion;
+  }
+
+  public void setCurrentStackVersion(String currentStackVersion) {
+    this.currentStackVersion = currentStackVersion;
+  }
+
+    @Override
   public boolean equals(Object o) {
     if (this == o) return true;
     if (o == null || getClass() != o.getClass()) return false;
@@ -58,8 +69,10 @@ public class ClusterStateEntity {
     ClusterStateEntity that = (ClusterStateEntity) o;
 
     if (clusterId != null ? !clusterId.equals(that.clusterId) : that.clusterId != null) return false;
-    if (currentClusterState != null ? !currentClusterState.equals(that.currentClusterState) : that.currentClusterState != null)
-      return false;
+    if (currentClusterState != null
+        ? !currentClusterState.equals(that.currentClusterState) : that.currentClusterState != null) return false;
+    if (currentStackVersion != null
+        ? !currentStackVersion.equals(that.currentStackVersion) : that.currentStackVersion != null) return false;
 
     return true;
   }
@@ -71,8 +84,6 @@ public class ClusterStateEntity {
     return result;
   }
 
-  private ClusterEntity clusterEntity;
-
   @OneToOne
   @javax.persistence.JoinColumn(name = "cluster_id", referencedColumnName = "cluster_id", nullable = false)
   public ClusterEntity getClusterEntity() {

ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java

@@ -69,38 +69,104 @@ public interface Cluster {
   public List<ServiceComponentHost> getServiceComponentHosts(String hostname);
 
   /**
-   * Get Stack Version
+   * Get desired stack version
    * @return
    */
   public StackId getDesiredStackVersion();
 
   /**
-   * Set stack version
+   * Set desired stack version
    * @param stackVersion
    */
   public void setDesiredStackVersion(StackId stackVersion);
 
+  /**
+   * Get current stack version
+   * @return
+   */
+  public StackId getCurrentStackVersion();
+
+  /**
+   * Set current stack version
+   * @param stackVersion
+   */
+  public void setCurrentStackVersion(StackId stackVersion) throws AmbariException;
+
+  /**
+   * Get desired config based on the given config type name
+   * @param configType
+   * @return
+   */
   public Map<String, Config> getDesiredConfigsByType(String configType);
 
+  /**
+   * Get the desired config of given type and version
+   * @param configType
+   * @param versionTag
+   * @return
+   */
   public Config getDesiredConfig(String configType, String versionTag);
 
+  /**
+   * Add the desired config for the cluster
+   * @param config
+   */
   public void addDesiredConfig(Config config);
 
+  /**
+   * Get all configs associated with the cluster
+   * @return
+   */
   public Collection<Config> getAllConfigs();
 
+  /**
+   * Creates a cluster response based on the current cluster definition
+   * @return
+   * @throws AmbariException
+   */
   public ClusterResponse convertToResponse() throws AmbariException;
 
+  /**
+   * Refreshes the cluster details
+   */
   public void refresh();
 
+  /**
+   * Creates a debug dump based on the current cluster state
+   * @param sb
+   */
   public void debugDump(StringBuilder sb);
 
-  Service addService(String serviceName) throws AmbariException;
-
+  /**
+   * Delete all the services associated with this cluster
+   * @throws AmbariException
+   */
   public void deleteAllServices() throws AmbariException;
 
+  /**
+   * Delete the named service associated with this cluster
+   * @param serviceName
+   * @throws AmbariException
+   */
   public void deleteService(String serviceName) throws AmbariException;
 
+  /**
+   * Gets if the cluster can be deleted
+   * @return
+   */
   public boolean canBeRemoved();
 
+  /**
+   * Delete the cluster
+   * @throws AmbariException
+   */
   public void delete() throws AmbariException;
+
+  /**
+   * Add service to the cluster
+   * @param serviceName
+   * @return
+   * @throws AmbariException
+   */
+  Service addService(String serviceName) throws AmbariException;
 }

ambari-server/src/main/java/org/apache/ambari/server/state/Clusters.java

@@ -34,29 +34,31 @@ public interface Clusters {
    * Add a new Cluster
    * @param clusterName
    */
-  public void addCluster(String clusterName) throws AmbariException;
+  public void addCluster(String clusterName)
+      throws AmbariException;
 
   /**
-   * Get the Cluster given the cluster name
+   * Gets the Cluster given the cluster name
    * @param clusterName Name of the Cluster to retrieve
-   * @return
+   * @return  <code>Cluster</code> identified by the given name
    */
-  public Cluster getCluster(String clusterName) throws AmbariException;
+  public Cluster getCluster(String clusterName)
+      throws AmbariException;
 
   /**
-   * Get all known clusters
-   * @return
+   * Get all clusters
+   * @return <code>Map</code> of clusters with cluster name as key
    */
   public Map<String, Cluster> getClusters();
 
   /**
    * Get all hosts being tracked by the Ambari server
-   * @return
+   * @return <code>List</code> of <code>Host</code>
    */
   public List<Host> getHosts();
 
   /**
-   * Returns all the cluster names for this hostname.
+   * Returns all the cluster names for this hostname
    * @param hostname
    * @return List of cluster names
    * @throws AmbariException
@@ -82,7 +84,7 @@ public interface Clusters {
 
   /**
    * Map host to the given cluster.
-   * A host can belong to multiple clusters.
+   * A host can belong to multiple clusters
    * @param hostname
    * @param clusterName
    * @throws AmbariException
@@ -90,22 +92,70 @@ public interface Clusters {
   public void mapHostToCluster(String hostname, String clusterName)
       throws AmbariException;
 
-
+  /**
+   * Maps a set of hosts to the given cluster
+   * @param hostnames
+   * @param clusterName
+   * @throws AmbariException
+   */
   public void mapHostsToCluster(Set<String> hostnames, String clusterName)
       throws AmbariException;
 
+  /**
+   * Updates the name of the cluster
+   * @param oldName
+   * @param newName
+   * @throws AmbariException
+   */
   public void updateClusterName(String oldName, String newName);
 
+  /**
+   * Gets the cluster using the id.
+   * @param id The identifier associated with the cluster
+   * @return <code>Cluster</code> identified by the identifier
+   * @throws AmbariException
+   */
   public Cluster getClusterById(long id) throws AmbariException;
 
+  /**
+   * Produces a debug dump into the supplied string buffer
+   * @param sb The string buffer to add the debug dump to
+   */
   public void debugDump(StringBuilder sb);
 
+  /**
+   * Gets all the hosts associated with the cluster
+   * @param clusterName The name of the cluster
+   * @return <code>Map</code> containing host name and <code>Host</code>
+   * @throws AmbariException
+   */
   public Map<String, Host> getHostsForCluster(String clusterName)
       throws AmbariException;
 
-  public void deleteCluster(String clusterName) throws AmbariException;
+  /**
+   * Deletes the cluster identified by the name
+   * @param clusterName The name of the cluster
+   * @throws AmbariException
+   */
+  public void deleteCluster(String clusterName)
+      throws AmbariException;
 
-  public void updateHostWithClusterAndAttributes(Map<String, Set<String>> hostsClusters, Map<String,
-    Map<String, String>> hostAttributes)
-    throws AmbariException;
+  /**
+   * Sets the current stack version for the cluster
+   * @param clusterName The name of the cluster
+   * @param stackId The identifier for the stack
+   * @throws AmbariException
+   */
+  public void setCurrentStackVersion(String clusterName, StackId stackId)
+      throws AmbariException;
+
+  /**
+   * Update the host set for clusters and the host attributes associated with the hosts
+   * @param hostsClusters
+   * @param hostAttributes
+   * @throws AmbariException
+   */
+  public void updateHostWithClusterAndAttributes(
+      Map<String, Set<String>> hostsClusters, Map<String, Map<String, String>> hostAttributes)
+      throws AmbariException;
 }

ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java

@@ -37,9 +37,11 @@ import org.apache.ambari.server.ServiceComponentHostNotFoundException;
 import org.apache.ambari.server.ServiceNotFoundException;
 import org.apache.ambari.server.controller.ClusterResponse;
 import org.apache.ambari.server.orm.dao.ClusterDAO;
+import org.apache.ambari.server.orm.dao.ClusterStateDAO;
 import org.apache.ambari.server.orm.entities.ClusterConfigEntity;
 import org.apache.ambari.server.orm.entities.ClusterEntity;
 import org.apache.ambari.server.orm.entities.ClusterServiceEntity;
+import org.apache.ambari.server.orm.entities.ClusterStateEntity;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.Config;
@@ -58,6 +60,8 @@ import com.google.inject.Injector;
 import com.google.inject.assistedinject.Assisted;
 import com.google.inject.persist.Transactional;
 
+import javax.persistence.RollbackException;
+
 public class ClusterImpl implements Cluster {
 
   private static final Logger LOG =
@@ -95,6 +99,8 @@ public class ClusterImpl implements Cluster {
 
   @Inject
   private ClusterDAO clusterDAO;
+  @Inject
+  private ClusterStateDAO clusterStateDAO;
 //  @Inject
 //  private ClusterServiceDAO clusterServiceDAO;
   @Inject
@@ -190,7 +196,7 @@ public class ClusterImpl implements Cluster {
   }
 
   private void loadServices() {
-    LOG.info("clusterEntity " + clusterEntity.getClusterServiceEntities() );
+    LOG.info("clusterEntity " + clusterEntity.getClusterServiceEntities());
     if (services == null) {
       writeLock.lock();
       try {
@@ -450,38 +456,50 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.writeLock().unlock();
     }
-
   }
 
-  public StackId getDesiredState() {
-    //TODO separate implementation, mapped to StackVersion for now
-//    return desiredState; for separate implementation
-    readWriteLock.readLock().lock();
-    try {
-      return getDesiredStackVersion();
-    } finally {
-      readWriteLock.readLock().unlock();
+  @Override
+  public StackId getCurrentStackVersion() {
+    ClusterStateEntity clusterStateEntity = clusterEntity.getClusterStateEntity();
+    if(clusterStateEntity != null)
+    {
+      String stackVersion = clusterStateEntity.getCurrentStackVersion();
+      if(stackVersion != null && !stackVersion.isEmpty())
+      {
+        return gson.fromJson(stackVersion, StackId.class);
+      }
     }
-
+    return null;
   }
 
-  public void setDesiredState(StackId desiredState) {
-    //TODO separate implementation, mapped to StackVersion for now
-//    LOG.debug("Changing desired state of cluster, clusterName={}, clusterId={}, oldState={}, newState={}",
-//        getClusterName(), getClusterId(), this.desiredState, desiredState);
-//    clusterEntity.setDesiredClusterState(gson.toJson(desiredState));
-//    clusterDAO.merge(clusterEntity);
-//    this.desiredState = desiredState;
-    readWriteLock.writeLock().lock();
+  @Override
+  public void setCurrentStackVersion(StackId stackVersion)
+  throws AmbariException {
+    writeLock.lock();
     try {
-      setDesiredStackVersion(desiredState);
+        ClusterStateEntity clusterStateEntity = clusterStateDAO.findByPK(clusterEntity.getClusterId());
+        if (clusterStateEntity == null) {
+          clusterStateEntity = new ClusterStateEntity();
+          clusterStateEntity.setClusterId(clusterEntity.getClusterId());
+          clusterStateEntity.setCurrentStackVersion(gson.toJson(stackVersion));
+          clusterStateEntity.setClusterEntity(clusterEntity);
+          clusterStateDAO.create(clusterStateEntity);
+          clusterStateEntity = clusterStateDAO.merge(clusterStateEntity);
+          clusterEntity.setClusterStateEntity(clusterStateEntity);
+          clusterEntity = clusterDAO.merge(clusterEntity);
+        } else {
+          clusterStateEntity.setCurrentStackVersion(gson.toJson(stackVersion));
+        }
+    } catch (RollbackException e) {
+      LOG.warn("Unable to set version " + stackVersion + " for cluster " + getClusterName());
+      throw new AmbariException("Unable to set"
+          + " version=" + stackVersion
+          + " for cluster " + getClusterName(), e);
     } finally {
-      readWriteLock.writeLock().unlock();
+      writeLock.unlock();
     }
-
   }
 
-
   @Override
   public Map<String, Config> getDesiredConfigsByType(String configType) {
     readWriteLock.writeLock().lock();
@@ -493,7 +511,6 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.writeLock().unlock();
     }
-
   }
 
   @Override
@@ -508,7 +525,6 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.readLock().unlock();
     }
-
   }
 
   @Override
@@ -529,7 +545,6 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.writeLock().unlock();
     }
-
   }
 
   public Collection<Config> getAllConfigs() {
@@ -545,7 +560,6 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.readLock().unlock();
     }
-
   }
 
   @Override
@@ -560,7 +574,6 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.readLock().unlock();
     }
-
   }
 
   public void debugDump(StringBuilder sb) {
@@ -585,7 +598,6 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.readLock().unlock();
     }
-
   }
 
   @Override
@@ -598,7 +610,6 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.writeLock().unlock();
     }
-
   }
 
   @Override
@@ -626,7 +637,6 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.writeLock().unlock();
     }
-
   }
 
   @Override
@@ -650,7 +660,6 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.writeLock().unlock();
     }
-
   }
 
   @Override
@@ -671,7 +680,6 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.readLock().unlock();
     }
-
   }
 
   @Override
@@ -685,7 +693,6 @@ public class ClusterImpl implements Cluster {
     } finally {
       readWriteLock.writeLock().unlock();
     }
-
   }
 
   @Transactional

ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClustersImpl.java

@@ -188,6 +188,29 @@ public class ClustersImpl implements Clusters {
     }
   }
 
+  @Override
+  public void setCurrentStackVersion(String clusterName, StackId stackId)
+      throws AmbariException{
+    if(stackId == null || clusterName == null || clusterName.isEmpty()){
+      LOG.warn("Unable to set version for cluster " + clusterName);
+      throw new AmbariException("Unable to set"
+          + " version=" + stackId
+          + " for cluster " + clusterName);
+    }
+
+    loadClustersAndHosts();
+    r.lock();
+    try {
+      if (!clusters.containsKey(clusterName)) {
+        throw new ClusterNotFoundException(clusterName);
+      }
+      Cluster cluster = clusters.get(clusterName);
+      cluster.setCurrentStackVersion(stackId);
+    } finally {
+      r.unlock();
+    }
+  }
+
   @Override
   @Transactional
   public List<Host> getHosts() {
@@ -521,7 +544,7 @@ public class ClustersImpl implements Clusters {
   }
 
   @Override
-  public synchronized void deleteCluster(String clusterName)
+  public void deleteCluster(String clusterName)
       throws AmbariException {
     loadClustersAndHosts();
     w.lock();

ambari-server/src/main/resources/Ambari-DDL-Postgres-CREATE.sql

@@ -36,7 +36,7 @@ CREATE TABLE ambari.clusterservices (service_name VARCHAR(255) NOT NULL, cluster
 
 GRANT ALL PRIVILEGES ON TABLE ambari.clusterservices TO :username;
 
-CREATE TABLE ambari.clusterstate (cluster_id BIGINT NOT NULL, current_cluster_state VARCHAR(255) NOT NULL, PRIMARY KEY (cluster_id));
+CREATE TABLE ambari.clusterstate (cluster_id BIGINT NOT NULL, current_cluster_state VARCHAR(255) NOT NULL, current_stack_version VARCHAR(255) NOT NULL, PRIMARY KEY (cluster_id));
 
 GRANT ALL PRIVILEGES ON TABLE ambari.clusterstate TO :username;
 

ambari-server/src/main/resources/Ambari-DDL.sql

@@ -173,6 +173,7 @@ CREATE TABLE ClusterState
 (
 cluster_id BIGINT NOT NULL references Clusters(cluster_id),
 current_cluster_state VARCHAR DEFAULT '' NOT NULL,
+current_stack_version VARCHAR DEFAULT '' NOT NULL,
 PRIMARY KEY (cluster_id)
 );
 

ambari-server/src/test/java/org/apache/ambari/server/agent/DummyHeartbeatConstants.java

@@ -38,5 +38,6 @@ public interface DummyHeartbeatConstants {
   String NAMENODE = Role.NAMENODE.name();
   String SECONDARY_NAMENODE = Role.SECONDARY_NAMENODE.name();
   String HBASE_MASTER = Role.HBASE_MASTER.name();
+  String HDFS_CLIENT  = Role.HDFS_CLIENT.name();
 
 }

ambari-server/src/test/java/org/apache/ambari/server/agent/TestHeartbeatHandler.java

@@ -28,8 +28,10 @@ import static org.apache.ambari.server.agent.DummyHeartbeatConstants.DummyStackI
 import static org.apache.ambari.server.agent.DummyHeartbeatConstants.HBASE;
 import static org.apache.ambari.server.agent.DummyHeartbeatConstants.HBASE_MASTER;
 import static org.apache.ambari.server.agent.DummyHeartbeatConstants.HDFS;
+import static org.apache.ambari.server.agent.DummyHeartbeatConstants.HDFS_CLIENT;
 import static org.apache.ambari.server.agent.DummyHeartbeatConstants.NAMENODE;
 import static org.apache.ambari.server.agent.DummyHeartbeatConstants.SECONDARY_NAMENODE;
+
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -175,13 +177,7 @@ public class TestHeartbeatHandler {
     ActionManager am = new ActionManager(0, 0, null, null,
             new ActionDBInMemoryImpl(), new HostsMap((String) null));
 
-    clusters.addHost(DummyHostname1);
-    clusters.getHost(DummyHostname1).setOsType(DummyOsType);
-    clusters.getHost(DummyHostname1).persist();
-    clusters.addCluster(DummyCluster);
-
-    Cluster cluster = clusters.getCluster(DummyCluster);
-    cluster.setDesiredStackVersion(new StackId(DummyStackId));
+    Cluster cluster = getDummyCluster();
 
     @SuppressWarnings("serial")
     Set<String> hostNames = new HashSet<String>(){{
@@ -198,17 +194,7 @@ public class TestHeartbeatHandler {
     hdfs.getServiceComponent(SECONDARY_NAMENODE).addServiceComponentHost(DummyHostname1).persist();
 
     ActionQueue aq = new ActionQueue();
-    HeartBeatHandler handler = new HeartBeatHandler(clusters, aq, am, injector);
-
-    Register reg = new Register();
-    HostInfo hi = new HostInfo();
-    hi.setHostName(DummyHostname1);
-    hi.setOS(DummyOs);
-    hi.setOSRelease(DummyOSRelease);
-    reg.setHostname(DummyHostname1);
-    reg.setResponseId(0);
-    reg.setHardwareProfile(hi);
-    handler.handleRegistration(reg);
+    HeartBeatHandler handler = getHeartBeatHandler(am, aq);
 
     ServiceComponentHost serviceComponentHost1 = clusters.getCluster(DummyCluster).getService(HDFS).
             getServiceComponent(DATANODE).getServiceComponentHost(DummyHostname1);
@@ -245,13 +231,7 @@ public class TestHeartbeatHandler {
   public void testLiveStatusUpdateAfterStopFailed() throws Exception {
     ActionManager am = new ActionManager(0, 0, null, null,
             new ActionDBInMemoryImpl(), new HostsMap((String) null));
-    clusters.addHost(DummyHostname1);
-    clusters.getHost(DummyHostname1).setOsType(DummyOsType);
-    clusters.getHost(DummyHostname1).persist();
-    clusters.addCluster(DummyCluster);
-
-    Cluster cluster = clusters.getCluster(DummyCluster);
-    cluster.setDesiredStackVersion(new StackId(DummyStackId));
+    Cluster cluster = getDummyCluster();
 
     @SuppressWarnings("serial")
     Set<String> hostNames = new HashSet<String>(){{
@@ -268,17 +248,7 @@ public class TestHeartbeatHandler {
             addServiceComponentHost(DummyHostname1).persist();
 
     ActionQueue aq = new ActionQueue();
-    HeartBeatHandler handler = new HeartBeatHandler(clusters, aq, am, injector);
-
-    Register reg = new Register();
-    HostInfo hi = new HostInfo();
-    hi.setHostName(DummyHostname1);
-    hi.setOS(DummyOs);
-    hi.setOSRelease(DummyOSRelease);
-    reg.setHostname(DummyHostname1);
-    reg.setResponseId(0);
-    reg.setHardwareProfile(hi);
-    handler.handleRegistration(reg);
+    HeartBeatHandler handler = getHeartBeatHandler(am, aq);
 
     ServiceComponentHost serviceComponentHost1 = clusters.
             getCluster(DummyCluster).getService(HDFS).
@@ -580,13 +550,7 @@ public class TestHeartbeatHandler {
   public void testTaskInProgressHandling() throws AmbariException, InvalidStateTransitionException {
     ActionManager am = new ActionManager(0, 0, null, null,
             new ActionDBInMemoryImpl(), new HostsMap((String) null));
-    clusters.addHost(DummyHostname1);
-    clusters.getHost(DummyHostname1).setOsType(DummyOsType);
-    clusters.getHost(DummyHostname1).persist();
-    clusters.addCluster(DummyCluster);
-
-    Cluster cluster = clusters.getCluster(DummyCluster);
-    cluster.setDesiredStackVersion(new StackId(DummyStackId));
+    Cluster cluster = getDummyCluster();
 
     @SuppressWarnings("serial")
     Set<String> hostNames = new HashSet<String>(){{
@@ -603,17 +567,7 @@ public class TestHeartbeatHandler {
     hdfs.getServiceComponent(SECONDARY_NAMENODE).addServiceComponentHost(DummyHostname1).persist();
 
     ActionQueue aq = new ActionQueue();
-    HeartBeatHandler handler = new HeartBeatHandler(clusters, aq, am, injector);
-
-    Register reg = new Register();
-    HostInfo hi = new HostInfo();
-    hi.setHostName(DummyHostname1);
-    hi.setOS(DummyOs);
-    hi.setOSRelease(DummyOSRelease);
-    reg.setHostname(DummyHostname1);
-    reg.setResponseId(0);
-    reg.setHardwareProfile(hi);
-    handler.handleRegistration(reg);
+    HeartBeatHandler handler = getHeartBeatHandler(am, aq);
 
     ServiceComponentHost serviceComponentHost1 = clusters.getCluster(DummyCluster).getService(HDFS).
             getServiceComponent(DATANODE).getServiceComponentHost(DummyHostname1);
@@ -645,4 +599,113 @@ public class TestHeartbeatHandler {
     State componentState1 = serviceComponentHost1.getState();
     assertEquals("Host state should still be installing", State.INSTALLING, componentState1);
   }
+
+  @Test
+  public void testStatusHeartbeatWithVersion() throws Exception {
+    ActionManager am = new ActionManager(0, 0, null, null,
+        new ActionDBInMemoryImpl(), new HostsMap((String) null));
+    Cluster cluster = getDummyCluster();
+
+    @SuppressWarnings("serial")
+    Set<String> hostNames = new HashSet<String>(){{
+      add(DummyHostname1);
+    }};
+
+    clusters.mapHostsToCluster(hostNames, DummyCluster);
+
+    Service hdfs = cluster.addService(HDFS);
+    hdfs.persist();
+    hdfs.addServiceComponent(DATANODE).persist();
+    hdfs.getServiceComponent(DATANODE).addServiceComponentHost(DummyHostname1).persist();
+    hdfs.addServiceComponent(NAMENODE).persist();
+    hdfs.getServiceComponent(NAMENODE).addServiceComponentHost(DummyHostname1).persist();
+    hdfs.addServiceComponent(HDFS_CLIENT).persist();
+    hdfs.getServiceComponent(HDFS_CLIENT).addServiceComponentHost(DummyHostname1).persist();
+
+    ServiceComponentHost serviceComponentHost1 = clusters.getCluster(DummyCluster).getService(HDFS).
+        getServiceComponent(DATANODE).getServiceComponentHost(DummyHostname1);
+    ServiceComponentHost serviceComponentHost2 = clusters.getCluster(DummyCluster).getService(HDFS).
+        getServiceComponent(NAMENODE).getServiceComponentHost(DummyHostname1);
+    ServiceComponentHost serviceComponentHost3 = clusters.getCluster(DummyCluster).getService(HDFS).
+        getServiceComponent(HDFS_CLIENT).getServiceComponentHost(DummyHostname1);
+
+    StackId stack130 = new StackId("HDP-1.3.0");
+    StackId stack122 = new StackId("HDP-1.2.2");
+
+    serviceComponentHost1.setState(State.INSTALLED);
+    serviceComponentHost2.setState(State.STARTED);
+    serviceComponentHost3.setState(State.STARTED);
+    serviceComponentHost1.setStackVersion(stack130);
+    serviceComponentHost2.setStackVersion(stack122);
+    serviceComponentHost3.setStackVersion(stack122);
+
+    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>();
+    ComponentStatus componentStatus1 =
+        createComponentStatus(DummyCluster, HDFS, DummyHostStatus, State.STARTED, DATANODE, "HDP-1.3.0");
+    ComponentStatus componentStatus2 =
+        createComponentStatus(DummyCluster, HDFS, DummyHostStatus, State.STARTED, NAMENODE, "");
+    ComponentStatus componentStatus3 =
+        createComponentStatus(DummyCluster, HDFS, DummyHostStatus, State.INSTALLED, HDFS_CLIENT, "HDP-1.3.0");
+
+    componentStatuses.add(componentStatus1);
+    componentStatuses.add(componentStatus2);
+    componentStatuses.add(componentStatus3);
+    hb.setComponentStatus(componentStatuses);
+
+    ActionQueue aq = new ActionQueue();
+    HeartBeatHandler handler = getHeartBeatHandler(am, aq);
+    handler.handleHeartBeat(hb);
+    assertEquals("Matching value " + serviceComponentHost1.getStackVersion(),
+        stack130, serviceComponentHost1.getStackVersion());
+    assertEquals("Matching value " + serviceComponentHost2.getStackVersion(),
+        stack122, serviceComponentHost2.getStackVersion());
+    assertEquals("Matching value " + serviceComponentHost3.getStackVersion(),
+        stack130, serviceComponentHost3.getStackVersion());
+  }
+
+  private ComponentStatus createComponentStatus(String clusterName, String serviceName, String message,
+                                                State state, String componentName, String stackVersion) {
+    ComponentStatus componentStatus1 = new ComponentStatus();
+    componentStatus1.setClusterName(clusterName);
+    componentStatus1.setServiceName(serviceName);
+    componentStatus1.setMessage(message);
+    componentStatus1.setStatus(state.name());
+    componentStatus1.setComponentName(componentName);
+    componentStatus1.setStackVersion(stackVersion);
+    return componentStatus1;
+  }
+
+  private HeartBeatHandler getHeartBeatHandler(ActionManager am, ActionQueue aq)
+      throws InvalidStateTransitionException, AmbariException {
+    HeartBeatHandler handler = new HeartBeatHandler(clusters, aq, am, injector);
+    Register reg = new Register();
+    HostInfo hi = new HostInfo();
+    hi.setHostName(DummyHostname1);
+    hi.setOS(DummyOs);
+    hi.setOSRelease(DummyOSRelease);
+    reg.setHostname(DummyHostname1);
+    reg.setResponseId(0);
+    reg.setHardwareProfile(hi);
+    handler.handleRegistration(reg);
+    return handler;
+  }
+
+  private Cluster getDummyCluster()
+      throws AmbariException {
+    clusters.addHost(DummyHostname1);
+    clusters.getHost(DummyHostname1).setOsType(DummyOsType);
+    clusters.getHost(DummyHostname1).persist();
+    clusters.addCluster(DummyCluster);
+
+    Cluster cluster = clusters.getCluster(DummyCluster);
+    cluster.setDesiredStackVersion(new StackId(DummyStackId));
+    return cluster;
+  }
 }

ambari-server/src/test/java/org/apache/ambari/server/controller/AmbariManagementControllerTest.java

@@ -3774,8 +3774,5 @@ public class AmbariManagementControllerTest {
     Assert.assertEquals(1, taskStatuses.size());
     Assert.assertEquals(Role.PIG_SERVICE_CHECK.toString(),
         taskStatuses.get(0).getRole());
-
-
   }
-
 }

ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClustersTest.java

@@ -27,15 +27,16 @@ import com.google.inject.persist.PersistService;
 import junit.framework.Assert;
 
 import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.api.services.AmbariMetaInfo;
 import org.apache.ambari.server.ClusterNotFoundException;
 import org.apache.ambari.server.DuplicateResourceException;
 import org.apache.ambari.server.HostNotFoundException;
-import org.apache.ambari.server.api.services.AmbariMetaInfo;
-import org.apache.ambari.server.orm.GuiceJpaInitializer;
-import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
+import org.apache.ambari.server.orm.entities.ClusterStateEntity;
 import org.apache.ambari.server.orm.dao.*;
 import org.apache.ambari.server.orm.entities.HostComponentDesiredStateEntityPK;
 import org.apache.ambari.server.orm.entities.HostComponentStateEntityPK;
+import org.apache.ambari.server.orm.GuiceJpaInitializer;
+import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
 import org.apache.ambari.server.state.*;
 import org.junit.After;
 import org.junit.Before;
@@ -312,7 +313,7 @@ public class ClustersTest {
     Service hdfs = cluster.addService("HDFS");
     hdfs.persist();
 
-    assertNotNull(injector.getInstance(ClusterServiceDAO.class).findByClusterAndServiceNames(c1, "HDFS"));
+    Assert.assertNotNull(injector.getInstance(ClusterServiceDAO.class).findByClusterAndServiceNames(c1, "HDFS"));
 
     ServiceComponent nameNode = hdfs.addServiceComponent("NAMENODE");
     nameNode.persist();
@@ -338,17 +339,64 @@ public class ClustersTest {
     hkdspk.setServiceName(nameNodeHost.getServiceName());
     hkdspk.setComponentName(nameNodeHost.getServiceComponentName());
 
-    assertNotNull(injector.getInstance(HostComponentStateDAO.class).findByPK(hkspk));
-    assertNotNull(injector.getInstance(HostComponentDesiredStateDAO.class).findByPK(hkdspk));
-    assertEquals(1, injector.getProvider(EntityManager.class).get().createQuery("SELECT config FROM ClusterConfigEntity config").getResultList().size());
+    Assert.assertNotNull(injector.getInstance(HostComponentStateDAO.class).findByPK(hkspk));
+    Assert.assertNotNull(injector.getInstance(HostComponentDesiredStateDAO.class).findByPK(hkdspk));
+    Assert.assertEquals(1, injector.getProvider(EntityManager.class).get().createQuery("SELECT config FROM ClusterConfigEntity config").getResultList().size());
 
     clusters.deleteCluster(c1);
 
-    assertEquals(2, injector.getInstance(HostDAO.class).findAll().size());
-    assertNull(injector.getInstance(HostComponentStateDAO.class).findByPK(hkspk));
-    assertNull(injector.getInstance(HostComponentDesiredStateDAO.class).findByPK(hkdspk));
+    Assert.assertEquals(2, injector.getInstance(HostDAO.class).findAll().size());
+    Assert.assertNull(injector.getInstance(HostComponentStateDAO.class).findByPK(hkspk));
+    Assert.assertNull(injector.getInstance(HostComponentDesiredStateDAO.class).findByPK(hkdspk));
     //configs are removed implicitly by cascade operation
-    assertEquals(0, injector.getProvider(EntityManager.class).get().createQuery("SELECT config FROM ClusterConfigEntity config").getResultList().size());
+    Assert.assertEquals(0, injector.getProvider(EntityManager.class).get().createQuery("SELECT config FROM ClusterConfigEntity config").getResultList().size());
+
+  }
+  @Test
+  public void testSetCurrentStackVersion() throws AmbariException {
+
+    String c1 = "foo3";
+
+    try
+    {
+      clusters.setCurrentStackVersion("", null);
+      fail("Exception should be thrown on invalid set");
+    }
+      catch (AmbariException e) {
+      // Expected
+    }
 
+    try
+    {
+      clusters.setCurrentStackVersion(c1, null);
+      fail("Exception should be thrown on invalid set");
+    }
+    catch (AmbariException e) {
+      // Expected
+    }
+
+    StackId stackId = new StackId("HDP-0.1");
+
+    try
+    {
+      clusters.setCurrentStackVersion(c1, stackId);
+      fail("Exception should be thrown on invalid set");
+    }
+    catch (AmbariException e) {
+      // Expected
+      Assert.assertTrue(e.getMessage().contains("Cluster not found"));
+    }
+
+    clusters.addCluster(c1);
+    clusters.setCurrentStackVersion(c1, stackId);
+
+    Assert.assertNotNull(clusters.getCluster(c1));
+    ClusterStateEntity entity = injector.getInstance(ClusterStateDAO.class).findByPK(clusters.getCluster(c1).getClusterId());
+    Assert.assertNotNull(entity);
+    Assert.assertTrue(entity.getCurrentStackVersion().contains(stackId.getStackName()) &&
+        entity.getCurrentStackVersion().contains(stackId.getStackVersion()));
+    Assert.assertTrue(clusters.getCluster(c1).getCurrentStackVersion().getStackName().equals(stackId.getStackName()));
+    Assert.assertTrue(
+        clusters.getCluster(c1).getCurrentStackVersion().getStackVersion().equals(stackId.getStackVersion()));
   }
 }