AMBARI-8924. Add OUT_OF_SYNC set to cluster stack version API (dlysnichenko)

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ClusterStackVersionResourceProvider.java

@@ -387,6 +387,7 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou
         cluster.transitionClusterVersion(stackId, desiredRepoVersion, RepositoryVersionState.INSTALLING);
       }
       cluster.inferHostVersions(existingCSVer);
+      cluster.recalculateClusterVersionState(desiredRepoVersion);
 
       req.persist();
 

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostResourceProvider.java

@@ -437,6 +437,7 @@ public class HostResourceProvider extends BaseBlueprintProcessor {
 
     Map<String, Set<String>> hostClustersMap = new HashMap<String, Set<String>>();
     Map<String, Map<String, String>> hostAttributes = new HashMap<String, Map<String, String>>();
+    Set<String> allClusterSet = new HashSet<String>();
 
     for (HostRequest hostRequest : hostRequests) {
       if (hostRequest.getHostname() != null &&
@@ -446,6 +447,7 @@ public class HostResourceProvider extends BaseBlueprintProcessor {
 
         Set<String> clusterSet = new HashSet<String>();
         clusterSet.add(hostRequest.getClusterName());
+        allClusterSet.add(hostRequest.getClusterName());
         hostClustersMap.put(hostRequest.getHostname(), clusterSet);
         if (hostRequest.getHostAttributes() != null) {
           hostAttributes.put(hostRequest.getHostname(), hostRequest.getHostAttributes());
@@ -453,6 +455,10 @@ public class HostResourceProvider extends BaseBlueprintProcessor {
       }
     }
     clusters.updateHostWithClusterAndAttributes(hostClustersMap, hostAttributes);
+
+    for (String clusterName : allClusterSet) {
+      clusters.getCluster(clusterName).recalculateAllClusterVersionStates();
+    }
   }
 
   private void createHostResource(Clusters clusters, Set<String> duplicates,
@@ -793,6 +799,11 @@ public class HostResourceProvider extends BaseBlueprintProcessor {
           }
         }
       }
+
+      if (null != request.getClusterName() && !request.getClusterName().isEmpty()) {
+        clusters.getCluster(request.getClusterName()).recalculateAllClusterVersionStates();
+      }
+
       //todo: if attempt was made to update a property other than those
       //todo: that are allowed above, should throw exception
     }
@@ -864,6 +875,7 @@ public class HostResourceProvider extends BaseBlueprintProcessor {
       if (null != hostRequest.getClusterName()) {
         clusters.unmapHostFromCluster(hostRequest.getHostname(),
             hostRequest.getClusterName());
+        clusters.getCluster(hostRequest.getClusterName()).recalculateAllClusterVersionStates();
       } else {
         clusters.deleteHost(hostRequest.getHostname());
       }

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostStackVersionResourceProvider.java

@@ -420,6 +420,7 @@ public class HostStackVersionResourceProvider extends AbstractControllerResource
     try {
       hostVersEntity.setState(RepositoryVersionState.INSTALLING);
       hostVersionDAO.merge(hostVersEntity);
+      cluster.recalculateClusterVersionState(desiredRepoVersion);
       req.persist();
     } catch (AmbariException e) {
       throw new SystemException("Can not persist request", e);

ambari-server/src/main/java/org/apache/ambari/server/events/ServiceComponentInstalledEvent.java

@@ -46,6 +46,14 @@ public class ServiceComponentInstalledEvent extends ServiceEvent {
     m_hostName = hostName;
   }
 
+  public String getComponentName() {
+    return m_componentName;
+  }
+
+  public String getHostName() {
+    return m_hostName;
+  }
+
   /**
    * {@inheritDoc}
    */

ambari-server/src/main/java/org/apache/ambari/server/events/listeners/upgrade/HostVersionOutOfSyncListener.java

@@ -0,0 +1,126 @@
+/**
+ * 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.events.listeners.upgrade;
+
+import com.google.common.eventbus.Subscribe;
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import com.google.inject.Singleton;
+import com.google.inject.persist.Transactional;
+import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.EagerSingleton;
+import org.apache.ambari.server.events.ServiceComponentInstalledEvent;
+import org.apache.ambari.server.events.ServiceInstalledEvent;
+import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
+import org.apache.ambari.server.orm.dao.HostVersionDAO;
+import org.apache.ambari.server.orm.entities.HostVersionEntity;
+import org.apache.ambari.server.state.Cluster;
+import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.RepositoryVersionState;
+import org.apache.ambari.server.state.StackId;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * The {@link org.apache.ambari.server.events.listeners.upgrade.HostVersionOutOfSyncListener} class
+ * handles {@link org.apache.ambari.server.events.ServiceInstalledEvent} and
+ * {@link org.apache.ambari.server.events.ServiceComponentInstalledEvent}
+ * to update {@link org.apache.ambari.server.state.RepositoryVersionState}
+ *
+ * @see org.apache.ambari.server.state.Cluster#recalculateClusterVersionState(String)
+ */
+@Singleton
+@EagerSingleton
+public class HostVersionOutOfSyncListener {
+  /**
+   * Logger.
+   */
+  private static final Logger LOG = LoggerFactory.getLogger(HostVersionOutOfSyncListener.class);
+
+  @Inject
+  private Provider<HostVersionDAO> hostVersionDAO;
+
+  @Inject
+  private Provider<Clusters> clusters;
+
+  private AmbariEventPublisher ambariEventPublisher;
+
+  @Inject
+  public HostVersionOutOfSyncListener(AmbariEventPublisher ambariEventPublisher) {
+    this.ambariEventPublisher = ambariEventPublisher;
+    ambariEventPublisher.register(this);
+  }
+
+  @Subscribe
+  @Transactional
+  public void onServiceComponentEvent(ServiceComponentInstalledEvent event) {
+    if (LOG.isDebugEnabled()) {
+      LOG.debug(event.toString());
+    }
+    try {
+      Cluster cluster = clusters.get().getClusterById(event.getClusterId());
+      List<HostVersionEntity> hostVersionEntities =
+          hostVersionDAO.get().findByClusterAndHost(cluster.getClusterName(), event.getHostName());
+      StackId currentStackId = cluster.getCurrentStackVersion();
+      for (HostVersionEntity hostVersionEntity : hostVersionEntities) {
+        if (hostVersionEntity.getRepositoryVersion().getStack().equals(currentStackId.getStackId())
+            && hostVersionEntity.getState().equals(RepositoryVersionState.INSTALLED)) {
+          hostVersionEntity.setState(RepositoryVersionState.OUT_OF_SYNC);
+          hostVersionDAO.get().merge(hostVersionEntity);
+          cluster.recalculateClusterVersionState(hostVersionEntity.getRepositoryVersion().getVersion());
+        }
+      }
+    } catch (AmbariException e) {
+      LOG.error("Can not update hosts about out of sync", e);
+    }
+  }
+
+  @Subscribe
+  @Transactional
+  public void onServiceEvent(ServiceInstalledEvent event) {
+    if (LOG.isDebugEnabled()) {
+      LOG.debug(event.toString());
+    }
+    try {
+      Cluster cluster = clusters.get().getClusterById(event.getClusterId());
+      Set<String> changedRepositoryVersions = new HashSet<String>();
+      StackId currentStackId = cluster.getCurrentStackVersion();
+      for (String hostName : clusters.get().getHostsForCluster(cluster.getClusterName()).keySet()) {
+        List<HostVersionEntity> hostVersionEntities =
+            hostVersionDAO.get().findByClusterAndHost(cluster.getClusterName(), hostName);
+        for (HostVersionEntity hostVersionEntity : hostVersionEntities) {
+          if (hostVersionEntity.getRepositoryVersion().getStack().equals(currentStackId.getStackId())
+              && hostVersionEntity.getState().equals(RepositoryVersionState.INSTALLED)) {
+            hostVersionEntity.setState(RepositoryVersionState.OUT_OF_SYNC);
+            hostVersionDAO.get().merge(hostVersionEntity);
+            changedRepositoryVersions.add(hostVersionEntity.getRepositoryVersion().getVersion());
+          }
+        }
+      }
+      for (String version : changedRepositoryVersions) {
+        cluster.recalculateClusterVersionState(version);
+      }
+    } catch (AmbariException e) {
+      LOG.error("Can not update hosts about out of sync", e);
+    }
+  }
+}

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

@@ -153,16 +153,25 @@ public interface Cluster {
   /**
    * Update state of a cluster stack version for cluster based on states of host versions.
    * May be called multiple times.
-   * As of now, only transition from INSTALLING to INSTALLING/INSTALLED/INSTALL_FAILED
+   * As of now, only transition from INSTALLING to INSTALLING/INSTALLED/INSTALL_FAILED/OUT_OF_SYNC
    * is supported
    * @throws AmbariException
    */
   void recalculateClusterVersionState(String repositoryVersion) throws AmbariException;
 
+  /**
+   * Update state of all cluster stack versions for cluster based on states of host versions.
+   * May be called multiple times.
+   * As of now, only transition from INSTALLING to INSTALLING/INSTALLED/INSTALL_FAILED/OUT_OF_SYNC
+   * is supported
+   * @throws AmbariException
+   */
+  public void recalculateAllClusterVersionStates() throws AmbariException;
+
   /**
    * Create a cluster version for the given stack and version, whose initial state must either
    * be either {@link RepositoryVersionState#CURRENT} (if no other cluster version exists) or
-   * {@link RepositoryVersionState#UPGRADING} (if at exactly one CURRENT cluster version already exists).
+   * {@link RepositoryVersionState#INSTALLING} (if at exactly one CURRENT cluster version already exists).
    * @param stack Stack name
    * @param version Stack version
    * @param userName User performing the operation

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

@@ -70,9 +70,9 @@ package org.apache.ambari.server.state;
  * UPGRADING -> UPGRADED | UPGRADE_FAILED
  * UPGRADE_FAILED -> UPGRADING
  * UPGRADED -> CURRENT
- * INSTALLING -> INSTALLED | INSTALL_FAILED
- * INSTALLED -> INSTALLED
- * INSTALLED -> INSTALLING
+ * INSTALLING -> INSTALLED | INSTALL_FAILED | OUT_OF_SYNC
+ * INSTALLED -> INSTALLED | INSTALLING | OUT_OF_SYNC
+ * OUT_OF_SYNC -> INSTALLING
  * INSTALL_FAILED -> INSTALLING
  * CURRENT -> INSTALLED
  * INSTALLED -> UPGRADING
@@ -82,30 +82,49 @@ public enum RepositoryVersionState {
   /**
    * Repository version that is in the process of being installed.
    */
-  INSTALLING,
+  INSTALLING(2),
   /**
    * Repository version that is installed and supported but not the active version.
    */
-  INSTALLED,
+  INSTALLED(3),
   /**
    * Repository version that during the install process failed to install some components.
    */
-  INSTALL_FAILED,
+  INSTALL_FAILED(0),
+  /**
+   * Repository version that is installed for some components but not for all.
+   */
+  OUT_OF_SYNC(1),
   /**
    * Repository version that is installed and supported and is the active version.
    */
-  CURRENT,
+  CURRENT(7),
   /**
    * Repository version that is in the process of upgrading to become the CURRENT active version,
    * and the previous active version transitions to an INSTALLED state.
    */
-  UPGRADING,
+  UPGRADING(5),
   /**
    * Repository version that during the upgrade process failed to become the active version and must be remedied.
    */
-  UPGRADE_FAILED,
+  UPGRADE_FAILED(4),
   /**
    * Repository version that finished upgrading and should be finalized to become CURRENT.
    */
-  UPGRADED
+  UPGRADED(6);
+
+  /**
+   * Is used to determine cluster version state.
+   * @see org.apache.ambari.server.state.Cluster#recalculateClusterVersionState(String)
+   */
+  private int priority;
+
+  RepositoryVersionState(int priority) {
+    this.priority = priority;
+  }
+
+  public int getPriority() {
+    return priority;
+  }
+
 }

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

@@ -1247,6 +1247,10 @@ public class ClusterImpl implements Cluster {
     }
   }
 
+  private RepositoryVersionState getWorstState(RepositoryVersionState currentState, RepositoryVersionState newState) {
+    return currentState.getPriority() < newState.getPriority() ? currentState : newState;
+  }
+
   @Override
   public void recalculateClusterVersionState(String repositoryVersion) throws AmbariException {
     clusterGlobalLock.readLock().lock();
@@ -1256,7 +1260,7 @@ public class ClusterImpl implements Cluster {
         Map<String, Host> hosts = clusters.getHostsForCluster(this.getClusterName());
         String stackId = this.getCurrentStackVersion().getStackId();
         ClusterVersionEntity clusterVersion = clusterVersionDAO.findByClusterAndStackAndVersion(this.getClusterName(),
-                stackId, repositoryVersion);
+            stackId, repositoryVersion);
 
         if (clusterVersion == null) {
           throw new AmbariException(String.format("Repository version %s not found for cluster %s",
@@ -1265,6 +1269,7 @@ public class ClusterImpl implements Cluster {
 
         RepositoryVersionState worstState;
         if (clusterVersion.getState() != RepositoryVersionState.INSTALL_FAILED &&
+                clusterVersion.getState() != RepositoryVersionState.OUT_OF_SYNC &&
                 clusterVersion.getState() != RepositoryVersionState.INSTALLING &&
                 clusterVersion.getState() != RepositoryVersionState.INSTALLED) {
           // anything else is not supported as of now
@@ -1275,23 +1280,19 @@ public class ClusterImpl implements Cluster {
         for (Host host : hosts.values()) {
           String hostName = host.getHostName();
           if (host.getState() != HostState.HEALTHY) {
-            worstState = RepositoryVersionState.INSTALL_FAILED;
+            worstState = getWorstState(worstState, RepositoryVersionState.OUT_OF_SYNC);
             LOG.warn(String.format("Host %s is in unhealthy state, treating as %s",
                     hostName, worstState));
-            continue;
           }
 
           HostVersionEntity hostVersion = hostVersionDAO.findByClusterStackVersionAndHost(this.getClusterName(),
                   stackId, repositoryVersion, hostName);
           if (hostVersion == null) {
-            throw new AmbariException(String.format("Repo version %s is not installed on host %s",
+            LOG.warn(String.format("Repo version %s is not installed on host %s",
                     repositoryVersion, hostName));
-          }
-          if (hostVersion.getState() == RepositoryVersionState.INSTALL_FAILED) {
-            worstState = hostVersion.getState();
-            break;
-          } else if (hostVersion.getState() == RepositoryVersionState.INSTALLING) {
-            worstState = RepositoryVersionState.INSTALLING;
+            worstState = getWorstState(worstState, RepositoryVersionState.OUT_OF_SYNC);
+          } else {
+            worstState = getWorstState(worstState, hostVersion.getState());
           }
         }
         if (worstState != clusterVersion.getState()) {
@@ -1308,10 +1309,32 @@ public class ClusterImpl implements Cluster {
     }
   }
 
+  @Override
+  public void recalculateAllClusterVersionStates() throws AmbariException {
+    clusterGlobalLock.readLock().lock();
+    try {
+      readWriteLock.writeLock().lock();
+      try {
+        List<ClusterVersionEntity> clusterVersionEntities = clusterVersionDAO.findByCluster(getClusterName());
+        StackId currentStackId = getCurrentStackVersion();
+        for (ClusterVersionEntity clusterVersionEntity : clusterVersionEntities) {
+          if (clusterVersionEntity.getRepositoryVersion().getStack().equals(currentStackId.getStackId())
+            && clusterVersionEntity.getState() != RepositoryVersionState.CURRENT) {
+            recalculateClusterVersionState(clusterVersionEntity.getRepositoryVersion().getVersion());
+          }
+        }
+      } finally {
+        readWriteLock.writeLock().unlock();
+      }
+    } finally {
+      clusterGlobalLock.readLock().unlock();
+    }
+  }
+
   /**
    * Create a cluster version for the given stack and version, whose initial state must either
    * be either {@link org.apache.ambari.server.state.RepositoryVersionState#CURRENT} (if no other cluster version exists) or
-   * {@link org.apache.ambari.server.state.RepositoryVersionState#UPGRADING} (if at exactly one CURRENT cluster version already exists).
+   * {@link org.apache.ambari.server.state.RepositoryVersionState#INSTALLING} (if at exactly one CURRENT cluster version already exists).
    * @param stack Stack name
    * @param version Stack version
    * @param userName User performing the operation
@@ -1325,7 +1348,8 @@ public class ClusterImpl implements Cluster {
       readWriteLock.writeLock().lock();
       try {
         Set<RepositoryVersionState> allowedStates = new HashSet<RepositoryVersionState>();
-        if (this.clusterEntity.getClusterVersionEntities() == null || this.clusterEntity.getClusterVersionEntities().isEmpty()) {
+        Collection<ClusterVersionEntity> allClusterVersions = getAllClusterVersions();
+        if (allClusterVersions == null || allClusterVersions.isEmpty()) {
           allowedStates.add(RepositoryVersionState.CURRENT);
         } else {
           allowedStates.add(RepositoryVersionState.INSTALLING);
@@ -1364,6 +1388,7 @@ public class ClusterImpl implements Cluster {
    * @throws AmbariException
    */
   @Override
+  @Transactional
   public void transitionClusterVersion(String stack, String version, RepositoryVersionState state) throws AmbariException {
     Set<RepositoryVersionState> allowedStates = new HashSet<RepositoryVersionState>();
     clusterGlobalLock.readLock().lock();
@@ -1378,22 +1403,36 @@ public class ClusterImpl implements Cluster {
         if (existingClusterVersion.getState() != state) {
           switch (existingClusterVersion.getState()) {
             case CURRENT:
-              allowedStates.add(RepositoryVersionState.INSTALLED);
+              // If CURRENT state is changed here cluster will not have CURRENT state.
+              // CURRENT state will be changed to INSTALLED when another CURRENT state is added.
+              // allowedStates.add(RepositoryVersionState.INSTALLED);
+              break;
             case INSTALLING:
               allowedStates.add(RepositoryVersionState.INSTALLED);
               allowedStates.add(RepositoryVersionState.INSTALL_FAILED);
+              allowedStates.add(RepositoryVersionState.OUT_OF_SYNC);
+              break;
             case INSTALL_FAILED:
               allowedStates.add(RepositoryVersionState.INSTALLING);
+              break;
             case INSTALLED:
               allowedStates.add(RepositoryVersionState.INSTALLING);
               allowedStates.add(RepositoryVersionState.UPGRADING);
+              allowedStates.add(RepositoryVersionState.OUT_OF_SYNC);
+              break;
+            case OUT_OF_SYNC:
+              allowedStates.add(RepositoryVersionState.INSTALLING);
+              break;
             case UPGRADING:
               allowedStates.add(RepositoryVersionState.UPGRADED);
               allowedStates.add(RepositoryVersionState.UPGRADE_FAILED);
+              break;
             case UPGRADED:
               allowedStates.add(RepositoryVersionState.CURRENT);
+              break;
             case UPGRADE_FAILED:
               allowedStates.add(RepositoryVersionState.UPGRADING);
+              break;
           }
 
           if (!allowedStates.contains(state)) {

ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java

@@ -33,6 +33,7 @@ import org.apache.ambari.server.agent.AlertDefinitionCommand;
 import org.apache.ambari.server.controller.ServiceComponentHostResponse;
 import org.apache.ambari.server.events.AlertHashInvalidationEvent;
 import org.apache.ambari.server.events.MaintenanceModeEvent;
+import org.apache.ambari.server.events.ServiceComponentInstalledEvent;
 import org.apache.ambari.server.events.ServiceComponentUninstalledEvent;
 import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
 import org.apache.ambari.server.orm.dao.HostComponentDesiredStateDAO;
@@ -1312,6 +1313,15 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
           host.refresh();
           serviceComponent.refresh();
           persisted = true;
+
+          // publish the service component installed event
+          StackId stackId = getDesiredStackVersion();
+
+          ServiceComponentInstalledEvent event = new ServiceComponentInstalledEvent(
+              getClusterId(), stackId.getStackName(),
+              stackId.getStackVersion(), getServiceName(), getServiceComponentName(), getHostName());
+
+          eventPublisher.publish(event);
         } else {
           saveIfPersisted();
         }

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

@@ -1487,6 +1487,7 @@ public class AmbariManagementControllerTest {
     Cluster c = clusters.getCluster("foo");
     StackId stackId = new StackId("HDP-0.1");
     c.setDesiredStackVersion(stackId);
+    c.setCurrentStackVersion(stackId);
     helper.getOrCreateRepositoryVersion(stackId.getStackName(), stackId.getStackVersion());
     c.createClusterVersion(stackId.getStackName(), stackId.getStackVersion(), "admin", RepositoryVersionState.CURRENT);
 
@@ -1509,6 +1510,7 @@ public class AmbariManagementControllerTest {
     Cluster c = clusters.getCluster("c1");
     StackId stackID = new StackId("HDP-0.1");
     c.setDesiredStackVersion(stackID);
+    c.setCurrentStackVersion(stackID);
     helper.getOrCreateRepositoryVersion(stackID.getStackName(), stackID.getStackVersion());
     c.createClusterVersion(stackID.getStackName(), stackID.getStackVersion(), "admin", RepositoryVersionState.CURRENT);
 

ambari-server/src/test/java/org/apache/ambari/server/orm/OrmTestHelper.java

@@ -43,6 +43,7 @@ import org.apache.ambari.server.orm.dao.AlertsDAO;
 import org.apache.ambari.server.orm.dao.ClusterDAO;
 import org.apache.ambari.server.orm.dao.HostDAO;
 import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
+import org.apache.ambari.server.orm.dao.HostVersionDAO;
 import org.apache.ambari.server.orm.dao.RepositoryVersionDAO;
 import org.apache.ambari.server.orm.dao.RequestDAO;
 import org.apache.ambari.server.orm.dao.ResourceTypeDAO;
@@ -56,6 +57,7 @@ import org.apache.ambari.server.orm.entities.ClusterServiceEntity;
 import org.apache.ambari.server.orm.entities.HostEntity;
 import org.apache.ambari.server.orm.entities.HostRoleCommandEntity;
 import org.apache.ambari.server.orm.entities.HostStateEntity;
+import org.apache.ambari.server.orm.entities.HostVersionEntity;
 import org.apache.ambari.server.orm.entities.PrincipalEntity;
 import org.apache.ambari.server.orm.entities.PrincipalTypeEntity;
 import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
@@ -111,6 +113,12 @@ public class OrmTestHelper {
   @Inject
   public RepositoryVersionDAO repositoryVersionDAO;
 
+  @Inject
+  public HostVersionDAO hostVersionDAO;
+
+  @Inject
+  public HostDAO hostDAO;
+
   public EntityManager getEntityManager() {
     return entityManagerProvider.get();
   }
@@ -562,4 +570,16 @@ public class OrmTestHelper {
     return repositoryVersion;
   }
 
+  /**
+   * Convenient method to create host version for given stack.
+   */
+  public HostVersionEntity createHostVersion(String hostName, RepositoryVersionEntity repositoryVersionEntity,
+                                             RepositoryVersionState repositoryVersionState) {
+    HostEntity hostEntity = hostDAO.findByName(hostName);
+    HostVersionEntity hostVersionEntity = new HostVersionEntity(hostName, repositoryVersionEntity, repositoryVersionState);
+    hostVersionEntity.setHostEntity(hostEntity);
+    hostVersionDAO.create(hostVersionEntity);
+    return hostVersionEntity;
+  }
+
 }

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

@@ -19,6 +19,7 @@
 package org.apache.ambari.server.state.cluster;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Mockito.mock;
@@ -37,7 +38,12 @@ import java.util.Map;
 import java.util.Set;
 
 import javax.persistence.EntityManager;
+import javax.persistence.RollbackException;
 
+import com.google.inject.AbstractModule;
+import com.google.inject.Singleton;
+import com.google.inject.persist.Transactional;
+import com.google.inject.util.Modules;
 import junit.framework.Assert;
 
 import org.apache.ambari.server.AmbariException;
@@ -51,11 +57,16 @@ import org.apache.ambari.server.controller.ServiceConfigVersionResponse;
 import org.apache.ambari.server.orm.GuiceJpaInitializer;
 import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
 import org.apache.ambari.server.orm.OrmTestHelper;
+import org.apache.ambari.server.orm.dao.ClusterVersionDAO;
+import org.apache.ambari.server.orm.dao.HostVersionDAO;
 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.orm.entities.ClusterVersionEntity;
 import org.apache.ambari.server.orm.entities.HostEntity;
 import org.apache.ambari.server.orm.entities.HostStateEntity;
+import org.apache.ambari.server.orm.entities.HostVersionEntity;
+import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
 import org.apache.ambari.server.orm.entities.ServiceDesiredStateEntity;
 import org.apache.ambari.server.state.AgentVersion;
 import org.apache.ambari.server.state.Cluster;
@@ -105,10 +116,33 @@ public class ClusterTest {
   private ConfigFactory configFactory;
   private ConfigGroupFactory configGroupFactory;
   private OrmTestHelper helper;
+  private HostVersionDAO hostVersionDAO;
+
+  @Singleton
+  static class ClusterVersionDAOMock extends ClusterVersionDAO {
+    static boolean failOnCurrentVersionState;
+
+    @Override
+    @Transactional
+    public ClusterVersionEntity merge(ClusterVersionEntity entity) {
+      if (!failOnCurrentVersionState || entity.getState() != RepositoryVersionState.CURRENT) {
+        return super.merge(entity);
+      } else {
+        throw new RollbackException();
+      }
+    }
+  }
+
+  private static class MockModule extends AbstractModule {
+    @Override
+    protected void configure() {
+      bind(ClusterVersionDAO.class).to(ClusterVersionDAOMock.class);
+    }
+  }
 
   @Before
   public void setup() throws Exception {
-    injector = Guice.createInjector(new InMemoryDefaultTestModule());
+    injector = Guice.createInjector(Modules.override(new InMemoryDefaultTestModule()).with(new MockModule()));
     injector.getInstance(GuiceJpaInitializer.class);
     clusters = injector.getInstance(Clusters.class);
     serviceFactory = injector.getInstance(ServiceFactory.class);
@@ -120,6 +154,7 @@ public class ClusterTest {
     configFactory = injector.getInstance(ConfigFactory.class);
     metaInfo = injector.getInstance(AmbariMetaInfo.class);
     helper = injector.getInstance(OrmTestHelper.class);
+    hostVersionDAO = injector.getInstance(HostVersionDAO.class);
     metaInfo.init();
     clusters.addCluster("c1");
     c1 = clusters.getCluster("c1");
@@ -138,9 +173,10 @@ public class ClusterTest {
     host.persist();
     StackId stackId = new StackId("HDP-0.1");
     c1.setDesiredStackVersion(stackId);
-    helper.getOrCreateRepositoryVersion(stackId.getStackName(), stackId.getStackVersion());
-    c1.createClusterVersion(stackId.getStackName(), stackId.getStackVersion(), "admin", RepositoryVersionState.CURRENT);
+    helper.getOrCreateRepositoryVersion(stackId.getStackId(), "1.0-2086");
+    c1.createClusterVersion(stackId.getStackId(), "1.0-2086", "admin", RepositoryVersionState.CURRENT);
     clusters.mapHostToCluster("h1", "c1");
+    ClusterVersionDAOMock.failOnCurrentVersionState = false;
   }
 
   @After
@@ -807,5 +843,221 @@ public class ClusterTest {
 
   }
 
+  private void checkStackVersionState(String stack, String version, RepositoryVersionState state) {
+    Collection<ClusterVersionEntity> allClusterVersions = c1.getAllClusterVersions();
+    for (ClusterVersionEntity entity : allClusterVersions) {
+      if (entity.getRepositoryVersion().getStack().equals(stack)
+          && entity.getRepositoryVersion().getVersion().equals(version)) {
+        assertEquals(state, entity.getState());
+      }
+    }
+  }
+
+  private void assertStateException(String stack, String version, RepositoryVersionState transitionState,
+                                    RepositoryVersionState stateAfter) {
+    try {
+      c1.transitionClusterVersion(stack, version, transitionState);
+      Assert.fail();
+    } catch (AmbariException e) {}
+    checkStackVersionState(stack, version, stateAfter);
+    assertNotNull(c1.getCurrentClusterVersion());
+  }
+
+  @Test
+  public void testTransitionClusterVersion() throws AmbariException {
+    String stack = "HDP";
+    String version = "0.2";
+
+    helper.getOrCreateRepositoryVersion(stack, version);
+    c1.createClusterVersion(stack, version, "admin", RepositoryVersionState.INSTALLING);
+
+    assertStateException(stack, version, RepositoryVersionState.CURRENT, RepositoryVersionState.INSTALLING);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADING, RepositoryVersionState.INSTALLING);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADED, RepositoryVersionState.INSTALLING);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADE_FAILED, RepositoryVersionState.INSTALLING);
+
+    c1.transitionClusterVersion(stack, version, RepositoryVersionState.INSTALL_FAILED);
+    checkStackVersionState(stack, version, RepositoryVersionState.INSTALL_FAILED);
+
+    assertStateException(stack, version, RepositoryVersionState.CURRENT, RepositoryVersionState.INSTALL_FAILED);
+    assertStateException(stack, version, RepositoryVersionState.INSTALLED, RepositoryVersionState.INSTALL_FAILED);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADING, RepositoryVersionState.INSTALL_FAILED);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADED, RepositoryVersionState.INSTALL_FAILED);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADE_FAILED, RepositoryVersionState.INSTALL_FAILED);
+    assertStateException(stack, version, RepositoryVersionState.OUT_OF_SYNC, RepositoryVersionState.INSTALL_FAILED);
+
+    c1.transitionClusterVersion(stack, version, RepositoryVersionState.INSTALLING);
+    checkStackVersionState(stack, version, RepositoryVersionState.INSTALLING);
+
+    c1.transitionClusterVersion(stack, version, RepositoryVersionState.INSTALLED);
+    checkStackVersionState(stack, version, RepositoryVersionState.INSTALLED);
+
+    assertStateException(stack, version, RepositoryVersionState.CURRENT, RepositoryVersionState.INSTALLED);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADED, RepositoryVersionState.INSTALLED);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADE_FAILED, RepositoryVersionState.INSTALLED);
+    assertStateException(stack, version, RepositoryVersionState.INSTALL_FAILED, RepositoryVersionState.INSTALLED);
+
+    c1.transitionClusterVersion(stack, version, RepositoryVersionState.OUT_OF_SYNC);
+    checkStackVersionState(stack, version, RepositoryVersionState.OUT_OF_SYNC);
+
+    assertStateException(stack, version, RepositoryVersionState.CURRENT, RepositoryVersionState.OUT_OF_SYNC);
+    assertStateException(stack, version, RepositoryVersionState.INSTALLED, RepositoryVersionState.OUT_OF_SYNC);
+    assertStateException(stack, version, RepositoryVersionState.INSTALL_FAILED, RepositoryVersionState.OUT_OF_SYNC);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADING, RepositoryVersionState.OUT_OF_SYNC);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADED, RepositoryVersionState.OUT_OF_SYNC);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADE_FAILED, RepositoryVersionState.OUT_OF_SYNC);
+
+    c1.transitionClusterVersion(stack, version, RepositoryVersionState.INSTALLING);
+    checkStackVersionState(stack, version, RepositoryVersionState.INSTALLING);
+
+    c1.transitionClusterVersion(stack, version, RepositoryVersionState.INSTALLED);
+    checkStackVersionState(stack, version, RepositoryVersionState.INSTALLED);
+
+    c1.transitionClusterVersion(stack, version, RepositoryVersionState.UPGRADING);
+    checkStackVersionState(stack, version, RepositoryVersionState.UPGRADING);
+
+    assertStateException(stack, version, RepositoryVersionState.CURRENT, RepositoryVersionState.UPGRADING);
+    assertStateException(stack, version, RepositoryVersionState.INSTALLED, RepositoryVersionState.UPGRADING);
+    assertStateException(stack, version, RepositoryVersionState.INSTALL_FAILED, RepositoryVersionState.UPGRADING);
+    assertStateException(stack, version, RepositoryVersionState.OUT_OF_SYNC, RepositoryVersionState.UPGRADING);
+
+    c1.transitionClusterVersion(stack, version, RepositoryVersionState.UPGRADE_FAILED);
+    checkStackVersionState(stack, version, RepositoryVersionState.UPGRADE_FAILED);
+
+    assertStateException(stack, version, RepositoryVersionState.CURRENT, RepositoryVersionState.UPGRADE_FAILED);
+    assertStateException(stack, version, RepositoryVersionState.INSTALLED, RepositoryVersionState.UPGRADE_FAILED);
+    assertStateException(stack, version, RepositoryVersionState.INSTALL_FAILED, RepositoryVersionState.UPGRADE_FAILED);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADED, RepositoryVersionState.UPGRADE_FAILED);
+    assertStateException(stack, version, RepositoryVersionState.OUT_OF_SYNC, RepositoryVersionState.UPGRADE_FAILED);
+
+    c1.transitionClusterVersion(stack, version, RepositoryVersionState.UPGRADING);
+    checkStackVersionState(stack, version, RepositoryVersionState.UPGRADING);
+
+    c1.transitionClusterVersion(stack, version, RepositoryVersionState.UPGRADED);
+    checkStackVersionState(stack, version, RepositoryVersionState.UPGRADED);
+
+    assertStateException(stack, version, RepositoryVersionState.INSTALLED, RepositoryVersionState.UPGRADED);
+    assertStateException(stack, version, RepositoryVersionState.INSTALL_FAILED, RepositoryVersionState.UPGRADED);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADING, RepositoryVersionState.UPGRADED);
+    assertStateException(stack, version, RepositoryVersionState.UPGRADE_FAILED, RepositoryVersionState.UPGRADED);
+    assertStateException(stack, version, RepositoryVersionState.OUT_OF_SYNC, RepositoryVersionState.UPGRADED);
+
+    c1.transitionClusterVersion(stack, version, RepositoryVersionState.CURRENT);
+    checkStackVersionState(stack, version, RepositoryVersionState.CURRENT);
+    checkStackVersionState("HDP", "0.1", RepositoryVersionState.INSTALLED);
+
+    // The only CURRENT state should not be changed
+    assertStateException(stack, version, RepositoryVersionState.INSTALLED, RepositoryVersionState.CURRENT);
+  }
 
+  @Test
+  public void testTransitionClusterVersionTransactionFail() throws AmbariException {
+    ClusterVersionDAOMock.failOnCurrentVersionState = true;
+    helper.getOrCreateRepositoryVersion("HDP", "0.2");
+    c1.createClusterVersion("HDP", "0.2", "admin", RepositoryVersionState.INSTALLING);
+    c1.transitionClusterVersion("HDP", "0.2", RepositoryVersionState.INSTALLED);
+    c1.transitionClusterVersion("HDP", "0.2", RepositoryVersionState.UPGRADING);
+    c1.transitionClusterVersion("HDP", "0.2", RepositoryVersionState.UPGRADED);
+    try {
+      c1.transitionClusterVersion("HDP", "0.2", RepositoryVersionState.CURRENT);
+      Assert.fail();
+    } catch (AmbariException e) {}
+
+    // There must be CURRENT state for cluster
+    assertNotNull(c1.getCurrentClusterVersion());
+  }
+
+  @Test
+  public void testInferHostVersions() throws AmbariException {
+    helper.getOrCreateRepositoryVersion("HDP", "0.2");
+    c1.createClusterVersion("HDP", "0.2", "admin", RepositoryVersionState.INSTALLING);
+    ClusterVersionEntity entityHDP2 = null;
+    for (ClusterVersionEntity entity : c1.getAllClusterVersions()) {
+      if (entity.getRepositoryVersion().getStack().equals("HDP")
+          && entity.getRepositoryVersion().getVersion().equals("0.2")) {
+        entityHDP2 = entity;
+        break;
+      }
+    }
+    assertNotNull(entityHDP2);
+
+    List<HostVersionEntity> hostVersionsH1Before = hostVersionDAO.findByClusterAndHost("c1", "h1");
+    assertEquals(1, hostVersionsH1Before.size());
+
+    c1.inferHostVersions(entityHDP2);
+
+    List<HostVersionEntity> hostVersionsH1After = hostVersionDAO.findByClusterAndHost("c1", "h1");
+    assertEquals(2, hostVersionsH1After.size());
+
+    boolean checked = false;
+    for (HostVersionEntity entity : hostVersionsH1After) {
+      if (entity.getRepositoryVersion().getStack().equals("HDP")
+          && entity.getRepositoryVersion().getVersion().equals("0.2")) {
+        assertEquals(RepositoryVersionState.INSTALLING, entity.getState());
+        checked = true;
+        break;
+      }
+    }
+    assertTrue(checked);
+
+    // Test for update of existing host stack version
+    c1.inferHostVersions(entityHDP2);
+
+    hostVersionsH1After = hostVersionDAO.findByClusterAndHost("c1", "h1");
+    assertEquals(2, hostVersionsH1After.size());
+
+    checked = false;
+    for (HostVersionEntity entity : hostVersionsH1After) {
+      if (entity.getRepositoryVersion().getStack().equals("HDP")
+          && entity.getRepositoryVersion().getVersion().equals("0.2")) {
+        assertEquals(RepositoryVersionState.INSTALLING, entity.getState());
+        checked = true;
+        break;
+      }
+    }
+    assertTrue(checked);
+  }
+
+  @Test
+  public void testRecalculateClusterVersionState() throws AmbariException {
+    Host h1 = clusters.getHost("h1");
+    h1.setState(HostState.HEALTHY);
+    StackId stackId = new StackId("HDP-0.1");
+    RepositoryVersionEntity repositoryVersionEntity = helper.getOrCreateRepositoryVersion(stackId.getStackId(),
+        "1.0-1000");
+    c1.createClusterVersion(stackId.getStackId(), "1.0-1000", "admin", RepositoryVersionState.INSTALLING);
+    c1.setCurrentStackVersion(stackId);
+    c1.recalculateClusterVersionState("1.0-1000");
+    checkStackVersionState(stackId.getStackId(), "1.0-1000", RepositoryVersionState.OUT_OF_SYNC);
+
+    helper.createHostVersion("h1", repositoryVersionEntity, RepositoryVersionState.INSTALLING);
+    c1.recalculateClusterVersionState("1.0-1000");
+    checkStackVersionState(stackId.getStackId(), "1.0-1000", RepositoryVersionState.INSTALLING);
+
+    h1.setState(HostState.UNHEALTHY);
+    c1.recalculateClusterVersionState("1.0-1000");
+    checkStackVersionState(stackId.getStackId(), "1.0-1000", RepositoryVersionState.OUT_OF_SYNC);
+
+    c1.recalculateClusterVersionState("1.0-2086");
+    checkStackVersionState(stackId.getStackId(), "1.0-2086", RepositoryVersionState.CURRENT);
+  }
+
+  @Test
+  public void testRecalculateAllClusterVersionStates() throws AmbariException {
+    Host h1 = clusters.getHost("h1");
+    h1.setState(HostState.HEALTHY);
+    StackId stackId = new StackId("HDP-0.1");
+    RepositoryVersionEntity repositoryVersionEntity = helper.getOrCreateRepositoryVersion(stackId.getStackId(),
+        "1.0-1000");
+    c1.createClusterVersion(stackId.getStackId(), "1.0-1000", "admin", RepositoryVersionState.INSTALLING);
+    c1.setCurrentStackVersion(stackId);
+    c1.recalculateAllClusterVersionStates();
+    checkStackVersionState(stackId.getStackId(), "1.0-1000", RepositoryVersionState.OUT_OF_SYNC);
+    checkStackVersionState(stackId.getStackId(), "1.0-2086", RepositoryVersionState.CURRENT);
+
+    helper.createHostVersion("h1", repositoryVersionEntity, RepositoryVersionState.INSTALLING);
+    c1.recalculateAllClusterVersionStates();
+    checkStackVersionState(stackId.getStackId(), "1.0-1000", RepositoryVersionState.INSTALLING);
+    checkStackVersionState(stackId.getStackId(), "1.0-2086", RepositoryVersionState.CURRENT);
+  }
 }