AMBARI-5077. Provision a cluster via a single REST API invocation by specifying a blueprint.

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

@@ -21,6 +21,7 @@ package org.apache.ambari.server.controller;
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.actionmanager.ActionManager;
 import org.apache.ambari.server.api.services.AmbariMetaInfo;
+import org.apache.ambari.server.controller.internal.RequestStageContainer;
 import org.apache.ambari.server.scheduler.ExecutionScheduleManager;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
@@ -420,7 +421,7 @@ public interface AmbariManagementController {
   public String getAuthName();
 
   /**
-   * Create the stages required to persist an action and return a result containing the
+   * Create and persist the request stages and return a response containing the
    * associated request and resulting tasks.
    *
    * @param cluster             the cluster
@@ -437,14 +438,40 @@ public interface AmbariManagementController {
    *
    * @throws AmbariException is thrown if the stages can not be created
    */
-  public RequestStatusResponse createStages(Cluster cluster, Map<String, String> requestProperties,
-                                            Map<String, String> requestParameters,
-                                            Map<State, List<Service>> changedServices,
-                                            Map<State, List<ServiceComponent>> changedComponents,
-                                            Map<String, Map<State, List<ServiceComponentHost>>> changedHosts,
-                                            Collection<ServiceComponentHost> ignoredHosts,
-                                            boolean runSmokeTest, boolean reconfigureClients) throws AmbariException;
+  public RequestStatusResponse createAndPersistStages(Cluster cluster, Map<String, String> requestProperties,
+                                                      Map<String, String> requestParameters,
+                                                      Map<State, List<Service>> changedServices,
+                                                      Map<State, List<ServiceComponent>> changedComponents,
+                                                      Map<String, Map<State, List<ServiceComponentHost>>> changedHosts,
+                                                      Collection<ServiceComponentHost> ignoredHosts,
+                                                      boolean runSmokeTest, boolean reconfigureClients)
+                                                      throws AmbariException;
 
+  /**
+   * Add stages to the request.
+   *
+   * @param requestStages       Stages currently associated with request
+   * @param cluster             cluster being acted on
+   * @param requestProperties   the request properties
+   * @param requestParameters   the request parameters; may be null
+   * @param changedServices     the services being changed; may be null
+   * @param changedComponents   the components being changed
+   * @param changedHosts        the hosts being changed
+   * @param ignoredHosts        the hosts to be ignored
+   * @param runSmokeTest        indicates whether or not the smoke tests should be run
+   * @param reconfigureClients  indicates whether or not the clients should be reconfigured
+   *
+   * @return request stages
+   *
+   * @throws AmbariException if stages can't be created
+   */
+  public RequestStageContainer addStages(RequestStageContainer requestStages, Cluster cluster, Map<String, String> requestProperties,
+                                 Map<String, String> requestParameters,
+                                 Map<State, List<Service>> changedServices,
+                                 Map<State, List<ServiceComponent>> changedComponents,
+                                 Map<String, Map<State, List<ServiceComponentHost>>> changedHosts,
+                                 Collection<ServiceComponentHost> ignoredHosts,
+                                 boolean runSmokeTest, boolean reconfigureClients) throws AmbariException;
 
   /**
    * Getter for the url of JDK, stored at server resources folder
@@ -511,7 +538,7 @@ public interface AmbariManagementController {
 
   /**
    * Get the Factory to create Request schedules
-   * @return
+   * @return the request execution factory
    */
   public RequestExecutionFactory getRequestExecutionFactory();
 

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

@@ -37,13 +37,13 @@ import org.apache.ambari.server.ServiceNotFoundException;
 import org.apache.ambari.server.StackAccessException;
 import org.apache.ambari.server.actionmanager.ActionManager;
 import org.apache.ambari.server.actionmanager.HostRoleCommand;
-import org.apache.ambari.server.actionmanager.Request;
 import org.apache.ambari.server.actionmanager.RequestFactory;
 import org.apache.ambari.server.actionmanager.Stage;
 import org.apache.ambari.server.actionmanager.StageFactory;
 import org.apache.ambari.server.agent.ExecutionCommand;
 import org.apache.ambari.server.api.services.AmbariMetaInfo;
 import org.apache.ambari.server.configuration.Configuration;
+import org.apache.ambari.server.controller.internal.RequestStageContainer;
 import org.apache.ambari.server.controller.internal.URLStreamProvider;
 import org.apache.ambari.server.customactions.ActionDefinition;
 import org.apache.ambari.server.metadata.ActionMetadata;
@@ -589,11 +589,11 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
     }
   }
 
-  private Stage createNewStage(Cluster cluster, long requestId, String requestContext, String clusterHostInfo) {
+  private Stage createNewStage(long id, Cluster cluster, long requestId, String requestContext, String clusterHostInfo) {
     String logDir = BASE_LOG_DIR + File.pathSeparator + requestId;
     Stage stage =
         stageFactory.createNew(requestId, logDir, cluster.getClusterName(), requestContext, clusterHostInfo);
-    stage.setStageId(0);
+    stage.setStageId(id);
     return stage;
   }
 
@@ -1324,7 +1324,8 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
     }
   }
 
-  private List<Stage> doStageCreation(Cluster cluster,
+  private List<Stage> doStageCreation(RequestStageContainer requestStages,
+      Cluster cluster,
       Map<State, List<Service>> changedServices,
       Map<State, List<ServiceComponent>> changedComps,
       Map<String, Map<State, List<ServiceComponentHost>>> changedScHosts,
@@ -1360,7 +1361,6 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
     if (!changedScHosts.isEmpty()
         || !smokeTestServices.isEmpty()) {
       long nowTimestamp = System.currentTimeMillis();
-      Long requestId = actionManager.getNextRequestId();
 
       // FIXME cannot work with a single stage
       // multiple stages may be needed for reconfigure
@@ -1369,7 +1369,8 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
 
       String clusterHostInfoJson = StageUtils.getGson().toJson(clusterHostInfo);
 
-      Stage stage = createNewStage(cluster, requestId, requestContext, clusterHostInfoJson);
+      Stage stage = createNewStage(requestStages.getLastStageId() + 1, cluster,
+          requestStages.getId(), requestContext, clusterHostInfoJson);
 
       //HACK
       String jobtrackerHost = getJobTrackerHost(cluster);
@@ -1436,8 +1437,12 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
                 ComponentInfo compInfo = ambariMetaInfo.getComponentCategory(
                     stackId.getStackName(), stackId.getStackVersion(), scHost.getServiceName(),
                     scHost.getServiceComponentName());
-                if (oldSchState == State.INSTALLED
-                    || oldSchState == State.STARTING) {
+
+
+                if (oldSchState == State.INSTALLED ||
+                    oldSchState == State.STARTING ||
+                    requestStages.getProjectedState(scHost.getHostName(),
+                        scHost.getServiceComponentName()) == State.INSTALLED) {
                   roleCommand = RoleCommand.START;
                   event = new ServiceComponentHostStartEvent(
                       scHost.getServiceComponentName(), scHost.getHostName(),
@@ -1496,7 +1501,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
 
             if (LOG.isDebugEnabled()) {
               LOG.debug("Create a new host action"
-                  + ", requestId=" + requestId
+                  + ", requestId=" + requestStages.getId()
                   + ", componentName=" + scHost.getServiceComponentName()
                   + ", hostname=" + scHost.getHostName()
                   + ", roleCommand=" + roleCommand.name());
@@ -1573,22 +1578,6 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
     return hostLevelParams;
   }
 
-  private void persistStages(List<Stage> stages) throws AmbariException {
-    if (stages != null && !stages.isEmpty()) {
-      persistRequest(requestFactory.createNewFromStages(stages));
-    }
-  }
-
-  //TODO use when request created externally
-  private void persistRequest(Request request) {
-    if (request != null && request.getStages()!= null && !request.getStages().isEmpty()) {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug(String.format("Triggering Action Manager, request=%s", request));
-      }
-      actionManager.sendActions(request, null);
-    }
-  }
-
   @Transactional
   void updateServiceStates(
       Map<State, List<Service>> changedServices,
@@ -1638,23 +1627,40 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
   }
 
   @Override
-  public RequestStatusResponse createStages(Cluster cluster, Map<String, String> requestProperties,
-                                            Map<String, String> requestParameters,
-                                            Map<State, List<Service>> changedServices,
-                                            Map<State, List<ServiceComponent>> changedComponents,
-                                            Map<String, Map<State, List<ServiceComponentHost>>> changedHosts,
-                                            Collection<ServiceComponentHost> ignoredHosts,
-                                            boolean runSmokeTest, boolean reconfigureClients) throws AmbariException {
-
-    List<Stage> stages = doStageCreation(cluster, changedServices, changedComponents,
+  public RequestStatusResponse createAndPersistStages(Cluster cluster, Map<String, String> requestProperties,
+                                                      Map<String, String> requestParameters,
+                                                      Map<State, List<Service>> changedServices,
+                                                      Map<State, List<ServiceComponent>> changedComponents,
+                                                      Map<String, Map<State, List<ServiceComponentHost>>> changedHosts,
+                                                      Collection<ServiceComponentHost> ignoredHosts,
+                                                      boolean runSmokeTest, boolean reconfigureClients) throws AmbariException {
+
+    RequestStageContainer request = addStages(null, cluster, requestProperties, requestParameters, changedServices,
+        changedComponents, changedHosts, ignoredHosts, runSmokeTest, reconfigureClients);
+
+    request.persist();
+    return request.getRequestStatusResponse();
+  }
+
+  @Override
+  public RequestStageContainer addStages(RequestStageContainer requestStages, Cluster cluster, Map<String, String> requestProperties,
+                                 Map<String, String> requestParameters, Map<State, List<Service>> changedServices,
+                                 Map<State, List<ServiceComponent>> changedComponents,
+                                 Map<String, Map<State, List<ServiceComponentHost>>> changedHosts,
+                                 Collection<ServiceComponentHost> ignoredHosts, boolean runSmokeTest,
+                                 boolean reconfigureClients) throws AmbariException {
+
+    if (requestStages == null) {
+      requestStages = new RequestStageContainer(actionManager.getNextRequestId(), null, requestFactory, actionManager);
+    }
+
+    List<Stage> stages = doStageCreation(requestStages, cluster, changedServices, changedComponents,
         changedHosts, requestParameters, requestProperties.get(REQUEST_CONTEXT_PROPERTY),
         runSmokeTest, reconfigureClients);
 
-    persistStages(stages);
-
+    requestStages.addStages(stages);
     updateServiceStates(changedServices, changedComponents, changedHosts, ignoredHosts);
-
-    return stages == null || stages.isEmpty() ? null : getRequestStatusResponse(stages.get(0).getRequestId());
+    return requestStages;
   }
 
   @Override
@@ -1894,7 +1900,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
 
     Cluster cluster = clusters.getCluster(clusterNames.iterator().next());
 
-    return createStages(cluster, requestProperties, null, null, null, changedScHosts, ignoredScHosts, runSmokeTest,
+    return createAndPersistStages(cluster, requestProperties, null, null, null, changedScHosts, ignoredScHosts, runSmokeTest,
         false);
   }
 
@@ -2106,7 +2112,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
    * fully populates a request resource including the set of task sub-resources
    * in the request response.
    */
-  private RequestStatusResponse getRequestStatusResponse(long requestId) {
+  RequestStatusResponse getRequestStatusResponse(long requestId) {
     RequestStatusResponse response = new RequestStatusResponse(requestId);
     List<HostRoleCommand> hostRoleCommands =
         actionManager.getRequestTasks(requestId);
@@ -2360,7 +2366,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
         clusters.getHostsForCluster(cluster.getClusterName()), cluster);
 
     String clusterHostInfoJson = StageUtils.getGson().toJson(clusterHostInfo);
-    Stage stage = createNewStage(cluster, actionManager.getNextRequestId(), requestContext, clusterHostInfoJson);
+    Stage stage = createNewStage(0, cluster, actionManager.getNextRequestId(), requestContext, clusterHostInfoJson);
 
     Map<String, String> params = createDefaultHostParams(cluster);
 

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

@@ -43,6 +43,7 @@ import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
 import org.apache.ambari.server.configuration.Configuration;
 import org.apache.ambari.server.controller.internal.AbstractControllerResourceProvider;
 import org.apache.ambari.server.controller.internal.BlueprintResourceProvider;
+import org.apache.ambari.server.controller.internal.ClusterResourceProvider;
 import org.apache.ambari.server.controller.internal.StackDefinedPropertyProvider;
 import org.apache.ambari.server.controller.nagios.NagiosPropertyProvider;
 import org.apache.ambari.server.orm.GuiceJpaInitializer;
@@ -505,6 +506,7 @@ public class AmbariServer {
     NagiosPropertyProvider.init(injector);
     AbstractControllerResourceProvider.init(injector.getInstance(ResourceProviderFactory.class));
     BlueprintResourceProvider.init(injector.getInstance(BlueprintDAO.class));
+    ClusterResourceProvider.injectBlueprintDAO(injector.getInstance(BlueprintDAO.class));
   }
 
   public static void main(String[] args) throws Exception {

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

@@ -236,7 +236,7 @@ public class ClusterControllerImpl implements ClusterController {
           return null;
         }
       }
-        return provider.updateResources(request, predicate);
+      return provider.updateResources(request, predicate);
     }
     return null;
   }
@@ -262,6 +262,24 @@ public class ClusterControllerImpl implements ClusterController {
   }
 
 
+  // ----- ClusterControllerImpl ---------------------------------------------
+
+  /**
+   * Get the resource provider for the given type, creating it if required.
+   *
+   * @param type  the resource type
+   *
+   * @return the resource provider
+   */
+  public ResourceProvider ensureResourceProvider(Resource.Type type) {
+    synchronized (resourceProviders) {
+      if (!resourceProviders.containsKey(type)) {
+        resourceProviders.put(type, providerModule.getResourceProvider(type));
+      }
+    }
+    return resourceProviders.get(type);   }
+
+
   // ----- helper methods ----------------------------------------------------
 
   /**
@@ -444,22 +462,6 @@ public class ClusterControllerImpl implements ClusterController {
     return size > provider.checkPropertyIds(requestPropertyIds).size();
   }
 
-  /**
-   * Get the resource provider for the given type, creating it if required.
-   *
-   * @param type  the resource type
-   *
-   * @return the resource provider
-   */
-  private ResourceProvider ensureResourceProvider(Resource.Type type) {
-    synchronized (resourceProviders) {
-      if (!resourceProviders.containsKey(type)) {
-        resourceProviders.put(type, providerModule.getResourceProvider(type));
-      }
-    }
-    return resourceProviders.get(type);
-  }
-
   /**
    * Get the list of property providers for the given type.
    *

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

@@ -18,17 +18,27 @@
 package org.apache.ambari.server.controller.internal;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 
 import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.StackAccessException;
+import org.apache.ambari.server.api.services.PersistKeyValueService;
 import org.apache.ambari.server.controller.AmbariManagementController;
 import org.apache.ambari.server.controller.ClusterRequest;
 import org.apache.ambari.server.controller.ClusterResponse;
 import org.apache.ambari.server.controller.ConfigurationRequest;
 import org.apache.ambari.server.controller.RequestStatusResponse;
+import org.apache.ambari.server.controller.StackConfigurationRequest;
+import org.apache.ambari.server.controller.StackConfigurationResponse;
+import org.apache.ambari.server.controller.StackServiceComponentRequest;
+import org.apache.ambari.server.controller.StackServiceComponentResponse;
+import org.apache.ambari.server.controller.StackServiceRequest;
+import org.apache.ambari.server.controller.StackServiceResponse;
 import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
 import org.apache.ambari.server.controller.spi.NoSuchResourceException;
 import org.apache.ambari.server.controller.spi.Predicate;
@@ -36,14 +46,20 @@ import org.apache.ambari.server.controller.spi.Request;
 import org.apache.ambari.server.controller.spi.RequestStatus;
 import org.apache.ambari.server.controller.spi.Resource;
 import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException;
+import org.apache.ambari.server.controller.spi.ResourceProvider;
 import org.apache.ambari.server.controller.spi.SystemException;
 import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
+import org.apache.ambari.server.controller.utilities.ClusterControllerHelper;
 import org.apache.ambari.server.controller.utilities.PropertyHelper;
+import org.apache.ambari.server.orm.dao.BlueprintDAO;
+import org.apache.ambari.server.orm.entities.BlueprintEntity;
+import org.apache.ambari.server.orm.entities.HostGroupComponentEntity;
+import org.apache.ambari.server.orm.entities.HostGroupEntity;
 
 /**
  * Resource provider for cluster resources.
  */
-class ClusterResourceProvider extends AbstractControllerResourceProvider {
+public class ClusterResourceProvider extends AbstractControllerResourceProvider {
 
   // ----- Property ID constants ---------------------------------------------
 
@@ -52,12 +68,31 @@ class ClusterResourceProvider extends AbstractControllerResourceProvider {
   protected static final String CLUSTER_NAME_PROPERTY_ID    = PropertyHelper.getPropertyId("Clusters", "cluster_name");
   protected static final String CLUSTER_VERSION_PROPERTY_ID = PropertyHelper.getPropertyId("Clusters", "version");
   protected static final String CLUSTER_DESIRED_CONFIGS_PROPERTY_ID = PropertyHelper.getPropertyId("Clusters", "desired_configs");
+  protected static final String BLUEPRINT_PROPERTY_ID = PropertyHelper.getPropertyId(null, "blueprint");
 
 
   private static Set<String> pkPropertyIds =
       new HashSet<String>(Arrays.asList(new String[]{
           CLUSTER_ID_PROPERTY_ID}));
 
+  /**
+   * Data access object used to obtain blueprint entities.
+   */
+  private static BlueprintDAO blueprintDAO;
+
+  /**
+   * Maps properties to updaters which update the property when provisioning a cluster via a blueprint
+   */
+  private Map<String, PropertyUpdater> propertyUpdaters =
+      new HashMap<String, PropertyUpdater>();
+
+  /**
+   * Maps configuration type (string) to associated properties
+   */
+  private Map<String, Map<String, String>> mapClusterConfigurations =
+      new HashMap<String, Map<String, String>>();
+
+
   // ----- Constructors ----------------------------------------------------
 
   /**
@@ -70,9 +105,21 @@ class ClusterResourceProvider extends AbstractControllerResourceProvider {
   ClusterResourceProvider(Set<String> propertyIds,
                           Map<Resource.Type, String> keyPropertyIds,
                           AmbariManagementController managementController) {
+
     super(propertyIds, keyPropertyIds, managementController);
+    registerPropertyUpdaters();
+  }
+
+  /**
+   * Inject the blueprint data access object which is used to obtain blueprint entities
+   *
+   * @param dao  blueprint data access object
+   */
+  public static void injectBlueprintDAO(BlueprintDAO dao) {
+    blueprintDAO = dao;
   }
 
+
 // ----- ResourceProvider ------------------------------------------------
 
   @Override
@@ -82,18 +129,17 @@ class ClusterResourceProvider extends AbstractControllerResourceProvider {
              ResourceAlreadyExistsException,
              NoSuchParentResourceException {
 
+    RequestStatusResponse createResponse = null;
     for (final Map<String, Object> properties : request.getProperties()) {
-      createResources(new Command<Void>() {
-        @Override
-        public Void invoke() throws AmbariException {
-          getManagementController().createCluster(getRequest(properties));
-          return null;
-        }
-      });
+      if (isCreateFromBlueprint(properties)) {
+        createResponse = processBlueprintCreate(properties);
+      } else {
+        createClusterResource(properties);
+      }
     }
-    notifyCreate(Resource.Type.Cluster, request);
 
-    return getRequestStatus(null);
+    notifyCreate(Resource.Type.Cluster, request);
+    return getRequestStatus(createResponse);
   }
 
   @Override
@@ -183,13 +229,27 @@ class ClusterResourceProvider extends AbstractControllerResourceProvider {
     return getRequestStatus(null);
   }
 
-  // ----- utility methods -------------------------------------------------
-
   @Override
   protected Set<String> getPKPropertyIds() {
     return pkPropertyIds;
   }
 
+  /**
+   * {@inheritDoc}  Overridden to support configuration.
+   */
+  @Override
+  public Set<String> checkPropertyIds(Set<String> propertyIds) {
+    Set<String> baseUnsupported = super.checkPropertyIds(propertyIds);
+
+    // extract to own method
+    baseUnsupported.remove("blueprint");
+    baseUnsupported.remove("host-groups");
+
+    return checkConfigPropertyIds(baseUnsupported, "Clusters");
+  }
+
+  // ----- utility methods -------------------------------------------------
+
   /**
    * Get a cluster request object from a map of property values.
    *
@@ -212,14 +272,936 @@ class ClusterResourceProvider extends AbstractControllerResourceProvider {
     
     return cr;
   }
-  
+
   /**
-   * {@inheritDoc}  Overridden to support configuration.
+   * Determine if the request is a create using a blueprint.
+   *
+   * @param properties  request properties
+   *
+   * @return true if request is a create using a blueprint; false otherwise
    */
-  @Override
-  public Set<String> checkPropertyIds(Set<String> propertyIds) {
-    Set<String> baseUnsupported = super.checkPropertyIds(propertyIds);
-    
-    return checkConfigPropertyIds(baseUnsupported, "Clusters");
+  private boolean isCreateFromBlueprint(Map<String, Object> properties) {
+    return properties.get("blueprint") != null;
+  }
+
+  /**
+   * Process a create request specifying a blueprint.  This includes creation of all resources,
+   * setting of configuration and installing and starting of all services.  The end result of this
+   * call will be a running cluster based on the topology and configuration specified in the blueprint.
+   *
+   * @param properties  request body properties
+   *
+   * @return asynchronous response information
+   *
+   * @throws ResourceAlreadyExistsException if cluster already exists
+   * @throws SystemException                if an unexpected exception occurs
+   * @throws UnsupportedPropertyException   if an invalid property is specified in the request
+   * @throws NoSuchParentResourceException  if a necessary parent resource doesn't exist
+   */
+  private RequestStatusResponse processBlueprintCreate(Map<String, Object> properties)
+      throws ResourceAlreadyExistsException, SystemException, UnsupportedPropertyException,
+      NoSuchParentResourceException {
+
+    String blueprintName = (String) properties.get(BLUEPRINT_PROPERTY_ID);
+
+    LOG.info("Creating Cluster '" + properties.get(CLUSTER_NAME_PROPERTY_ID) +
+        "' based on blueprint '" + blueprintName + "'.");
+
+    //todo: build up a proper topology object
+    BlueprintEntity blueprint = getBlueprint(blueprintName);
+    Stack stack = parseStack(blueprint);
+
+    Map<String, HostGroup> blueprintHostGroups = parseBlueprintHostGroups(blueprint, stack);
+    applyRequestInfoToHostGroups(properties, blueprintHostGroups);
+    processConfigurations(stack, blueprintHostGroups);
+
+    String clusterName = (String) properties.get(CLUSTER_NAME_PROPERTY_ID);
+    createClusterResource(buildClusterResourceProperties(stack, clusterName));
+    setConfigurationsOnCluster(clusterName);
+
+    Set<String> services = getServicesToDeploy(stack, blueprintHostGroups);
+
+    createServiceAndComponentResources(blueprintHostGroups, clusterName, services);
+    createHostAndComponentResources(blueprintHostGroups, clusterName);
+
+    persistInstallStateForUI();
+    return ((ServiceResourceProvider) getResourceProviderByType(Resource.Type.Service)).
+        installAndStart(clusterName);
+  }
+
+  /**
+   * Obtain a blueprint entity based on name.
+   *
+   * @param blueprintName  name of blueprint to obtain
+   *
+   * @return blueprint entity for the given name
+   * @throws IllegalArgumentException no blueprint with the given name found
+   */
+  private BlueprintEntity getBlueprint(String blueprintName) {
+    BlueprintEntity blueprint = blueprintDAO.findByName(blueprintName);
+    if (blueprint == null) {
+      throw new IllegalArgumentException("Specified blueprint doesn't exist: " + blueprintName);
+    }
+    return blueprint;
+  }
+
+  /**
+   * Create service and component resources.
+   *
+   * @param blueprintHostGroups  host groups contained in blueprint
+   * @param clusterName          cluster name
+   * @param services             services to be deployed
+   *
+   * @throws SystemException                an unexpected exception occurred
+   * @throws UnsupportedPropertyException   an unsupported property was specified in the request
+   * @throws ResourceAlreadyExistsException attempted to create a service or component that already exists
+   * @throws NoSuchParentResourceException  a required parent resource is missing
+   */
+  private void createServiceAndComponentResources(Map<String, HostGroup> blueprintHostGroups,
+                                                  String clusterName, Set<String> services)
+                                                  throws SystemException,
+                                                         UnsupportedPropertyException,
+                                                         ResourceAlreadyExistsException,
+                                                         NoSuchParentResourceException {
+
+    Set<Map<String, Object>> setServiceRequestProps = new HashSet<Map<String, Object>>();
+    for (String service : services) {
+      Map<String, Object> serviceProperties = new HashMap<String, Object>();
+      serviceProperties.put(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID, clusterName);
+      serviceProperties.put(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID, service);
+      setServiceRequestProps.add(serviceProperties);
+    }
+
+    Request serviceRequest = new RequestImpl(null, setServiceRequestProps, null, null);
+    getResourceProviderByType(Resource.Type.Service).createResources(serviceRequest);
+    createComponentResources(blueprintHostGroups, clusterName, services);
+  }
+
+  /**
+   * Build the cluster properties necessary for creating a cluster resource.
+   *
+   * @param stack        associated stack
+   * @param clusterName  cluster name
+   * @return map of cluster properties used to create a cluster resource
+   */
+  private Map<String, Object> buildClusterResourceProperties(Stack stack, String clusterName) {
+    Map<String, Object> clusterProperties = new HashMap<String, Object>();
+    clusterProperties.put(CLUSTER_NAME_PROPERTY_ID, clusterName);
+    clusterProperties.put(CLUSTER_VERSION_PROPERTY_ID, stack.name + "-" + stack.version);
+    return clusterProperties;
+  }
+
+  /**
+   * Create host and host_component resources.
+   *
+   * @param blueprintHostGroups  host groups specified in blueprint
+   * @param clusterName          cluster name
+   *
+   * @throws SystemException                an unexpected exception occurred
+   * @throws UnsupportedPropertyException   an invalid property was specified
+   * @throws ResourceAlreadyExistsException attempt to create a host or host_component which already exists
+   * @throws NoSuchParentResourceException  a required parent resource is missing
+   */
+  //todo: ambari agent must already be installed and registered on all hosts
+  private void createHostAndComponentResources(Map<String, HostGroup> blueprintHostGroups, String clusterName)
+      throws SystemException, UnsupportedPropertyException, ResourceAlreadyExistsException, NoSuchParentResourceException {
+
+    ResourceProvider hostProvider = getResourceProviderByType(Resource.Type.Host);
+    ResourceProvider hostComponentProvider = getResourceProviderByType(Resource.Type.HostComponent);
+    for (HostGroup group : blueprintHostGroups.values()) {
+      for (String host : group.getHostInfo()) {
+        Map<String, Object> hostProperties = new HashMap<String, Object>();
+        hostProperties.put("Hosts/cluster_name", clusterName);
+        hostProperties.put("Hosts/host_name", host);
+
+        hostProvider.createResources(new RequestImpl(
+            null, Collections.singleton(hostProperties), null, null));
+
+        // create clusters/hosts/host_components
+        Set<Map<String, Object>> setHostComponentRequestProps = new HashSet<Map<String, Object>>();
+        for (String hostComponent : group.getComponents()) {
+          // AMBARI_SERVER is not recognized by Ambari as a component
+          if (! hostComponent.equals("AMBARI_SERVER")) {
+            Map<String, Object> hostComponentProperties = new HashMap<String, Object>();
+            hostComponentProperties.put("HostRoles/cluster_name", clusterName);
+            hostComponentProperties.put("HostRoles/host_name", host);
+            hostComponentProperties.put("HostRoles/component_name", hostComponent);
+            setHostComponentRequestProps.add(hostComponentProperties);
+          }
+        }
+        hostComponentProvider.createResources(new RequestImpl(
+            null, setHostComponentRequestProps, null, null));
+      }
+    }
+  }
+
+  /**
+   * Create component resources.
+   *
+   * @param blueprintHostGroups  host groups specified in blueprint
+   * @param clusterName          cluster name
+   * @param services             services to be deployed
+   *
+   * @throws SystemException                an unexpected exception occurred
+   * @throws UnsupportedPropertyException   an invalid property was specified
+   * @throws ResourceAlreadyExistsException attempt to create a component which already exists
+   * @throws NoSuchParentResourceException  a required parent resource is missing
+   */
+  private void createComponentResources(Map<String, HostGroup> blueprintHostGroups,
+                                        String clusterName, Set<String> services)
+                                        throws SystemException,
+                                               UnsupportedPropertyException,
+                                               ResourceAlreadyExistsException,
+                                               NoSuchParentResourceException {
+    for (String service : services) {
+      Set<String> components = new HashSet<String>();
+      for (HostGroup hostGroup : blueprintHostGroups.values()) {
+        Collection<String> serviceComponents = hostGroup.getComponents(service);
+        if (serviceComponents != null && !serviceComponents.isEmpty()) {
+          components.addAll(serviceComponents);
+        }
+      }
+
+      Set<Map<String, Object>> setComponentRequestProps = new HashSet<Map<String, Object>>();
+      for (String component : components) {
+        Map<String, Object> componentProperties = new HashMap<String, Object>();
+        componentProperties.put("ServiceComponentInfo/cluster_name", clusterName);
+        componentProperties.put("ServiceComponentInfo/service_name", service);
+        componentProperties.put("ServiceComponentInfo/component_name", component);
+        setComponentRequestProps.add(componentProperties);
+      }
+      Request componentRequest = new RequestImpl(null, setComponentRequestProps, null, null);
+      ResourceProvider componentProvider = getResourceProviderByType(Resource.Type.Component);
+      componentProvider.createResources(componentRequest);
+    }
+  }
+
+  /**
+   * Set all configurations on the cluster resource.
+   *
+   * @param clusterName  cluster name
+   *
+   * @throws SystemException an unexpected exception occurred
+   */
+  private void setConfigurationsOnCluster(String clusterName) throws SystemException {
+    for (Map.Entry<String, Map<String, String>> entry : mapClusterConfigurations.entrySet()) {
+      String type = entry.getKey();
+      if (type.endsWith(".xml")) {
+        type = type.substring(0, type.length() - 4);
+      }
+      try {
+        //todo: properly handle non system exceptions
+        setConfigurationsOnCluster(clusterName, type, entry.getValue());
+      } catch (AmbariException e) {
+        throw new SystemException("Unable to set configurations on cluster.", e);
+      }
+    }
+  }
+
+  /**
+   * Set configuration of a specific type on the cluster resource.
+   *
+   * @param clusterName  cluster name
+   * @param type         configuration type that is to be set
+   * @param properties   properties to set
+   *
+   * @throws AmbariException if an exception occurs setting the properties
+   */
+  private void setConfigurationsOnCluster(String clusterName, String type,
+                                          Map<String, String> properties) throws AmbariException {
+
+    Map<String, Object> clusterProperties = new HashMap<String, Object>();
+    clusterProperties.put(CLUSTER_NAME_PROPERTY_ID, clusterName);
+    clusterProperties.put(CLUSTER_DESIRED_CONFIGS_PROPERTY_ID + "/type", type);
+    clusterProperties.put(CLUSTER_DESIRED_CONFIGS_PROPERTY_ID + "/tag", "1");
+    for (Map.Entry<String, String> entry : properties.entrySet()) {
+      clusterProperties.put(CLUSTER_DESIRED_CONFIGS_PROPERTY_ID +
+          "/properties/" + entry.getKey(), entry.getValue());
+    }
+
+    getManagementController().updateClusters(
+        Collections.singleton(getRequest(clusterProperties)), null);
+  }
+
+  /**
+   * Apply the information contained in the cluster request body such as host an configuration properties to
+   * the associated blueprint.
+   *
+   * @param properties           request properties
+   * @param blueprintHostGroups  blueprint host groups
+   *
+   * @throws IllegalArgumentException a host-group in the request doesn't match a host-group in the blueprint
+   */
+  private void applyRequestInfoToHostGroups(Map<String, Object> properties, Map<String,
+                                            HostGroup> blueprintHostGroups)
+                                            throws IllegalArgumentException {
+
+    @SuppressWarnings("unchecked")
+    Collection<Map<String, Object>> hostGroups =
+        (Collection<Map<String, Object>>) properties.get("host-groups");
+
+    // iterate over host groups provided in request body
+    for (Map<String, Object> hostGroupProperties : hostGroups) {
+      String name = (String) hostGroupProperties.get("name");
+      HostGroup hostGroup = blueprintHostGroups.get(name);
+
+      if (hostGroup == null) {
+        throw new IllegalArgumentException("Invalid host-group specified: " + name +
+          ".  All request host groups must have a corresponding host group in the specified blueprint");
+      }
+
+      Collection hosts = (Collection) hostGroupProperties.get("hosts");
+      for (Object oHost : hosts) {
+        @SuppressWarnings("unchecked")
+        Map<String, String> mapHostProperties = (Map<String, String>) oHost;
+        //add host information to host group
+        hostGroup.addHostInfo(mapHostProperties.get("fqdn"));
+      }
+    }
+  }
+
+  /**
+   * Parse blueprint host groups.
+   *
+   * @param blueprint  associated blueprint
+   * @param stack      associated stack
+   *
+   * @return map of host group name to host group
+   */
+  private Map<String, HostGroup> parseBlueprintHostGroups(BlueprintEntity blueprint, Stack stack) {
+    Map<String, HostGroup> mapHostGroups = new HashMap<String, HostGroup>();
+
+    for (HostGroupEntity hostGroup : blueprint.getHostGroups()) {
+      mapHostGroups.put(hostGroup.getName(), new HostGroup(hostGroup, stack));
+    }
+    return mapHostGroups;
+  }
+
+  /**
+   * Parse stack information.
+   *
+   * @param blueprint  associated blueprint
+   *
+   * @return stack instance
+   *
+   * @throws SystemException an unexpected exception occurred
+   */
+  private Stack parseStack(BlueprintEntity blueprint) throws SystemException {
+    Stack stack;
+    try {
+      stack = new Stack(blueprint.getStackName(), blueprint.getStackVersion());
+    } catch (StackAccessException e) {
+      throw new IllegalArgumentException("Invalid stack information provided for cluster.  " +
+          "stack name: " + blueprint.getStackName() +
+          " stack version: " + blueprint.getStackVersion());
+    } catch (AmbariException e) {
+      //todo: review all exception handling associated with cluster creation via blueprint
+      throw new SystemException("Unable to obtain stack information.", e);
+    }
+    return stack;
+  }
+
+  /**
+   * Create the cluster resource.
+   *
+   * @param properties  cluster resource request properties
+   *
+   * @throws ResourceAlreadyExistsException  cluster resource already exists
+   * @throws SystemException                 an unexpected exception occurred
+   * @throws NoSuchParentResourceException   shouldn't be thrown as a cluster doesn't have a parent resource
+   */
+  private void createClusterResource(final Map<String, Object> properties)
+      throws ResourceAlreadyExistsException, SystemException, NoSuchParentResourceException {
+
+    createResources(new Command<Void>() {
+      @Override
+      public Void invoke() throws AmbariException {
+        getManagementController().createCluster(getRequest(properties));
+        return null;
+      }
+    });
+  }
+  /**
+   * Obtain a resource provider based on type.
+   *
+   * @param type  resource provider type
+   *
+   * @return resource provider for the specified type
+   */
+  //todo: inject a better mechanism for getting resource providers
+  ResourceProvider getResourceProviderByType(Resource.Type type) {
+    return ((ClusterControllerImpl) ClusterControllerHelper.getClusterController()).
+        ensureResourceProvider(type);
+  }
+
+  /**
+   * Persist cluster state for the ambari UI.  Setting this state informs that UI that a cluster has been
+   * installed and started and that the monitoring screen for the cluster should be displayed to the user.
+   *
+   * @throws SystemException if an unexpected exception occurs
+   */
+  private void persistInstallStateForUI() throws SystemException {
+    PersistKeyValueService persistService = new PersistKeyValueService();
+    try {
+      persistService.update("{\"CLUSTER_CURRENT_STATUS\": \"{\\\"clusterState\\\":\\\"CLUSTER_STARTED_5\\\"}\"}");
+    } catch (Exception e) {
+      throw new SystemException("Unable to finalize state of cluster for UI.  " +
+          "Cluster creation will not be affected but the cluster may be inaccessible by Ambari UI." );
+    }
+  }
+
+  /**
+   * Process cluster configurations.  This includes obtaining the default configuration properties from the stack,
+   * overlaying configuration properties specified in the cluster create request and updating properties with
+   * topology specific information.
+   *
+   * @param stack                associated stack
+   * @param blueprintHostGroups  host groups contained in the blueprint
+   */
+  // processing at cluster level only now
+  public void processConfigurations(Stack stack, Map<String, HostGroup> blueprintHostGroups)  {
+    Set<String> services = getServicesToDeploy(stack, blueprintHostGroups);
+
+    for (String service : services) {
+      Collection<String> configTypes = stack.getConfigurationTypes(service);
+      for (String type : configTypes) {
+        Map<String, String> properties = stack.getConfigurationProperties(service, type);
+        for (Map.Entry<String, String> entry : properties.entrySet()) {
+          String propName = entry.getKey();
+          String value    = entry.getValue();
+
+          Map<String, String> typeProps = mapClusterConfigurations.get(type);
+          if (typeProps == null) {
+            typeProps = new HashMap<String, String>();
+            mapClusterConfigurations.put(type, typeProps);
+          }
+          // todo: should an exception be thrown if a property is included under multiple services
+          if (! typeProps.containsKey(propName)) {
+            // see if property needs to be updated
+            PropertyUpdater propertyUpdater = propertyUpdaters.get(propName);
+            if (propertyUpdater != null) {
+              value = propertyUpdater.update(blueprintHostGroups, entry.getValue());
+            }
+            typeProps.put(propName, value);
+          }
+        }
+      }
+    }
+    // AMBARI-4921
+    //todo: hard-coding default values for required global config properties which are not in stack definition
+    Map<String, String> globalProperties = mapClusterConfigurations.get("global.xml");
+    if (globalProperties == null) {
+      globalProperties = new HashMap<String, String>();
+      mapClusterConfigurations.put("global.xml", globalProperties);
+    }
+    globalProperties.put("user_group", "hadoop");
+    globalProperties.put("smokeuser", "ambari-qa");
+    globalProperties.put("nagios_contact", "default@REPLACEME.NOWHERE");
+    globalProperties.put("nagios_web_password", "admin");
+  }
+
+  /**
+   * Get set of services which are to be deployed.
+   *
+   * @param stack                stack information
+   * @param blueprintHostGroups  host groups contained in blueprint
+   *
+   * @return set of service names which will be deployed
+   */
+  private Set<String> getServicesToDeploy(Stack stack, Map<String, HostGroup> blueprintHostGroups) {
+    Set<String> services = new HashSet<String>();
+    for (HostGroup group : blueprintHostGroups.values()) {
+      if (! group.getHostInfo().isEmpty()) {
+        services.addAll(stack.getServicesForComponents(group.getComponents()));
+      }
+    }
+    //remove entry associated with Ambari Server since this isn't recognized by Ambari
+    services.remove(null);
+
+    return services;
+  }
+
+  /**
+   * Register updaters for configuration properties.
+   */
+  private void registerPropertyUpdaters() {
+    // NAMENODE
+    propertyUpdaters.put("dfs.http.address", new SingleHostPropertyUpdater("NAMENODE"));
+    propertyUpdaters.put("dfs.namenode.http-address", new SingleHostPropertyUpdater("NAMENODE"));
+    propertyUpdaters.put("dfs.https.address", new SingleHostPropertyUpdater("NAMENODE"));
+    propertyUpdaters.put("dfs.namenode.https-address", new SingleHostPropertyUpdater("NAMENODE"));
+    propertyUpdaters.put("fs.default.name", new SingleHostPropertyUpdater("NAMENODE"));
+    propertyUpdaters.put("fs.defaultFS", new SingleHostPropertyUpdater("NAMENODE"));
+    propertyUpdaters.put("hbase.rootdir", new SingleHostPropertyUpdater("NAMENODE"));
+
+    // SECONDARY_NAMENODE
+    propertyUpdaters.put("dfs.secondary.http.address", new SingleHostPropertyUpdater("SECONDARY_NAMENODE"));
+    propertyUpdaters.put("dfs.namenode.secondary.http-address", new SingleHostPropertyUpdater("SECONDARY_NAMENODE"));
+
+    // HISTORY_SERVER
+    propertyUpdaters.put("yarn.log.server.url", new SingleHostPropertyUpdater("HISTORYSERVER"));
+    propertyUpdaters.put("mapreduce.jobhistory.webapp.address", new SingleHostPropertyUpdater("HISTORYSERVER"));
+    propertyUpdaters.put("mapreduce.jobhistory.address", new SingleHostPropertyUpdater("HISTORYSERVER"));
+
+    // RESOURCEMANAGER
+    propertyUpdaters.put("yarn.resourcemanager.hostname", new SingleHostPropertyUpdater("RESOURCEMANAGER"));
+    propertyUpdaters.put("yarn.resourcemanager.resource-tracker.address", new SingleHostPropertyUpdater("RESOURCEMANAGER"));
+    propertyUpdaters.put("yarn.resourcemanager.webapp.address", new SingleHostPropertyUpdater("RESOURCEMANAGER"));
+    propertyUpdaters.put("yarn.resourcemanager.scheduler.address", new SingleHostPropertyUpdater("RESOURCEMANAGER"));
+    propertyUpdaters.put("yarn.resourcemanager.address", new SingleHostPropertyUpdater("RESOURCEMANAGER"));
+    propertyUpdaters.put("yarn.resourcemanager.admin.address", new SingleHostPropertyUpdater("RESOURCEMANAGER"));
+
+    // JOBTRACKER
+    propertyUpdaters.put("mapred.job.tracker", new SingleHostPropertyUpdater("JOBTRACKER"));
+    propertyUpdaters.put("mapred.job.tracker.http.address", new SingleHostPropertyUpdater("JOBTRACKER"));
+    propertyUpdaters.put("mapreduce.history.server.http.address", new SingleHostPropertyUpdater("JOBTRACKER"));
+
+    // HIVE_SERVER
+    propertyUpdaters.put("hive.metastore.uris", new SingleHostPropertyUpdater("HIVE_SERVER"));
+    propertyUpdaters.put("hive_ambari_host", new SingleHostPropertyUpdater("HIVE_SERVER"));
+
+    // OOZIE_SERVER
+    propertyUpdaters.put("oozie.base.url", new SingleHostPropertyUpdater("OOZIE_SERVER"));
+    propertyUpdaters.put("oozie_ambari_host", new SingleHostPropertyUpdater("OOZIE_SERVER"));
+
+    // ZOOKEEPER_SERVER
+    propertyUpdaters.put("hbase.zookeeper.quorum", new MultipleHostPropertyUpdater("ZOOKEEPER_SERVER"));
+    propertyUpdaters.put("templeton.zookeeper.hosts", new MultipleHostPropertyUpdater("ZOOKEEPER_SERVER"));
+
+    // properties which need "m' appended.  Required due to AMBARI-4933
+    propertyUpdaters.put("namenode_heapsize", new MPropertyUpdater());
+    propertyUpdaters.put("namenode_opt_newsize", new MPropertyUpdater());
+    propertyUpdaters.put("namenode_opt_maxnewsize", new MPropertyUpdater());
+    propertyUpdaters.put("dtnode_heapsize", new MPropertyUpdater());
+    propertyUpdaters.put("jtnode_opt_newsize", new MPropertyUpdater());
+    propertyUpdaters.put("jtnode_opt_maxnewsize", new MPropertyUpdater());
+    propertyUpdaters.put("jtnode_heapsize", new MPropertyUpdater());
+    propertyUpdaters.put("hbase_master_heapsize", new MPropertyUpdater());
+    propertyUpdaters.put("hbase_regionserver_heapsize", new MPropertyUpdater());
+  }
+
+  /**
+   * Get host groups which contain a component.
+   *
+   * @param component   component name
+   * @param hostGroups  collection of host groups to check
+   *
+   * @return collection of host groups which contain the specified component
+   */
+  private Collection<HostGroup> getHostGroupsForComponent(String component, Collection<HostGroup> hostGroups) {
+    Collection<HostGroup> resultGroups = new HashSet<HostGroup>();
+    for (HostGroup group : hostGroups ) {
+      if (group.getComponents().contains(component)) {
+        resultGroups.add(group);
+      }
+    }
+    return resultGroups;
+  }
+
+
+  // ----- Inner Classes -----------------------------------------------------
+
+  /**
+   * Encapsulates stack information.
+   */
+  private class Stack {
+    /**
+     * Stack name
+     */
+    private String name;
+
+    /**
+     * Stack version
+     */
+    private String version;
+
+    /**
+     * Map of service name to components
+     */
+    private Map<String, Collection<String>> serviceComponents = new HashMap<String, Collection<String>>();
+
+    /**
+     * Map of component to service
+     */
+    private Map<String, String> componentService = new HashMap<String, String>();
+
+    /**
+     * Map of service to config type properties
+     */
+    private Map<String, Map<String, Map<String, String>>> serviceConfigurations =
+        new HashMap<String, Map<String, Map<String, String>>>();
+
+    /**
+     * Constructor.
+     *
+     * @param name     stack name
+     * @param version  stack version
+     *
+     * @throws AmbariException an exception occurred getting stack information
+     *                         for the specified name and version
+     */
+    public Stack(String name, String version) throws AmbariException {
+      this.name = name;
+      this.version = version;
+
+      Set<StackServiceResponse> stackServices = getManagementController().getStackServices(
+          Collections.singleton(new StackServiceRequest(name, version, null)));
+
+      for (StackServiceResponse stackService : stackServices) {
+        String serviceName = stackService.getServiceName();
+        parseComponents(serviceName);
+        parseConfigurations(serviceName);
+      }
+    }
+
+    /**
+     * Get services contained in the stack.
+     *
+     * @return collection of all services for the stack
+     */
+    public Collection<String> getServices() {
+      return serviceComponents.keySet();
+    }
+
+    /**
+     * Get components contained in the stack for the specified service.
+     *
+     * @param service  service name
+     *
+     * @return collection of component names for the specified service
+     */
+    public Collection<String> getComponents(String service) {
+      return serviceComponents.get(service);
+    }
+
+    /**
+     * Get configuration types for the specified service.
+     *
+     * @param service  service name
+     *
+     * @return collection of configuration types for the specified service
+     */
+    public Collection<String> getConfigurationTypes(String service) {
+      return serviceConfigurations.get(service).keySet();
+    }
+
+    /**
+     * Get config properties for the specified service and configuration type.
+     *
+     * @param service  service name
+     * @param type     configuration type
+     *
+     * @return map of property names to values for the specified service and configuration type
+     */
+    public Map<String, String> getConfigurationProperties(String service, String type) {
+      return serviceConfigurations.get(service).get(type);
+    }
+
+    /**
+     * Get the service for the specified component.
+     *
+     * @param component  component name
+     *
+     * @return service name that contains tha specified component
+     */
+    public String getServiceForComponent(String component) {
+      return componentService.get(component);
+    }
+
+    /**
+     * Get the names of the services which contains the specified components.
+     *
+     * @param components collection of components
+     *
+     * @return collection of services which contain the specified components
+     */
+    public Collection<String> getServicesForComponents(Collection<String> components) {
+      Set<String> services = new HashSet<String>();
+      for (String component : components) {
+        services.add(getServiceForComponent(component));
+      }
+
+      return services;
+    }
+
+    /**
+     * Parse components for the specified service from the stack definition.
+     *
+     * @param service  service name
+     *
+     * @throws AmbariException an exception occurred getting components from the stack definition
+     */
+    private void parseComponents(String service) throws AmbariException{
+      Collection<String> componentSet = new HashSet<String>();
+
+      Set<StackServiceComponentResponse> components = getManagementController().getStackComponents(
+          Collections.singleton(new StackServiceComponentRequest(name, version, service, null)
+      ));
+
+      // stack service components
+      for (StackServiceComponentResponse component : components) {
+        String componentName = component.getComponentName();
+        componentSet.add(componentName);
+        componentService.put(componentName, service);
+      }
+      this.serviceComponents.put(service, componentSet);
+    }
+
+    /**
+     * Parse configurations for the specified service from the stack definition.
+     *
+     * @param service  service name
+     *
+     * @throws AmbariException an exception occurred getting configurations from the stack definition
+     */
+    private void parseConfigurations(String service) throws AmbariException {
+      Map<String, Map<String, String>> mapServiceConfig =
+          new HashMap<String, Map<String, String>>();
+
+      serviceConfigurations.put(service, mapServiceConfig);
+
+      Set<StackConfigurationResponse> serviceConfigs = getManagementController().getStackConfigurations(
+          Collections.singleton(new StackConfigurationRequest(name, version, service, null)
+      ));
+
+      for (StackConfigurationResponse config : serviceConfigs) {
+        String type = config.getType();
+        Map<String, String> mapTypeConfig = mapServiceConfig.get(type);
+        if (mapTypeConfig == null) {
+          mapTypeConfig = new HashMap<String, String>();
+          mapServiceConfig.put(type, mapTypeConfig);
+        }
+        mapTypeConfig.put(config.getPropertyName(), config.getPropertyValue());
+      }
+    }
+  }
+
+  /**
+   * Host group representation.
+   */
+  private class HostGroup {
+    /**
+     * Host group entity
+     */
+    private HostGroupEntity hostGroup;
+
+    /**
+     * Components contained in the host group
+     */
+    private Collection<String> components = new HashSet<String>();
+
+    /**
+     * Hosts contained associated with the host group
+     */
+    private Collection<String> hosts = new HashSet<String>();
+
+    /**
+     * Map of service to components for the host group
+     */
+    private Map<String, Set<String>> componentsForService = new HashMap<String, Set<String>>();
+
+    /**
+     * Associated stack
+     */
+    private Stack stack;
+
+    /**
+     * Constructor.
+     *
+     * @param hostGroup  host group
+     * @param stack      stack
+     */
+    public HostGroup(HostGroupEntity hostGroup, Stack stack) {
+      this.hostGroup = hostGroup;
+      this.stack = stack;
+      parseComponents();
+    }
+
+    /**
+     * Associate a host with the host group.
+     *
+     * @param fqdn  fully qualified domain name of the host being added
+     */
+    public void addHostInfo(String fqdn) {
+      this.hosts.add(fqdn);
+    }
+
+    /**
+     * Get associated host information.
+     *
+     * @return collection of hosts associated with the host group
+     */
+    public Collection<String> getHostInfo() {
+      return this.hosts;
+    }
+
+    /**
+     * Get the components associated with the host group.
+     *
+     * @return  collection of component names for the host group
+     */
+    public Collection<String> getComponents() {
+      return this.components;
+    }
+
+    /**
+     * Get the components for the specified service which are associated with the host group.
+     *
+     * @param service  service name
+     *
+     * @return set of component names
+     */
+    public Collection<String> getComponents(String service) {
+      return componentsForService.get(service);
+    }
+
+    /**
+     * Parse component information.
+     */
+    private void parseComponents() {
+      for (HostGroupComponentEntity componentEntity : hostGroup.getComponents() ) {
+        String name = componentEntity.getName();
+        components.add(name);
+        String service = stack.getServiceForComponent(name);
+        Set<String> serviceComponents = componentsForService.get(service);
+        if (serviceComponents == null) {
+          serviceComponents = new HashSet<String>();
+          componentsForService.put(service, serviceComponents);
+        }
+        serviceComponents.add(name);
+      }
+    }
+  }
+
+  /**
+   * Provides functionality to update a property value.
+   */
+  public interface PropertyUpdater {
+    /**
+     * Update a property value.
+     *
+     * @param hostGroups  host groups
+     * @param origValue   original value of property
+     *
+     * @return new property value
+     */
+    public String update(Map<String, HostGroup> hostGroups, String origValue);
+  }
+
+  /**
+   * Topology based updater which replaces the original host name of a property with the host name
+   * which runs the associated (master) component in the new cluster.
+   */
+  private class SingleHostPropertyUpdater implements PropertyUpdater {
+    /**
+     * Component name
+     */
+    private String component;
+
+    /**
+     * Constructor.
+     *
+     * @param component  component name associated with the property
+     */
+    public SingleHostPropertyUpdater(String component) {
+      this.component = component;
+    }
+
+    /**
+     * Update the property with the new host name which runs the associated component.
+     *
+     * @param hostGroups  host groups                 host groups
+     * @param origValue   original value of property  original property value
+     *
+     * @return updated property value with old host name replaced by new host name
+     */
+    public String update(Map<String, HostGroup> hostGroups, String origValue)  {
+      Collection<HostGroup> matchingGroups = getHostGroupsForComponent(component, hostGroups.values());
+      if (matchingGroups.size() == 1) {
+        return origValue.replace("localhost", matchingGroups.iterator().next().getHostInfo().iterator().next());
+      } else {
+        throw new IllegalArgumentException("Unable to update configuration properties with topology information. " +
+            "Component '" + this.component + "' is not mapped to any host group or is mapped to multiple groups.");
+      }
+    }
+  }
+
+  /**
+   * Topology based updater which replaces original host names (possibly more than one) contained in a property
+   * value with the host names which runs the associated component in the new cluster.
+   */
+  private class MultipleHostPropertyUpdater implements PropertyUpdater {
+    /**
+     * Component name
+     */
+    private String component;
+
+    /**
+     * Constructor.
+     *
+     * @param component  component name associated with the property
+     */
+    public MultipleHostPropertyUpdater(String component) {
+      this.component = component;
+    }
+
+    //todo: specific to default values of EXACTLY 'localhost' or 'localhost:port'.
+    //todo: when blueprint contains source configurations, these props will contain actual host names, not localhost.
+    //todo: currently assuming that all hosts will share the same port
+    /**
+     * Update all host names included in the original property value with new host names which run the associated
+     * component.
+     *
+     * @param hostGroups  host groups                 host groups
+     * @param origValue   original value of property  original value
+     *
+     * @return updated property value with old host names replaced by new host names
+     */
+    public String update(Map<String, HostGroup> hostGroups, String origValue) {
+      String newValue;
+      Collection<HostGroup> matchingGroups = getHostGroupsForComponent(component, hostGroups.values());
+      boolean containsPort = origValue.contains(":");
+
+      if (containsPort) {
+        String port = origValue.substring(origValue.indexOf(":") + 1);
+        StringBuilder sb = new StringBuilder();
+        boolean firstHost = true;
+        for (HostGroup group : matchingGroups) {
+          for (String host : group.getHostInfo()) {
+            if (! firstHost) {
+              sb.append(',');
+            } else {
+              firstHost = false;
+            }
+            sb.append(host);
+            sb.append(":");
+            sb.append(port);
+          }
+        }
+        newValue = sb.toString();
+      } else {
+        newValue = matchingGroups.iterator().next().getHostInfo().iterator().next();
+      }
+      return newValue;
+    }
+  }
+
+  /**
+   * Updater which appends "m" to the original property value.
+   * For example, "1024" would be updated to "1024m".
+   */
+  private class MPropertyUpdater implements PropertyUpdater {
+    /**
+     * Append 'm' to the original property value if it doesn't already exist.
+     *
+     * @param hostGroups  host groups                 host groups
+     * @param origValue   original value of property  original property value
+     *
+     * @return property with 'm' appended
+     */
+    public String update(Map<String, HostGroup> hostGroups, String origValue) {
+      return origValue.endsWith("m") ? origValue : origValue + 'm';
+    }
   }
 }
+

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

@@ -802,7 +802,7 @@ public class ComponentResourceProvider extends AbstractControllerResourceProvide
     // modified?
     Cluster cluster = clusters.getCluster(clusterNames.iterator().next());
 
-    return getManagementController().createStages(cluster, requestProperties, null, null, changedComps, changedScHosts,
+    return getManagementController().createAndPersistStages(cluster, requestProperties, null, null, changedComps, changedScHosts,
         ignoredScHosts, runSmokeTest, false);
   }
 

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

@@ -0,0 +1,199 @@
+/**
+ * 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.controller.internal;
+
+import org.apache.ambari.server.RoleCommand;
+import org.apache.ambari.server.actionmanager.ActionManager;
+import org.apache.ambari.server.actionmanager.HostRoleCommand;
+import org.apache.ambari.server.actionmanager.Request;
+import org.apache.ambari.server.actionmanager.RequestFactory;
+import org.apache.ambari.server.actionmanager.Stage;
+import org.apache.ambari.server.controller.RequestStatusResponse;
+import org.apache.ambari.server.controller.ShortTaskStatus;
+import org.apache.ambari.server.state.State;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+
+/**
+ * Contains stages associated with a request.
+ */
+public class RequestStageContainer {
+  /**
+   * Request id
+   */
+  private Long id;
+
+  /**
+   * Request stages
+   */
+  private List<Stage> stages;
+
+  /**
+   * Request Factory used to create Request instance
+   */
+  private RequestFactory requestFactory;
+
+  /**
+   * Action Manager
+   */
+  private ActionManager actionManager;
+
+  /**
+   * Logger
+   */
+  private final static Logger LOG =
+      LoggerFactory.getLogger(RequestStageContainer.class);
+
+  /**
+   * Constructor.
+   *
+   * @param id       request id
+   * @param stages   stages
+   * @param factory  request factory
+   * @param manager  action manager
+   */
+  public RequestStageContainer(Long id, List<Stage> stages, RequestFactory factory, ActionManager manager) {
+    this.id = id;
+    this.stages = stages == null ? new ArrayList<Stage>() : stages;
+    this.requestFactory = factory;
+    this.actionManager = manager;
+  }
+
+  /**
+   * Get the request id.
+   *
+   * @return request id
+   */
+  public Long getId()  {
+    return id;
+  }
+
+  /**
+   * Add stages to request.
+   *
+   * @param stages stages to add
+   */
+  public void addStages(List<Stage> stages) {
+    if (stages != null) {
+      this.stages.addAll(stages);
+    }
+  }
+
+  /**
+   * Get request stages.
+   *
+   * @return  list of stages
+   */
+  public List<Stage> getStages() {
+    return stages;
+  }
+
+  /**
+   * Get the stage id of the last stage.
+   *
+   * @return stage id of the last stage or -1 if no stages present
+   */
+  public long getLastStageId() {
+    return stages.isEmpty() ? -1 : stages.get(stages.size() - 1).getStageId();
+  }
+
+  /**
+   * Determine the projected state for a host component from the existing stages.
+   *
+   * @param host       host name
+   * @param component  component name
+   *
+   * @return the projected state of a host component after all stages successfully complete
+   *         or null if the host component state is not modified in the current stages
+   */
+  public State getProjectedState(String host, String component) {
+    RoleCommand lastCommand = null;
+
+    ListIterator<Stage> iterator = stages.listIterator(stages.size());
+    while (lastCommand == null && iterator.hasPrevious()) {
+      Stage stage = iterator.previous();
+      HostRoleCommand hostRoleCommand = stage.getHostRoleCommand(host, component);
+      if (hostRoleCommand != null && hostRoleCommand.getRoleCommand() != RoleCommand.SERVICE_CHECK) {
+        lastCommand = hostRoleCommand.getRoleCommand();
+      }
+    }
+
+    State resultingState = null;
+    if (lastCommand != null) {
+      switch(lastCommand) {
+        case INSTALL:
+        case STOP:
+          resultingState = State.INSTALLED;
+          break;
+        case START:
+          resultingState = State.STARTED;
+          break;
+        case UNINSTALL:
+          resultingState = State.INIT;
+          break;
+        default:
+          resultingState = State.UNKNOWN;
+      }
+    }
+    return resultingState;
+  }
+
+  /**
+   * Persist the stages.
+   */
+  public void persist() {
+    if (!stages.isEmpty()) {
+      Request request = requestFactory.createNewFromStages(stages);
+      if (request != null && request.getStages()!= null && !request.getStages().isEmpty()) {
+        if (LOG.isDebugEnabled()) {
+          LOG.debug(String.format("Triggering Action Manager, request=%s", request));
+        }
+        actionManager.sendActions(request, null);
+      }
+    }
+  }
+
+  /**
+   * Build a request status response.
+   *
+   * @return a {@link org.apache.ambari.server.controller.RequestStatusResponse} for the request
+   */
+  public RequestStatusResponse getRequestStatusResponse() {
+    RequestStatusResponse response = null;
+
+    if (! stages.isEmpty()) {
+      response = new RequestStatusResponse(id);
+      List<HostRoleCommand> hostRoleCommands =
+          actionManager.getRequestTasks(id);
+
+      response.setRequestContext(actionManager.getRequestContext(id));
+      List<ShortTaskStatus> tasks = new ArrayList<ShortTaskStatus>();
+
+      for (HostRoleCommand hostRoleCommand : hostRoleCommands) {
+        tasks.add(new ShortTaskStatus(hostRoleCommand));
+      }
+      response.setTasks(tasks);
+    }
+    return response;
+  }
+}

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

@@ -43,6 +43,8 @@ import org.apache.ambari.server.controller.ServiceComponentHostRequest;
 import org.apache.ambari.server.controller.ServiceComponentHostResponse;
 import org.apache.ambari.server.controller.ServiceRequest;
 import org.apache.ambari.server.controller.ServiceResponse;
+import org.apache.ambari.server.controller.predicate.AndPredicate;
+import org.apache.ambari.server.controller.predicate.EqualsPredicate;
 import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
 import org.apache.ambari.server.controller.spi.NoSuchResourceException;
 import org.apache.ambari.server.controller.spi.Predicate;
@@ -82,7 +84,7 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
   protected static final String SERVICE_CLUSTER_NAME_PROPERTY_ID    = PropertyHelper.getPropertyId("ServiceInfo", "cluster_name");
   protected static final String SERVICE_SERVICE_NAME_PROPERTY_ID    = PropertyHelper.getPropertyId("ServiceInfo", "service_name");
   protected static final String SERVICE_SERVICE_STATE_PROPERTY_ID   = PropertyHelper.getPropertyId("ServiceInfo", "state");
-  protected static final String SERVICE_MAINTENANCE_STATE_PROPERTY_ID   = PropertyHelper.getPropertyId("ServiceInfo", "maintenance_state");
+  protected static final String SERVICE_MAINTENANCE_STATE_PROPERTY_ID = PropertyHelper.getPropertyId("ServiceInfo", "maintenance_state");
 
   //Parameters from the predicate
   private static final String QUERY_PARAMETERS_RUN_SMOKE_TEST_ID =
@@ -188,28 +190,12 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
   public RequestStatus updateResources(final Request request, Predicate predicate)
       throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
 
-    final Set<ServiceRequest> requests = new HashSet<ServiceRequest>();
-    RequestStatusResponse     response = null;
-
-    Iterator<Map<String,Object>> iterator = request.getProperties().iterator();
-    if (iterator.hasNext()) {
-      for (Map<String, Object> propertyMap : getPropertyMaps(iterator.next(), predicate)) {
-        requests.add(getRequest(propertyMap));
-      }
+    RequestStageContainer requestStages = doUpdateResources(null, request, predicate);
 
-      final boolean runSmokeTest = "true".equals(getQueryParameterValue(
-          QUERY_PARAMETERS_RUN_SMOKE_TEST_ID, predicate));
-
-      final boolean reconfigureClients = !"false".equals(getQueryParameterValue(
-          QUERY_PARAMETERS_RECONFIGURE_CLIENT, predicate));
-
-      response = modifyResources(new Command<RequestStatusResponse>() {
-        @Override
-        public RequestStatusResponse invoke() throws AmbariException {
-          return updateServices(requests,
-              request.getRequestInfoProperties(), runSmokeTest, reconfigureClients);
-        }
-      });
+    RequestStatusResponse response = null;
+    if (requestStages != null) {
+      requestStages.persist();
+      response = requestStages.getRequestStatusResponse();
     }
     notifyUpdate(Resource.Type.Service, request, predicate);
 
@@ -264,7 +250,78 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
   }
 
 
+  // ----- ServiceResourceProvider -----------------------------------------
+
+  RequestStatusResponse installAndStart(String clusterName) throws  SystemException,
+      UnsupportedPropertyException, NoSuchParentResourceException {
+
+    final RequestStageContainer requestStages;
+    Map<String, Object> installProperties = new HashMap<String, Object>();
+    installProperties.put(SERVICE_SERVICE_STATE_PROPERTY_ID, "INSTALLED");
+    Map<String, String> requestInfo = new HashMap<String, String>();
+    requestInfo.put("context", "Install and start all services");
+    Request installRequest = new RequestImpl(null, Collections.singleton(installProperties), requestInfo, null);
+    Predicate statePredicate = new EqualsPredicate<String>(SERVICE_SERVICE_STATE_PROPERTY_ID, "INIT");
+    Predicate clusterPredicate = new EqualsPredicate<String>(SERVICE_CLUSTER_NAME_PROPERTY_ID, clusterName);
+    Predicate installPredicate = new AndPredicate(statePredicate, clusterPredicate);
+
+    final Request startRequest;
+    Predicate startPredicate;
+    try {
+      LOG.info("Installing all services");
+      requestStages = doUpdateResources(null, installRequest, installPredicate);
+      notifyUpdate(Resource.Type.Service, installRequest, installPredicate);
+
+      Map<String, Object> startProperties = new HashMap<String, Object>();
+      startProperties.put(SERVICE_SERVICE_STATE_PROPERTY_ID, "STARTED");
+      startRequest = new RequestImpl(null, Collections.singleton(startProperties), requestInfo, null);
+      Predicate installedStatePredicate = new EqualsPredicate<String>(SERVICE_SERVICE_STATE_PROPERTY_ID, "INSTALLED");
+      Predicate serviceClusterPredicate = new EqualsPredicate<String>(SERVICE_CLUSTER_NAME_PROPERTY_ID, clusterName);
+      startPredicate = new AndPredicate(installedStatePredicate, serviceClusterPredicate);
+
+      LOG.info("Starting all services");
+      doUpdateResources(requestStages, startRequest, startPredicate);
+      notifyUpdate(Resource.Type.Service, startRequest, startPredicate);
+      requestStages.persist();
+      return requestStages.getRequestStatusResponse();
+
+    } catch (NoSuchResourceException e) {
+      throw new SystemException("Attempted to modify a non-existing service",  e);
+    }
+  }
+
+
   // ----- utility methods -------------------------------------------------
+
+  private RequestStageContainer doUpdateResources(final RequestStageContainer stages, final Request request, Predicate predicate)
+      throws UnsupportedPropertyException, SystemException, NoSuchResourceException, NoSuchParentResourceException {
+
+    final Set<ServiceRequest> requests = new HashSet<ServiceRequest>();
+    RequestStageContainer requestStages = null;
+
+    Iterator<Map<String,Object>> iterator = request.getProperties().iterator();
+    if (iterator.hasNext()) {
+      for (Map<String, Object> propertyMap : getPropertyMaps(iterator.next(), predicate)) {
+        requests.add(getRequest(propertyMap));
+      }
+
+      final boolean runSmokeTest = "true".equals(getQueryParameterValue(
+          QUERY_PARAMETERS_RUN_SMOKE_TEST_ID, predicate));
+
+      final boolean reconfigureClients = !"false".equals(getQueryParameterValue(
+          QUERY_PARAMETERS_RECONFIGURE_CLIENT, predicate));
+
+      requestStages = modifyResources(new Command<RequestStageContainer>() {
+        @Override
+        public RequestStageContainer invoke() throws AmbariException {
+          return updateServices(stages, requests, request.getRequestInfoProperties(),
+              runSmokeTest, reconfigureClients);
+        }
+      });
+    }
+    return requestStages;
+  }
+
   /**
    * Get a service request object from a map of property values.
    *
@@ -482,9 +539,9 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
   }
 
   // Update services based on the given requests.
-  protected synchronized RequestStatusResponse updateServices(
-      Set<ServiceRequest> requests, Map<String, String> requestProperties,
-      boolean runSmokeTest, boolean reconfigureClients) throws AmbariException {
+  protected synchronized RequestStageContainer updateServices(RequestStageContainer requestStages, Set<ServiceRequest> requests,
+                                                      Map<String, String> requestProperties, boolean runSmokeTest,
+                                                      boolean reconfigureClients) throws AmbariException {
 
     AmbariManagementController controller = getManagementController();
 
@@ -708,7 +765,7 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
            * This is hack for now wherein we don't fail if the
            * sch is in INSTALL_FAILED
            */
-          if (!State.isValidStateTransition(oldSchState, newState)) {
+          if (! isValidStateTransition(requestStages, oldSchState, newState, sch)) {
             String error = "Invalid transition for"
                 + " servicecomponenthost"
                 + ", clusterName=" + cluster.getClusterName()
@@ -768,7 +825,7 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
 
     Cluster cluster = clusters.getCluster(clusterNames.iterator().next());
 
-    return controller.createStages(cluster, requestProperties, null, changedServices, changedComps, changedScHosts,
+    return controller.addStages(requestStages, cluster, requestProperties, null, changedServices, changedComps, changedScHosts,
         ignoredScHosts, runSmokeTest, reconfigureClients);
   }
 
@@ -1068,5 +1125,27 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
       return State.UNKNOWN;
     }
   }
-  
+
+  /**
+   * Determine whether a service state change is valid.
+   * Looks at projected state from the current stages associated with the request.
+   *
+   *
+   * @param stages        request stages
+   * @param startState    service start state
+   * @param desiredState  service desired state
+   * @param host          host where state change is occurring
+   *
+   * @return whether the state transition is valid
+   */
+  private boolean isValidStateTransition(RequestStageContainer stages, State startState,
+                                         State desiredState, ServiceComponentHost host) {
+
+    if (stages != null) {
+      State projectedState = stages.getProjectedState(host.getHostName(), host.getServiceComponentName());
+      startState = projectedState == null ? startState : projectedState;
+    }
+
+    return State.isValidStateTransition(startState, desiredState);
+  }
 }

ambari-server/src/main/java/org/apache/ambari/server/controller/spi/SystemException.java

@@ -22,6 +22,15 @@ package org.apache.ambari.server.controller.spi;
  * Indicates that a system exception occurred.
  */
 public class SystemException extends Exception {
+  /**
+   * Constructor.
+   *
+   * @param msg  message
+   */
+  public SystemException(String msg) {
+    super(msg);
+  }
+
   /**
    * Constructor.
    *

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

@@ -973,7 +973,6 @@ public class AmbariManagementControllerTest {
       fail("ServiceComponentHost creation should fail for invalid state");
     } catch (Exception e) {
       // Expected
-      e.printStackTrace();
     }
 
     try {

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ClusterResourceProviderTest.java

@@ -18,25 +18,46 @@
 
 package org.apache.ambari.server.controller.internal;
 
+import static org.easymock.EasyMock.capture;
 import static org.easymock.EasyMock.createMock;
 import static org.easymock.EasyMock.createNiceMock;
+import static org.easymock.EasyMock.createStrictMock;
 import static org.easymock.EasyMock.eq;
 import static org.easymock.EasyMock.expect;
 import static org.easymock.EasyMock.replay;
 import static org.easymock.EasyMock.verify;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 import java.util.*;
 
+import org.apache.ambari.server.api.services.PersistKeyValueImpl;
+import org.apache.ambari.server.api.services.PersistKeyValueService;
 import org.apache.ambari.server.controller.AmbariManagementController;
 import org.apache.ambari.server.controller.ClusterRequest;
 import org.apache.ambari.server.controller.ClusterResponse;
+import org.apache.ambari.server.controller.ConfigurationRequest;
 import org.apache.ambari.server.controller.RequestStatusResponse;
+import org.apache.ambari.server.controller.StackConfigurationRequest;
+import org.apache.ambari.server.controller.StackConfigurationResponse;
+import org.apache.ambari.server.controller.StackServiceComponentRequest;
+import org.apache.ambari.server.controller.StackServiceComponentResponse;
+import org.apache.ambari.server.controller.StackServiceRequest;
+import org.apache.ambari.server.controller.StackServiceResponse;
 import org.apache.ambari.server.controller.spi.Predicate;
 import org.apache.ambari.server.controller.spi.Request;
+import org.apache.ambari.server.controller.spi.RequestStatus;
 import org.apache.ambari.server.controller.spi.Resource;
 import org.apache.ambari.server.controller.spi.ResourceProvider;
 import org.apache.ambari.server.controller.utilities.PredicateBuilder;
 import org.apache.ambari.server.controller.utilities.PropertyHelper;
+import org.apache.ambari.server.orm.dao.BlueprintDAO;
+import org.apache.ambari.server.orm.entities.BlueprintEntity;
+import org.apache.ambari.server.orm.entities.HostGroupComponentEntity;
+import org.apache.ambari.server.orm.entities.HostGroupEntity;
+import org.easymock.Capture;
 import org.easymock.EasyMock;
 import org.junit.Assert;
 import org.junit.Test;
@@ -111,6 +132,305 @@ public class ClusterResourceProviderTest {
     verify(managementController, response);
   }
 
+  @Test
+  public void testCreateResource_blueprint() throws Exception {
+    String blueprintName = "test-blueprint";
+    String stackName = "test";
+    String stackVersion = "1.23";
+
+    BlueprintDAO blueprintDAO = createStrictMock(BlueprintDAO.class);
+    AmbariManagementController managementController = createStrictMock(AmbariManagementController.class);
+    Request request = createNiceMock(Request.class);
+    RequestStatusResponse response = createNiceMock(RequestStatusResponse.class);
+    BlueprintEntity blueprint = createNiceMock(BlueprintEntity.class);
+    StackServiceResponse stackServiceResponse1 = createNiceMock(StackServiceResponse.class);
+    StackServiceResponse stackServiceResponse2 = createNiceMock(StackServiceResponse.class);
+    Capture<Set<StackServiceRequest>> stackServiceRequestCapture = new Capture<Set<StackServiceRequest>>();
+
+    StackServiceComponentResponse stackServiceComponentResponse1 = createNiceMock(StackServiceComponentResponse.class);
+    StackServiceComponentResponse stackServiceComponentResponse2 = createNiceMock(StackServiceComponentResponse.class);
+    StackServiceComponentResponse stackServiceComponentResponse3 = createNiceMock(StackServiceComponentResponse.class);
+    Capture<Set<StackServiceComponentRequest>> serviceComponentRequestCapture1 = new Capture<Set<StackServiceComponentRequest>>();
+    Capture<Set<StackServiceComponentRequest>> serviceComponentRequestCapture2 = new Capture<Set<StackServiceComponentRequest>>();
+
+    StackConfigurationResponse stackConfigurationResponse1 = createNiceMock(StackConfigurationResponse.class);
+    StackConfigurationResponse stackConfigurationResponse2 = createNiceMock(StackConfigurationResponse.class);
+    StackConfigurationResponse stackConfigurationResponse3 = createNiceMock(StackConfigurationResponse.class);
+    Capture<Set<StackConfigurationRequest>> serviceConfigurationRequestCapture1 = new Capture<Set<StackConfigurationRequest>>();
+    Capture<Set<StackConfigurationRequest>> serviceConfigurationRequestCapture2 = new Capture<Set<StackConfigurationRequest>>();
+
+    HostGroupEntity hostGroup = createNiceMock(HostGroupEntity.class);
+    HostGroupComponentEntity hostGroupComponent1 = createNiceMock(HostGroupComponentEntity.class);
+    HostGroupComponentEntity hostGroupComponent2 = createNiceMock(HostGroupComponentEntity.class);
+    HostGroupComponentEntity hostGroupComponent3 = createNiceMock(HostGroupComponentEntity.class);
+
+    ServiceResourceProvider serviceResourceProvider = createStrictMock(ServiceResourceProvider.class);
+    ResourceProvider componentResourceProvider = createStrictMock(ResourceProvider.class);
+    ResourceProvider hostResourceProvider = createStrictMock(ResourceProvider.class);
+    ResourceProvider hostComponentResourceProvider = createStrictMock(ResourceProvider.class);
+    PersistKeyValueImpl persistKeyValue = createNiceMock(PersistKeyValueImpl.class);
+
+    Capture<ClusterRequest> createClusterRequestCapture = new Capture<ClusterRequest>();
+    Capture<Set<ClusterRequest>> updateClusterRequestCapture = new Capture<Set<ClusterRequest>>();
+    Capture<Map<String, String>> updateClusterPropertyMapCapture = new Capture<Map<String, String>>();
+    Capture<Set<ClusterRequest>> updateClusterRequestCapture2 = new Capture<Set<ClusterRequest>>();
+    Capture<Map<String, String>> updateClusterPropertyMapCapture2 = new Capture<Map<String, String>>();
+    Capture<Set<ClusterRequest>> updateClusterRequestCapture3 = new Capture<Set<ClusterRequest>>();
+    Capture<Map<String, String>> updateClusterPropertyMapCapture3 = new Capture<Map<String, String>>();
+
+
+    Capture<Request> serviceRequestCapture = new Capture<Request>();
+    Capture<Request> componentRequestCapture = new Capture<Request>();
+    Capture<Request> componentRequestCapture2 = new Capture<Request>();
+    Capture<Request> hostRequestCapture = new Capture<Request>();
+    Capture<Request> hostComponentRequestCapture = new Capture<Request>();
+
+    Set<StackServiceResponse> stackServiceResponses = new LinkedHashSet<StackServiceResponse>();
+    stackServiceResponses.add(stackServiceResponse1);
+    stackServiceResponses.add(stackServiceResponse2);
+
+    // service1 has 2 components
+    Set<StackServiceComponentResponse> stackServiceComponentResponses1 = new LinkedHashSet<StackServiceComponentResponse>();
+    stackServiceComponentResponses1.add(stackServiceComponentResponse1);
+    stackServiceComponentResponses1.add(stackServiceComponentResponse2);
+
+    // service2 has 1 components
+    Set<StackServiceComponentResponse> stackServiceComponentResponses2 = new LinkedHashSet<StackServiceComponentResponse>();
+    stackServiceComponentResponses2.add(stackServiceComponentResponse3);
+
+    // service1 has 1 config
+    Set<StackConfigurationResponse> stackConfigurationResponses1 = new LinkedHashSet<StackConfigurationResponse>();
+    stackConfigurationResponses1.add(stackConfigurationResponse1);
+
+    // service2 has 2 config
+    Set<StackConfigurationResponse> stackConfigurationResponses2 = new LinkedHashSet<StackConfigurationResponse>();
+    stackConfigurationResponses2.add(stackConfigurationResponse2);
+    stackConfigurationResponses2.add(stackConfigurationResponse3);
+
+    Collection<HostGroupComponentEntity> hostGroupComponents = new LinkedHashSet<HostGroupComponentEntity>();
+    hostGroupComponents.add(hostGroupComponent1);
+    hostGroupComponents.add(hostGroupComponent2);
+    hostGroupComponents.add(hostGroupComponent3);
+
+    // request properties
+    Set<Map<String, Object>> propertySet = new LinkedHashSet<Map<String, Object>>();
+    Map<String, Object> properties = new LinkedHashMap<String, Object>();
+
+    properties.put(ClusterResourceProvider.CLUSTER_NAME_PROPERTY_ID, "c1");
+    properties.put(ClusterResourceProvider.BLUEPRINT_PROPERTY_ID, blueprintName);
+    propertySet.add(properties);
+
+    Collection<Map<String, Object>> hostGroups = new ArrayList<Map<String, Object>>();
+    Map<String, Object> hostGroupProperties = new HashMap<String, Object>();
+    hostGroups.add(hostGroupProperties);
+    hostGroupProperties.put("name", "group1");
+    Collection<Map<String, String>> hostGroupHosts = new ArrayList<Map<String, String>>();
+    hostGroupProperties.put("hosts", hostGroupHosts);
+    Map<String, String> hostGroupHostProperties = new HashMap<String, String>();
+    hostGroupHostProperties.put("fqdn", "host.domain");
+    hostGroupHosts.add(hostGroupHostProperties);
+    properties.put("host-groups", hostGroups);
+
+    // expectations
+    expect(request.getProperties()).andReturn(propertySet).anyTimes();
+    expect(blueprintDAO.findByName(blueprintName)).andReturn(blueprint);
+    expect(blueprint.getStackName()).andReturn(stackName);
+    expect(blueprint.getStackVersion()).andReturn(stackVersion);
+
+    expect(managementController.getStackServices(capture(stackServiceRequestCapture))).andReturn(stackServiceResponses);
+    expect(stackServiceResponse1.getServiceName()).andReturn("service1");
+    expect(stackServiceResponse2.getServiceName()).andReturn("service2");
+
+    expect(managementController.getStackComponents(capture(serviceComponentRequestCapture1))).
+        andReturn(stackServiceComponentResponses1);
+    expect(stackServiceComponentResponse1.getComponentName()).andReturn("component1");
+    expect(stackServiceComponentResponse2.getComponentName()).andReturn("component2");
+
+    expect(managementController.getStackConfigurations(capture(serviceConfigurationRequestCapture1))).
+        andReturn(stackConfigurationResponses1);
+    expect(stackConfigurationResponse1.getType()).andReturn("core-site.xml");
+    expect(stackConfigurationResponse1.getPropertyName()).andReturn("property1");
+    expect(stackConfigurationResponse1.getPropertyValue()).andReturn("value1");
+
+    expect(managementController.getStackComponents(capture(serviceComponentRequestCapture2))).
+        andReturn(stackServiceComponentResponses2);
+    expect(stackServiceComponentResponse3.getComponentName()).andReturn("component3");
+
+    expect(managementController.getStackConfigurations(capture(serviceConfigurationRequestCapture2))).
+        andReturn(stackConfigurationResponses2);
+    expect(stackConfigurationResponse2.getType()).andReturn("hdfs-site.xml");
+    expect(stackConfigurationResponse2.getPropertyName()).andReturn("property2");
+    expect(stackConfigurationResponse2.getPropertyValue()).andReturn("value2");
+    expect(stackConfigurationResponse3.getType()).andReturn("core-site.xml");
+    expect(stackConfigurationResponse3.getPropertyName()).andReturn("property3");
+    expect(stackConfigurationResponse3.getPropertyValue()).andReturn("value3");
+
+    expect(blueprint.getHostGroups()).andReturn(Collections.singleton(hostGroup));
+    expect(hostGroup.getName()).andReturn("group1");
+    expect(hostGroup.getComponents()).andReturn(hostGroupComponents);
+    expect(hostGroupComponent1.getName()).andReturn("component1");
+    expect(hostGroupComponent2.getName()).andReturn("component2");
+    expect(hostGroupComponent3.getName()).andReturn("component3");
+
+    managementController.createCluster(capture(createClusterRequestCapture));
+    expect(managementController.updateClusters(capture(updateClusterRequestCapture),
+        capture(updateClusterPropertyMapCapture))).andReturn(null);
+    expect(managementController.updateClusters(capture(updateClusterRequestCapture2),
+        capture(updateClusterPropertyMapCapture2))).andReturn(null);
+    expect(managementController.updateClusters(capture(updateClusterRequestCapture3),
+        capture(updateClusterPropertyMapCapture3))).andReturn(null);
+
+    expect(serviceResourceProvider.createResources(capture(serviceRequestCapture))).andReturn(null);
+    expect(componentResourceProvider.createResources(capture(componentRequestCapture))).andReturn(null);
+    expect(componentResourceProvider.createResources(capture(componentRequestCapture2))).andReturn(null);
+    expect(hostResourceProvider.createResources(capture(hostRequestCapture))).andReturn(null);
+    expect(hostComponentResourceProvider.createResources(capture(hostComponentRequestCapture))).andReturn(null);
+
+    expect(serviceResourceProvider.installAndStart("c1")).andReturn(response);
+
+    persistKeyValue.put("CLUSTER_CURRENT_STATUS", "{\"clusterState\":\"CLUSTER_STARTED_5\"}");
+
+    replay(blueprintDAO, managementController, request, response, blueprint, stackServiceResponse1, stackServiceResponse2,
+           stackServiceComponentResponse1, stackServiceComponentResponse2, stackServiceComponentResponse3,
+           stackConfigurationResponse1, stackConfigurationResponse2, stackConfigurationResponse3, hostGroup,
+           hostGroupComponent1, hostGroupComponent2, hostGroupComponent3, serviceResourceProvider,
+           componentResourceProvider, hostResourceProvider, hostComponentResourceProvider, persistKeyValue);
+
+    // test
+    ClusterResourceProvider.injectBlueprintDAO(blueprintDAO);
+    PersistKeyValueService.init(persistKeyValue);
+    ResourceProvider provider = new TestClusterResourceProvider(
+        managementController, serviceResourceProvider, componentResourceProvider,
+        hostResourceProvider, hostComponentResourceProvider);
+
+    RequestStatus requestStatus = provider.createResources(request);
+
+    assertEquals(RequestStatus.Status.InProgress, requestStatus.getStatus());
+
+    Set<StackServiceRequest> stackServiceRequests = stackServiceRequestCapture.getValue();
+    assertEquals(1, stackServiceRequests.size());
+    StackServiceRequest ssr = stackServiceRequests.iterator().next();
+    assertNull(ssr.getServiceName());
+    assertEquals("test", ssr.getStackName());
+    assertEquals("1.23", ssr.getStackVersion());
+
+    Set<StackServiceComponentRequest> stackServiceComponentRequests1 = serviceComponentRequestCapture1.getValue();
+    Set<StackServiceComponentRequest> stackServiceComponentRequests2 = serviceComponentRequestCapture2.getValue();
+    assertEquals(1, stackServiceComponentRequests1.size());
+    assertEquals(1, stackServiceComponentRequests2.size());
+    StackServiceComponentRequest scr1 = stackServiceComponentRequests1.iterator().next();
+    StackServiceComponentRequest scr2 = stackServiceComponentRequests2.iterator().next();
+    assertNull(scr1.getComponentName());
+    assertNull(scr2.getComponentName());
+    assertEquals("1.23", scr1.getStackVersion());
+    assertEquals("1.23", scr2.getStackVersion());
+    assertEquals("test", scr1.getStackName());
+    assertEquals("test", scr2.getStackName());
+    assertTrue(scr1.getServiceName().equals("service1") || scr1.getServiceName().equals("service2"));
+    assertTrue(scr2.getServiceName().equals("service1") || scr2.getServiceName().equals("service2") &&
+        ! scr2.getServiceName().equals(scr1.getServiceName()));
+
+    Set<StackConfigurationRequest> serviceConfigurationRequest1 = serviceConfigurationRequestCapture1.getValue();
+    Set<StackConfigurationRequest> serviceConfigurationRequest2 = serviceConfigurationRequestCapture2.getValue();
+    assertEquals(1, serviceConfigurationRequest1.size());
+    assertEquals(1, serviceConfigurationRequest2.size());
+    StackConfigurationRequest configReq1 = serviceConfigurationRequest1.iterator().next();
+    StackConfigurationRequest configReq2 = serviceConfigurationRequest2.iterator().next();
+    assertNull(configReq1.getPropertyName());
+    assertNull(configReq2.getPropertyName());
+    assertEquals("1.23", configReq1.getStackVersion());
+    assertEquals("1.23", configReq2.getStackVersion());
+    assertEquals("test", configReq1.getStackName());
+    assertEquals("test", configReq2.getStackName());
+    assertTrue(configReq1.getServiceName().equals("service1") || configReq1.getServiceName().equals("service2"));
+    assertTrue(configReq2.getServiceName().equals("service1") || configReq2.getServiceName().equals("service2") &&
+        ! configReq2.getServiceName().equals(configReq1.getServiceName()));
+
+    ClusterRequest clusterRequest = createClusterRequestCapture.getValue();
+    assertEquals("c1", clusterRequest.getClusterName());
+    assertEquals("test-1.23", clusterRequest.getStackVersion());
+
+    Set<ClusterRequest> updateClusterRequest1 = updateClusterRequestCapture.getValue();
+    Set<ClusterRequest> updateClusterRequest2 = updateClusterRequestCapture2.getValue();
+    Set<ClusterRequest> updateClusterRequest3 = updateClusterRequestCapture3.getValue();
+    assertEquals(1, updateClusterRequest1.size());
+    assertEquals(1, updateClusterRequest2.size());
+    assertEquals(1, updateClusterRequest3.size());
+    ClusterRequest ucr1 = updateClusterRequest1.iterator().next();
+    ClusterRequest ucr2 = updateClusterRequest2.iterator().next();
+    ClusterRequest ucr3 = updateClusterRequest3.iterator().next();
+    assertEquals("c1", ucr1.getClusterName());
+    assertEquals("c1", ucr2.getClusterName());
+    assertEquals("c1", ucr3.getClusterName());
+    ConfigurationRequest cr1 = ucr1.getDesiredConfig();
+    ConfigurationRequest cr2 = ucr2.getDesiredConfig();
+    ConfigurationRequest cr3 = ucr3.getDesiredConfig();
+    assertEquals("1", cr1.getVersionTag());
+    assertEquals("1", cr2.getVersionTag());
+    assertEquals("1", cr3.getVersionTag());
+    Map<String, ConfigurationRequest> mapConfigRequests = new HashMap<String, ConfigurationRequest>();
+    mapConfigRequests.put(cr1.getType(), cr1);
+    mapConfigRequests.put(cr2.getType(), cr2);
+    mapConfigRequests.put(cr3.getType(), cr3);
+    assertEquals(3, mapConfigRequests.size());
+    ConfigurationRequest globalConfigRequest = mapConfigRequests.get("global");
+    assertEquals(4, globalConfigRequest.getProperties().size());
+    assertEquals("hadoop", globalConfigRequest.getProperties().get("user_group"));
+    assertEquals("ambari-qa", globalConfigRequest.getProperties().get("smokeuser"));
+    assertEquals("default@REPLACEME.NOWHERE", globalConfigRequest.getProperties().get("nagios_contact"));
+    assertEquals("admin", globalConfigRequest.getProperties().get("nagios_web_password"));
+    ConfigurationRequest hdfsConfigRequest = mapConfigRequests.get("hdfs-site");
+    assertEquals(1, hdfsConfigRequest.getProperties().size());
+    assertEquals("value2", hdfsConfigRequest.getProperties().get("property2"));
+    ConfigurationRequest coreConfigRequest = mapConfigRequests.get("core-site");
+    assertEquals(2, coreConfigRequest.getProperties().size());
+    assertEquals("value1", coreConfigRequest.getProperties().get("property1"));
+    assertEquals("value3", coreConfigRequest.getProperties().get("property3"));
+    assertNull(updateClusterPropertyMapCapture.getValue());
+    assertNull(updateClusterPropertyMapCapture2.getValue());
+    assertNull(updateClusterPropertyMapCapture3.getValue());
+
+    Request serviceRequest = serviceRequestCapture.getValue();
+    assertEquals(2, serviceRequest.getProperties().size());
+    Request componentRequest = componentRequestCapture.getValue();
+    Request componentRequest2 = componentRequestCapture2.getValue();
+    assertEquals(2, componentRequest.getProperties().size());
+    Set<String> componentRequest1Names = new HashSet<String>();
+    for (Map<String, Object> componentRequest1Properties : componentRequest.getProperties()) {
+      assertEquals(3, componentRequest1Properties.size());
+      assertEquals("c1", componentRequest1Properties.get("ServiceComponentInfo/cluster_name"));
+      assertEquals("service1", componentRequest1Properties.get("ServiceComponentInfo/service_name"));
+      componentRequest1Names.add((String) componentRequest1Properties.get("ServiceComponentInfo/component_name"));
+    }
+    assertTrue(componentRequest1Names.contains("component1") && componentRequest1Names.contains("component2"));
+    assertEquals(1, componentRequest2.getProperties().size());
+    Map<String, Object> componentRequest2Properties = componentRequest2.getProperties().iterator().next();
+    assertEquals("c1", componentRequest2Properties.get("ServiceComponentInfo/cluster_name"));
+    assertEquals("service2", componentRequest2Properties.get("ServiceComponentInfo/service_name"));
+    assertEquals("component3", componentRequest2Properties.get("ServiceComponentInfo/component_name"));
+    Request hostRequest = hostRequestCapture.getValue();
+    assertEquals(1, hostRequest.getProperties().size());
+    assertEquals("c1", hostRequest.getProperties().iterator().next().get("Hosts/cluster_name"));
+    assertEquals("host.domain", hostRequest.getProperties().iterator().next().get("Hosts/host_name"));
+    Request hostComponentRequest = hostComponentRequestCapture.getValue();
+    assertEquals(3, hostComponentRequest.getProperties().size());
+    Set<String> componentNames = new HashSet<String>();
+    for (Map<String, Object> hostComponentProperties : hostComponentRequest.getProperties()) {
+      assertEquals(3, hostComponentProperties.size());
+      assertEquals("c1", hostComponentProperties.get("HostRoles/cluster_name"));
+      assertEquals("host.domain", hostComponentProperties.get("HostRoles/host_name"));
+      componentNames.add((String) hostComponentProperties.get("HostRoles/component_name"));
+    }
+    assertTrue(componentNames.contains("component1") && componentNames.contains("component2") &&
+        componentNames.contains("component3"));
+
+    verify(blueprintDAO, managementController, request, response, blueprint, stackServiceResponse1, stackServiceResponse2,
+        stackServiceComponentResponse1, stackServiceComponentResponse2, stackServiceComponentResponse3,
+        stackConfigurationResponse1, stackConfigurationResponse2, stackConfigurationResponse3, hostGroup,
+        hostGroupComponent1, hostGroupComponent2, hostGroupComponent3, serviceResourceProvider,
+        componentResourceProvider, hostResourceProvider, hostComponentResourceProvider, persistKeyValue);
+  }
+
   @Test
   public void testGetResources() throws Exception{
     Resource.Type type = Resource.Type.Cluster;
@@ -251,7 +571,7 @@ public class ClusterResourceProviderTest {
     // verify
     verify(managementController, response);
   }
-  
+
   @Test
   public void testUpdateWithConfiguration() throws Exception {
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
@@ -368,4 +688,44 @@ public class ClusterResourceProviderTest {
     // verify
     verify(managementController, response);
   }
+
+  private class TestClusterResourceProvider extends ClusterResourceProvider {
+
+    private ResourceProvider serviceResourceProvider;
+    private ResourceProvider componentResourceProvider;
+    private ResourceProvider hostResourceProvider;
+    private ResourceProvider hostComponentResourceProvider;
+
+    TestClusterResourceProvider(AmbariManagementController managementController,
+                                ResourceProvider serviceResourceProvider,
+                                ResourceProvider componentResourceProvider,
+                                ResourceProvider hostResourceProvider,
+                                ResourceProvider hostComponentResourceProvider) {
+
+      super(PropertyHelper.getPropertyIds(Resource.Type.Cluster),
+            PropertyHelper.getKeyPropertyIds(Resource.Type.Cluster),
+            managementController);
+
+      this.serviceResourceProvider = serviceResourceProvider;
+      this.componentResourceProvider = componentResourceProvider;
+      this.hostResourceProvider = hostResourceProvider;
+      this.hostComponentResourceProvider = hostComponentResourceProvider;
+    }
+
+    @Override
+    ResourceProvider getResourceProviderByType(Resource.Type type) {
+      if (type == Resource.Type.Service) {
+        return this.serviceResourceProvider;
+      } else if (type == Resource.Type.Component) {
+        return this.componentResourceProvider;
+      } else if (type == Resource.Type.Host) {
+        return this.hostResourceProvider;
+      } else if (type == Resource.Type.HostComponent) {
+        return this.hostComponentResourceProvider;
+      } else {
+        fail("Unexpected resource provider type requested");
+      }
+      return null;
+    }
+  }
 }

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ComponentResourceProviderTest.java

@@ -290,7 +290,7 @@ public class ComponentResourceProviderTest {
     Capture<Collection<ServiceComponentHost>> ignoredScHostsCapture = new Capture<Collection<ServiceComponentHost>>();
     Capture<Cluster> clusterCapture = new Capture<Cluster>();
 
-    expect(managementController.createStages(capture(clusterCapture), capture(requestPropertiesCapture), capture(requestParametersCapture), capture(changedServicesCapture), capture(changedCompsCapture), capture(changedScHostsCapture), capture(ignoredScHostsCapture), anyBoolean(), anyBoolean()
+    expect(managementController.createAndPersistStages(capture(clusterCapture), capture(requestPropertiesCapture), capture(requestParametersCapture), capture(changedServicesCapture), capture(changedCompsCapture), capture(changedScHostsCapture), capture(ignoredScHostsCapture), anyBoolean(), anyBoolean()
     )).andReturn(requestStatusResponse);
 
 

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/RequestStageContainerTest.java

@@ -0,0 +1,209 @@
+/**
+ * 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.controller.internal;
+
+import org.apache.ambari.server.Role;
+import org.apache.ambari.server.RoleCommand;
+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.actionmanager.Request;
+import org.apache.ambari.server.actionmanager.RequestFactory;
+import org.apache.ambari.server.actionmanager.Stage;
+import org.apache.ambari.server.controller.RequestStatusResponse;
+import org.apache.ambari.server.controller.ShortTaskStatus;
+import org.apache.ambari.server.state.State;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.easymock.EasyMock.createNiceMock;
+import static org.easymock.EasyMock.createStrictMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * RequestStageContainer unit tests.
+ */
+public class RequestStageContainerTest {
+  @Test
+  public void testGetId() {
+    RequestStageContainer requestStages = new RequestStageContainer(500L, null, null, null);
+    assertEquals(500, requestStages.getId().longValue());
+  }
+
+  @Test
+  public void testGetAddStages() {
+    RequestStageContainer requestStages = new RequestStageContainer(500L, null, null, null);
+    assertTrue(requestStages.getStages().isEmpty());
+
+    Stage stage = createNiceMock(Stage.class);
+    requestStages.addStages(Collections.singletonList(stage));
+    assertEquals(1, requestStages.getStages().size());
+    assertTrue(requestStages.getStages().contains(stage));
+
+    Stage stage2 = createNiceMock(Stage.class);
+    Stage stage3 = createNiceMock(Stage.class);
+    List<Stage> listStages = new ArrayList<Stage>();
+    listStages.add(stage2);
+    listStages.add(stage3);
+    requestStages.addStages(listStages);
+    assertEquals(3, requestStages.getStages().size());
+    listStages = requestStages.getStages();
+    assertEquals(stage, listStages.get(0));
+    assertEquals(stage2, listStages.get(1));
+    assertEquals(stage3, listStages.get(2));
+  }
+
+  @Test
+  public void testGetLastStageId() {
+    RequestStageContainer requestStages = new RequestStageContainer(1L, null, null, null);
+    assertEquals(-1, requestStages.getLastStageId());
+
+    Stage stage1 = createNiceMock(Stage.class);
+    Stage stage2 = createNiceMock(Stage.class);
+    List<Stage> listStages = new ArrayList<Stage>();
+    listStages.add(stage1);
+    listStages.add(stage2);
+
+    expect(stage2.getStageId()).andReturn(22L);
+    replay(stage1, stage2);
+
+    requestStages = new RequestStageContainer(1L, listStages, null, null);
+    assertEquals(22, requestStages.getLastStageId());
+  }
+
+  @Test
+  public void testGetProjectedState() {
+    String hostname = "host";
+    String componentName = "component";
+
+    Stage stage1 = createNiceMock(Stage.class);
+    Stage stage2 = createNiceMock(Stage.class);
+    Stage stage3 = createNiceMock(Stage.class);
+    Stage stage4 = createNiceMock(Stage.class);
+    HostRoleCommand command1 = createNiceMock(HostRoleCommand.class);
+    HostRoleCommand command2 = createNiceMock(HostRoleCommand.class);
+    HostRoleCommand command3 = createNiceMock(HostRoleCommand.class);
+
+    List<Stage> stages = new ArrayList<Stage>();
+    stages.add(stage1);
+    stages.add(stage2);
+    stages.add(stage3);
+    stages.add(stage4);
+
+    //expectations
+    expect(stage1.getHostRoleCommand(hostname, componentName)).andReturn(command1).anyTimes();
+    expect(stage2.getHostRoleCommand(hostname, componentName)).andReturn(command2).anyTimes();
+    expect(stage3.getHostRoleCommand(hostname, componentName)).andReturn(command3).anyTimes();
+    expect(stage4.getHostRoleCommand(hostname, componentName)).andReturn(null).anyTimes();
+
+    expect(command3.getRoleCommand()).andReturn(RoleCommand.SERVICE_CHECK).anyTimes();
+    expect(command2.getRoleCommand()).andReturn(RoleCommand.INSTALL).anyTimes();
+    replay(stage1, stage2, stage3, stage4, command1, command2, command3);
+
+    RequestStageContainer requestStages = new RequestStageContainer(1L, stages, null, null);
+    assertEquals(State.INSTALLED, requestStages.getProjectedState(hostname, componentName));
+
+    verify(stage1, stage2, stage3, stage4, command1, command2, command3);
+  }
+
+  @Test
+  public void testPersist() {
+    ActionManager actionManager = createStrictMock(ActionManager.class);
+    RequestFactory requestFactory = createStrictMock(RequestFactory.class);
+    Request request = createStrictMock(Request.class);
+    Stage stage1 = createNiceMock(Stage.class);
+    Stage stage2 = createNiceMock(Stage.class);
+    List<Stage> stages = new ArrayList<Stage>();
+    stages.add(stage1);
+    stages.add(stage2);
+
+    //expectations
+    expect(requestFactory.createNewFromStages(stages)).andReturn(request);
+    expect(request.getStages()).andReturn(stages).anyTimes();
+    actionManager.sendActions(request, null);
+
+    replay(actionManager, requestFactory, request, stage1, stage2);
+
+    RequestStageContainer requestStages = new RequestStageContainer(1L, stages, requestFactory, actionManager);
+    requestStages.persist();
+
+    verify(actionManager, requestFactory, request, stage1, stage2);
+  }
+
+  @Test
+  public void testPersist_noStages() {
+    ActionManager actionManager = createStrictMock(ActionManager.class);
+    RequestFactory requestFactory = createStrictMock(RequestFactory.class);
+
+    // no expectations due to empty stage list
+    replay(actionManager, requestFactory);
+
+    RequestStageContainer requestStages = new RequestStageContainer(1L, null, requestFactory, actionManager);
+    requestStages.persist();
+
+    verify(actionManager, requestFactory);
+  }
+
+  @Test
+  public void testGetRequestStatusResponse() {
+    ActionManager actionManager = createStrictMock(ActionManager.class);
+    Stage stage1 = createNiceMock(Stage.class);
+    Stage stage2 = createNiceMock(Stage.class);
+    HostRoleCommand command1 = createNiceMock(HostRoleCommand.class);
+    Role role = createNiceMock(Role.class);
+    List<Stage> stages = new ArrayList<Stage>();
+    RoleCommand roleCommand = RoleCommand.INSTALL;
+    HostRoleStatus status = HostRoleStatus.IN_PROGRESS;
+    stages.add(stage1);
+    stages.add(stage2);
+    List<HostRoleCommand> hostRoleCommands = new ArrayList<HostRoleCommand>();
+    hostRoleCommands.add(command1);
+
+    expect(actionManager.getRequestTasks(100)).andReturn(hostRoleCommands);
+    expect(actionManager.getRequestContext(100)).andReturn("test");
+    expect(command1.getTaskId()).andReturn(1L);
+    expect(command1.getRoleCommand()).andReturn(roleCommand);
+    expect(command1.getRole()).andReturn(role);
+    expect(command1.getStatus()).andReturn(status);
+
+    replay(actionManager, stage1, stage2, command1, role);
+
+    RequestStageContainer requestStages = new RequestStageContainer(100L, stages, null, actionManager);
+    RequestStatusResponse response = requestStages.getRequestStatusResponse();
+
+    assertEquals(100, response.getRequestId());
+    List<ShortTaskStatus> tasks = response.getTasks();
+    assertEquals(1, tasks.size());
+    ShortTaskStatus task = tasks.get(0);
+    assertEquals(1, task.getTaskId());
+    assertEquals(roleCommand.toString(), task.getCommand());
+    assertEquals(status.toString(), task.getStatus());
+
+    assertEquals("test", response.getRequestContext());
+
+    verify(actionManager, stage1, stage2, command1, role);
+  }
+}

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ServiceResourceProviderTest.java

@@ -24,6 +24,7 @@ import static org.easymock.EasyMock.capture;
 import static org.easymock.EasyMock.createMock;
 import static org.easymock.EasyMock.createNiceMock;
 import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.isNull;
 import static org.easymock.EasyMock.replay;
 import static org.easymock.EasyMock.verify;
 
@@ -256,6 +257,7 @@ public class ServiceResourceProviderTest {
     Service service0 = createNiceMock(Service.class);
     ServiceFactory serviceFactory = createNiceMock(ServiceFactory.class);
     AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
+    RequestStageContainer requestStages = createNiceMock(RequestStageContainer.class);
     RequestStatusResponse requestStatusResponse = createNiceMock(RequestStatusResponse.class);
 
     Map<String, String> mapRequestProps = new HashMap<String, String>();
@@ -281,12 +283,16 @@ public class ServiceResourceProviderTest {
     Capture<Collection<ServiceComponentHost>> ignoredScHostsCapture = new Capture<Collection<ServiceComponentHost>>();
     Capture<Cluster> clusterCapture = new Capture<Cluster>();
 
-    expect(managementController.createStages(capture(clusterCapture), capture(requestPropertiesCapture), capture(requestParametersCapture), capture(changedServicesCapture), capture(changedCompsCapture), capture(changedScHostsCapture), capture(ignoredScHostsCapture), anyBoolean(), anyBoolean()
-    )).andReturn(requestStatusResponse);
+    expect(managementController.addStages((RequestStageContainer) isNull(), capture(clusterCapture), capture(requestPropertiesCapture),
+        capture(requestParametersCapture), capture(changedServicesCapture), capture(changedCompsCapture),
+        capture(changedScHostsCapture), capture(ignoredScHostsCapture), anyBoolean(), anyBoolean()
+    )).andReturn(requestStages);
+    requestStages.persist();
+    expect(requestStages.getRequestStatusResponse()).andReturn(requestStatusResponse);
 
     // replay
     replay(managementController, clusters, cluster,
-        service0, serviceFactory, ambariMetaInfo, requestStatusResponse);
+        service0, serviceFactory, ambariMetaInfo, requestStages, requestStatusResponse);
 
     ResourceProvider provider = getServiceProvider(managementController);
 
@@ -305,7 +311,7 @@ public class ServiceResourceProviderTest {
 
     // verify
     verify(managementController, clusters, cluster,
-        service0, serviceFactory, ambariMetaInfo, requestStatusResponse);
+        service0, serviceFactory, ambariMetaInfo, requestStages, requestStatusResponse);
   }
 
   @Test
@@ -318,6 +324,8 @@ public class ServiceResourceProviderTest {
     Service service0 = createNiceMock(Service.class);
     ServiceResponse serviceResponse0 = createNiceMock(ServiceResponse.class);
     AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
+    RequestStageContainer requestStages1 = createNiceMock(RequestStageContainer.class);
+    RequestStageContainer requestStages2 = createNiceMock(RequestStageContainer.class);
 
     RequestStatusResponse response1 = createNiceMock(RequestStatusResponse.class);
     RequestStatusResponse response2 = createNiceMock(RequestStatusResponse
@@ -357,14 +365,24 @@ public class ServiceResourceProviderTest {
     Capture<Collection<ServiceComponentHost>> ignoredScHostsCapture = new Capture<Collection<ServiceComponentHost>>();
     Capture<Cluster> clusterCapture = new Capture<Cluster>();
 
-    expect(managementController1.createStages(capture(clusterCapture), capture(requestPropertiesCapture), capture(requestParametersCapture), capture(changedServicesCapture), capture(changedCompsCapture), capture(changedScHostsCapture), capture(ignoredScHostsCapture), anyBoolean(), anyBoolean()
-    )).andReturn(response1);
+    expect(managementController1.addStages((RequestStageContainer) isNull(), capture(clusterCapture), capture(requestPropertiesCapture),
+        capture(requestParametersCapture), capture(changedServicesCapture), capture(changedCompsCapture),
+        capture(changedScHostsCapture), capture(ignoredScHostsCapture), anyBoolean(), anyBoolean()
+    )).andReturn(requestStages1);
 
-    expect(managementController2.createStages(capture(clusterCapture), capture(requestPropertiesCapture), capture(requestParametersCapture), capture(changedServicesCapture), capture(changedCompsCapture), capture(changedScHostsCapture), capture(ignoredScHostsCapture), anyBoolean(), anyBoolean()
-    )).andReturn(response2);
+    expect(managementController2.addStages((RequestStageContainer) isNull(), capture(clusterCapture), capture(requestPropertiesCapture),
+        capture(requestParametersCapture), capture(changedServicesCapture), capture(changedCompsCapture),
+        capture(changedScHostsCapture), capture(ignoredScHostsCapture), anyBoolean(), anyBoolean()
+    )).andReturn(requestStages2);
+
+    requestStages1.persist();
+    expect(requestStages1.getRequestStatusResponse()).andReturn(response1);
+
+    requestStages2.persist();
+    expect(requestStages2.getRequestStatusResponse()).andReturn(response2);
 
     // replay
-    replay(managementController1, response1, managementController2, response2,
+    replay(managementController1, response1, managementController2, requestStages1, requestStages2, response2,
         clusters, cluster, service0, serviceResponse0, ambariMetaInfo);
 
     ResourceProvider provider1 = getServiceProvider(managementController1);
@@ -397,7 +415,7 @@ public class ServiceResourceProviderTest {
     provider2.updateResources(request, predicate2);
 
     // verify
-    verify(managementController1, response1, managementController2, response2,
+    verify(managementController1, response1, managementController2, requestStages1, requestStages2, response2,
         clusters, cluster, service0, serviceResponse0, ambariMetaInfo);
   }
 
@@ -1105,7 +1123,10 @@ public class ServiceResourceProviderTest {
                                                      boolean reconfigureClients) throws AmbariException
   {
     ServiceResourceProvider provider = getServiceProvider(controller);
-    return provider.updateServices(requests, requestProperties, runSmokeTest, reconfigureClients);
+
+    RequestStageContainer request = provider.updateServices(null, requests, requestProperties, runSmokeTest, reconfigureClients);
+    request.persist();
+    return request.getRequestStatusResponse();
   }
 
   public static RequestStatusResponse deleteServices(AmbariManagementController controller, Set<ServiceRequest> requests)