AMBARI-13730: Express Upgrade: Add max degree of parallelism to control number of tasks run in parallel (jluniya)

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

@@ -289,7 +289,6 @@ public class UpgradeHelper {
 
       // Attempt to get the function of the group, during a NonRolling Upgrade
       Task.Type functionName = null;
-      boolean scheduleInParallel = false;
       // NonRolling defaults to not performing service checks on a group.
       // Of course, a Service Check Group does indeed run them.
       if (upgradePack.getType() == UpgradeType.NON_ROLLING) {
@@ -297,15 +296,12 @@ public class UpgradeHelper {
 
         if (RestartGrouping.class.isInstance(group)) {
           functionName = ((RestartGrouping) group).getFunction();
-          scheduleInParallel = true;
         }
         if (StartGrouping.class.isInstance(group)) {
           functionName = ((StartGrouping) group).getFunction();
-          scheduleInParallel = true;
         }
         if (StopGrouping.class.isInstance(group)) {
           functionName = ((StopGrouping) group).getFunction();
-          scheduleInParallel = true;
         }
       }
 
@@ -410,7 +406,7 @@ public class UpgradeHelper {
                   hostsType.hosts = order;
 
                   builder.add(context, hostsType, service.serviceName,
-                      svc.isClientOnlyService(), pc, null, false);
+                      svc.isClientOnlyService(), pc, null);
                 }
                 break;
               case NON_ROLLING:
@@ -435,21 +431,21 @@ public class UpgradeHelper {
 
 
                   builder.add(context, ht1, service.serviceName,
-                      svc.isClientOnlyService(), pc, h1Params, false);
+                      svc.isClientOnlyService(), pc, h1Params);
 
                   builder.add(context, ht2, service.serviceName,
-                      svc.isClientOnlyService(), pc, h2Params, false);
+                      svc.isClientOnlyService(), pc, h2Params);
                 } else {
                   // If no NameNode HA, then don't need to change hostsType.hosts since there should be exactly one.
                   builder.add(context, hostsType, service.serviceName,
-                      svc.isClientOnlyService(), pc, null, false);
+                      svc.isClientOnlyService(), pc, null);
                 }
 
                 break;
             }
           } else {
             builder.add(context, hostsType, service.serviceName,
-                svc.isClientOnlyService(), pc, null, scheduleInParallel);
+                svc.isClientOnlyService(), pc, null);
           }
         }
       }

ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ClusterGrouping.java

@@ -108,7 +108,7 @@ public class ClusterGrouping extends Grouping {
 
     @Override
     public void add(UpgradeContext ctx, HostsType hostsType, String service,
-        boolean clientOnly, ProcessingComponent pc, Map<String, String> params, boolean scheduleInParallel) {
+        boolean clientOnly, ProcessingComponent pc, Map<String, String> params) {
       // !!! no-op in this case
     }
 

ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ColocatedGrouping.java

@@ -81,7 +81,7 @@ public class ColocatedGrouping extends Grouping {
 
     @Override
     public void add(UpgradeContext ctx, HostsType hostsType, String service,
-        boolean clientOnly, ProcessingComponent pc, Map<String, String> params, boolean scheduleInParallel) {
+        boolean clientOnly, ProcessingComponent pc, Map<String, String> params) {
 
       boolean forUpgrade = ctx.getDirection().isUpgrade();
 

ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/Grouping.java

@@ -32,6 +32,7 @@ import org.apache.ambari.server.stack.HostsType;
 import org.apache.ambari.server.state.UpgradeContext;
 import org.apache.ambari.server.state.stack.UpgradePack;
 import org.apache.ambari.server.state.stack.UpgradePack.ProcessingComponent;
+import org.apache.ambari.server.utils.SetUtils;
 import org.apache.commons.lang.StringUtils;
 
 /**
@@ -61,6 +62,9 @@ public class Grouping {
   @XmlElement(name="direction")
   public Direction intendedDirection = null;
 
+  @XmlElement(name="parallel-scheduler")
+  public ParallelScheduler parallelScheduler;
+
   /**
    * Gets the default builder.
    */
@@ -89,7 +93,7 @@ public class Grouping {
      */
     @Override
     public void add(UpgradeContext ctx, HostsType hostsType, String service,
-       boolean clientOnly, ProcessingComponent pc, Map<String, String> params, boolean scheduleInParallel) {
+       boolean clientOnly, ProcessingComponent pc, Map<String, String> params) {
 
       boolean forUpgrade = ctx.getDirection().isUpgrade();
 
@@ -112,14 +116,18 @@ public class Grouping {
       // Add the processing component
       if (null != pc.tasks && 1 == pc.tasks.size()) {
         Task t = pc.tasks.get(0);
-        if(scheduleInParallel) {
-          // Create single stage for all
-          StageWrapper stage = new StageWrapper(
-              t.getStageWrapperType(),
-              getStageText(t.getActionVerb(), ctx.getComponentDisplay(service, pc.name), hostsType.hosts),
-              params,
-              new TaskWrapper(service, pc.name, hostsType.hosts, params, t));
-          m_stages.add(stage);
+        if(m_grouping.parallelScheduler != null) {
+          List<Set<String>> hostSets = SetUtils.split(hostsType.hosts, m_grouping.parallelScheduler.maxDegreeOfParallelism);
+          int batchNum = 1;
+          for(Set<String> hosts : hostSets) {
+            // Create single stage for all
+            StageWrapper stage = new StageWrapper(
+                t.getStageWrapperType(),
+                getStageText(t.getActionVerb(), ctx.getComponentDisplay(service, pc.name), hosts, batchNum++, hostSets.size()),
+                params,
+                new TaskWrapper(service, pc.name, hosts, params, t));
+            m_stages.add(stage);
+          }
         } else {
           for (String hostName : hostsType.hosts) {
             StageWrapper stage = new StageWrapper(

ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ParallelScheduler.java

@@ -0,0 +1,31 @@
+/**
+ * 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.state.stack.upgrade;
+
+import javax.xml.bind.annotation.XmlElement;
+
+/**
+ *  Identifies if component instances should be upgraded in parallel (optional)
+ */
+public class ParallelScheduler {
+
+  public static int DEFAULT_MAX_DEGREE_OF_PARALLELISM = 100;
+
+  @XmlElement(name="max-degree-of-parallelism")
+  public int maxDegreeOfParallelism = DEFAULT_MAX_DEGREE_OF_PARALLELISM;
+}

ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ServiceCheckGrouping.java

@@ -103,7 +103,7 @@ public class ServiceCheckGrouping extends Grouping {
      */
     @Override
     public void add(UpgradeContext ctx, HostsType hostsType, String service,
-        boolean clientOnly, ProcessingComponent pc, Map<String, String> params, boolean scheduleInParallel) {
+        boolean clientOnly, ProcessingComponent pc, Map<String, String> params) {
       // !!! nothing to do here
     }
 

ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/StageWrapperBuilder.java

@@ -71,7 +71,7 @@ public abstract class StageWrapperBuilder {
    *          additional parameters
    */
   public abstract void add(UpgradeContext upgradeContext, HostsType hostsType, String service,
-      boolean clientOnly, ProcessingComponent pc, Map<String, String> params, boolean scheduleInParallel);
+      boolean clientOnly, ProcessingComponent pc, Map<String, String> params);
 
   /**
    * Builds the stage wrappers, including any pre- and post-procesing that needs
@@ -160,6 +160,19 @@ public abstract class StageWrapperBuilder {
     return stageWrappers;
   }
 
+  /**
+   * Consistently formats a string.
+   * @param prefix
+   * @param component
+   * @param batchNum
+   * @param totalBatches
+   * @return the prepared string
+   */
+  protected String getStageText(String prefix, String component, Set<String> hosts, int batchNum, int totalBatches) {
+    String stageText = getStageText(prefix, component, hosts);
+    String batchText = 1 == totalBatches? "" : String.format(" ( Batch %s of %s )", batchNum, totalBatches);
+    return stageText + batchText;
+  }
 
   /**
    * Consistently formats a string.

ambari-server/src/main/java/org/apache/ambari/server/utils/SetUtils.java

@@ -0,0 +1,62 @@
+/**
+ * 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.utils;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.Collections;
+
+/**
+ * Static Helper methods for sets
+ */
+public class SetUtils {
+
+  /**
+   * Split a set into subsets
+   * @param original The original set to be split
+   * @param subsetSize Size of the subset (except for final subset)
+   * @param <T> Data type of set elements
+   * @return List of subsets
+   */
+  public static <T> List<Set<T>> split(Set<T> original, int subsetSize) {
+
+    if(subsetSize <= 0) {
+      throw new IllegalArgumentException("Incorrect max size");
+    }
+
+    if(original == null || original.isEmpty()) {
+      return Collections.emptyList();
+    }
+
+    int subsetCount = (int) (Math.ceil((double)original.size() / subsetSize));
+    ArrayList<Set<T>> subsets = new ArrayList<Set<T>>(subsetCount);
+    Iterator<T> iterator = original.iterator();
+
+    for(int i = 0; i < subsetCount; i++) {
+      Set<T> subset = new LinkedHashSet<T>(subsetSize);
+      for(int j = 0; j < subsetSize && iterator.hasNext(); j++) {
+        subset.add(iterator.next());
+      }
+      subsets.add(subset);
+    }
+    return subsets;
+  }
+}

ambari-server/src/main/resources/stacks/HDP/2.1/upgrades/nonrolling-upgrade-2.3.xml

@@ -58,6 +58,7 @@
       <direction>UPGRADE</direction>
       <skippable>true</skippable>
       <service-check>false</service-check>
+      <parallel-scheduler></parallel-scheduler>
 
       <service name="FLUME">
         <component>FLUME_HANDLER</component>
@@ -130,6 +131,7 @@
       <direction>UPGRADE</direction>
       <skippable>true</skippable>
       <service-check>false</service-check>
+      <parallel-scheduler></parallel-scheduler>
 
       <service name="HBASE">
         <component>HBASE_REGIONSERVER</component>
@@ -328,6 +330,7 @@
     <group xsi:type="restart" name="ZOOKEEPER" title="ZooKeeper">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="ZOOKEEPER">
         <service-check>false</service-check>
         <component>ZOOKEEPER_SERVER</component>
@@ -338,6 +341,7 @@
     <group xsi:type="restart" name="HDFS" title="HDFS">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HDFS">
         <component>JOURNALNODE</component>
         <component>ZKFC</component>
@@ -364,6 +368,7 @@
     <group xsi:type="restart" name="YARN and MapReduce2" title="YARN and MapReduce2">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="MAPREDUCE2">
         <component>HISTORYSERVER</component>
         <component>MAPREDUCE2_CLIENT</component>
@@ -379,6 +384,7 @@
     <group xsi:type="restart" name="HBASE" title="HBASE">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HBASE">
         <component>HBASE_MASTER</component>
         <component>HBASE_REGIONSERVER</component>
@@ -389,6 +395,7 @@
     <group xsi:type="restart" name="CLIENTS" title="Tez, Pig, Sqoop Clients">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="TEZ">
         <component>TEZ_CLIENT</component>
       </service>
@@ -416,6 +423,7 @@
     <group xsi:type="restart" name="HIVE" title="Hive">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HIVE">
         <component>HIVE_METASTORE</component>
         <component>HIVE_SERVER</component>
@@ -428,6 +436,7 @@
     <group xsi:type="restart" name="OOZIE" title="Oozie">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="OOZIE">
         <component>OOZIE_SERVER</component>
         <component>OOZIE_CLIENT</component>
@@ -437,6 +446,7 @@
     <group xsi:type="restart" name="FALCON" title="Falcon">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="FALCON">
         <component>FALCON_SERVER</component>
         <component>FALCON_CLIENT</component>
@@ -446,6 +456,7 @@
     <group xsi:type="restart" name="STORM" title="Storm">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="STORM">
         <component>NIMBUS</component>
         <component>STORM_REST_API</component>
@@ -464,6 +475,7 @@
     <group xsi:type="restart" name="FLUME" title="Flume">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="FLUME">
         <component>FLUME_HANDLER</component>
       </service>

ambari-server/src/main/resources/stacks/HDP/2.2/upgrades/nonrolling-upgrade-2.2.xml

@@ -57,6 +57,7 @@
       <direction>UPGRADE</direction>
       <skippable>true</skippable>
       <service-check>false</service-check>
+      <parallel-scheduler></parallel-scheduler>
 
       <service name="FLUME">
         <component>FLUME_HANDLER</component>
@@ -151,6 +152,7 @@
     <group xsi:type="stop" name="STOP_LOW_LEVEL_SERVICE_COMPONENTS" title="Stop Components for Core Services">
       <direction>UPGRADE</direction>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service-check>false</service-check>
 
       <service name="HBASE">
@@ -235,6 +237,7 @@
     <group xsi:type="restart" name="ZOOKEEPER" title="ZooKeeper">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="ZOOKEEPER">
         <service-check>false</service-check>        <!-- TODO, enable service-check once done testing -->
         <component>ZOOKEEPER_SERVER</component>
@@ -245,6 +248,7 @@
     <group xsi:type="restart" name="RANGER" title="Ranger">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="RANGER">
         <component>RANGER_ADMIN</component>
         <component>RANGER_USERSYNC</component>
@@ -254,6 +258,7 @@
     <group xsi:type="restart" name="HDFS" title="HDFS">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HDFS">
         <component>JOURNALNODE</component>
         <component>ZKFC</component>
@@ -280,6 +285,7 @@
     <group xsi:type="restart" name="YARN and MapReduce2" title="YARN and MapReduce2">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="MAPREDUCE2">
         <component>HISTORYSERVER</component>
         <component>MAPREDUCE2_CLIENT</component>    <!-- TODO, parallelize -->
@@ -295,6 +301,7 @@
     <group xsi:type="restart" name="HBASE" title="HBASE">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HBASE">
         <component>HBASE_MASTER</component>
         <component>HBASE_REGIONSERVER</component>   <!-- TODO, parallelize -->
@@ -305,6 +312,7 @@
     <group xsi:type="restart" name="CLIENTS" title="Tez, Pig, Sqoop Clients">  <!-- TODO, parallelize -->
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="TEZ">
         <component>TEZ_CLIENT</component>
       </service>
@@ -333,6 +341,7 @@
     <group xsi:type="restart" name="HIVE" title="Hive">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HIVE">
         <component>HIVE_METASTORE</component>
         <component>HIVE_SERVER</component>
@@ -345,6 +354,7 @@
     <group xsi:type="restart" name="SPARK" title="Spark">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="SPARK">
         <component>SPARK_JOBHISTORYSERVER</component>
         <component>SPARK_CLIENT</component>         <!-- TODO, parallelize -->
@@ -378,6 +388,7 @@
     <group xsi:type="restart" name="OOZIE" title="Oozie">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="OOZIE">
         <component>OOZIE_SERVER</component>
         <component>OOZIE_CLIENT</component>         <!-- TODO, parallelize -->
@@ -387,6 +398,7 @@
     <group xsi:type="restart" name="FALCON" title="Falcon">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="FALCON">
         <component>FALCON_SERVER</component>
         <component>FALCON_CLIENT</component>        <!-- TODO, parallelize -->
@@ -396,6 +408,7 @@
     <group xsi:type="restart" name="KAFKA" title="Kafka">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="KAFKA">
         <component>KAFKA_BROKER</component>
       </service>
@@ -404,6 +417,7 @@
     <group xsi:type="restart" name="KNOX" title="Knox">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="KNOX">
         <component>KNOX_GATEWAY</component>
       </service>
@@ -412,6 +426,7 @@
     <group xsi:type="restart" name="STORM" title="Storm">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="STORM">
         <component>NIMBUS</component>
         <component>SUPERVISOR</component>
@@ -430,6 +445,7 @@
     <group xsi:type="restart" name="SLIDER" title="Slider">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="SLIDER">
         <component>SLIDER</component>
       </service>
@@ -438,6 +454,7 @@
     <group xsi:type="restart" name="FLUME" title="Flume">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="FLUME">
         <component>FLUME_HANDLER</component>
       </service>

ambari-server/src/main/resources/stacks/HDP/2.2/upgrades/nonrolling-upgrade-2.3.xml

@@ -79,6 +79,7 @@
       <direction>UPGRADE</direction>
       <skippable>true</skippable>
       <service-check>false</service-check>
+      <parallel-scheduler></parallel-scheduler>
 
       <service name="FLUME">
         <component>FLUME_HANDLER</component>
@@ -183,6 +184,7 @@
       <direction>UPGRADE</direction>
       <skippable>true</skippable>
       <service-check>false</service-check>
+      <parallel-scheduler></parallel-scheduler>
 
       <service name="HBASE">
         <component>HBASE_REGIONSERVER</component>
@@ -495,6 +497,7 @@
     <group xsi:type="restart" name="ZOOKEEPER" title="ZooKeeper">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="ZOOKEEPER">
         <service-check>false</service-check>
         <component>ZOOKEEPER_SERVER</component>
@@ -505,6 +508,7 @@
     <group xsi:type="restart" name="RANGER" title="Ranger">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="RANGER">
         <component>RANGER_ADMIN</component>
         <component>RANGER_USERSYNC</component>
@@ -514,6 +518,7 @@
     <group xsi:type="restart" name="HDFS" title="HDFS">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HDFS">
         <component>JOURNALNODE</component>
         <component>ZKFC</component>
@@ -540,6 +545,7 @@
     <group xsi:type="restart" name="YARN and MapReduce2" title="YARN and MapReduce2">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="MAPREDUCE2">
         <component>HISTORYSERVER</component>
         <component>MAPREDUCE2_CLIENT</component>
@@ -555,6 +561,7 @@
     <group xsi:type="restart" name="HBASE" title="HBASE">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HBASE">
         <component>HBASE_MASTER</component>
         <component>HBASE_REGIONSERVER</component>
@@ -565,7 +572,7 @@
     <group xsi:type="restart" name="CLIENTS" title="Tez, Pig, Sqoop Clients">
       <service-check>false</service-check>
       <skippable>true</skippable>
-
+      <parallel-scheduler></parallel-scheduler>
       <service name="TEZ">
         <component>TEZ_CLIENT</component>
       </service>
@@ -593,6 +600,7 @@
     <group xsi:type="restart" name="HIVE" title="Hive">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HIVE">
         <component>HIVE_METASTORE</component>
         <component>HIVE_SERVER</component>
@@ -605,6 +613,7 @@
     <group xsi:type="restart" name="SPARK" title="Spark">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="SPARK">
         <component>SPARK_JOBHISTORYSERVER</component>
         <component>SPARK_CLIENT</component>         <!-- TODO, parallelize -->
@@ -638,6 +647,7 @@
     <group xsi:type="restart" name="OOZIE" title="Oozie">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="OOZIE">
         <component>OOZIE_SERVER</component>
         <component>OOZIE_CLIENT</component>
@@ -647,6 +657,7 @@
     <group xsi:type="restart" name="FALCON" title="Falcon">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="FALCON">
         <component>FALCON_SERVER</component>
         <component>FALCON_CLIENT</component>
@@ -656,6 +667,7 @@
     <group xsi:type="restart" name="KAFKA" title="Kafka">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="KAFKA">
         <component>KAFKA_BROKER</component>
       </service>
@@ -664,6 +676,7 @@
     <group xsi:type="restart" name="KNOX" title="Knox">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="KNOX">
         <component>KNOX_GATEWAY</component>
       </service>
@@ -672,6 +685,7 @@
     <group xsi:type="restart" name="STORM" title="Storm">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="STORM">
         <component>NIMBUS</component>
         <component>SUPERVISOR</component>
@@ -689,6 +703,7 @@
     <group xsi:type="restart" name="SLIDER" title="Slider">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="SLIDER">
         <component>SLIDER</component>
       </service>
@@ -697,6 +712,7 @@
     <group xsi:type="restart" name="FLUME" title="Flume">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="FLUME">
         <component>FLUME_HANDLER</component>
       </service>
@@ -705,6 +721,7 @@
     <group xsi:type="restart" name="ACCUMULO" title="Accumulo">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="ACCUMULO">
         <component>ACCUMULO_MASTER</component>
         <component>ACCUMULO_TSERVER</component>

ambari-server/src/main/resources/stacks/HDP/2.3/upgrades/nonrolling-upgrade-2.3.xml

@@ -56,6 +56,7 @@
     <group xsi:type="stop" name="STOP_HIGH_LEVEL_SERVICE_COMPONENTS" title="Stop Components for High-Level Services">
       <direction>UPGRADE</direction>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service-check>false</service-check>
 
       <service name="ATLAS">
@@ -171,6 +172,7 @@
       <direction>UPGRADE</direction>
       <skippable>true</skippable>
       <service-check>false</service-check>
+      <parallel-scheduler></parallel-scheduler>
 
       <service name="HBASE">
         <component>HBASE_REGIONSERVER</component>
@@ -266,6 +268,7 @@
     <group xsi:type="restart" name="ZOOKEEPER" title="ZooKeeper">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="ZOOKEEPER">
         <service-check>false</service-check>        <!-- TODO (Alejandro), enable service-check once done testing -->
         <component>ZOOKEEPER_SERVER</component>
@@ -276,6 +279,7 @@
     <group xsi:type="restart" name="RANGER" title="Ranger">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="RANGER">
         <component>RANGER_ADMIN</component>
         <component>RANGER_USERSYNC</component>
@@ -285,6 +289,7 @@
     <group xsi:type="restart" name="RANGER_KMS" title="Ranger KMS">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="RANGER_KMS">
         <component>RANGER_KMS_SERVER</component>
       </service>
@@ -293,6 +298,7 @@
     <group xsi:type="restart" name="HDFS" title="HDFS">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HDFS">
         <component>JOURNALNODE</component>
         <component>ZKFC</component>
@@ -320,6 +326,7 @@
     <group xsi:type="restart" name="YARN and MapReduce2" title="YARN and MapReduce2">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="MAPREDUCE2">
         <component>HISTORYSERVER</component>
         <component>MAPREDUCE2_CLIENT</component>
@@ -335,6 +342,7 @@
     <group xsi:type="restart" name="HBASE" title="HBASE">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HBASE">
         <component>HBASE_MASTER</component>
         <component>HBASE_REGIONSERVER</component>
@@ -346,7 +354,7 @@
     <group xsi:type="restart" name="CLIENTS" title="Tez, Pig, Sqoop Clients">
       <service-check>false</service-check>
       <skippable>true</skippable>
-
+      <parallel-scheduler></parallel-scheduler>
       <service name="TEZ">
         <component>TEZ_CLIENT</component>
       </service>
@@ -378,6 +386,7 @@
     <group xsi:type="restart" name="HIVE" title="Hive">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="HIVE">
         <component>HIVE_METASTORE</component>
         <component>HIVE_SERVER</component>
@@ -390,6 +399,7 @@
     <group xsi:type="restart" name="SPARK" title="Spark">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="SPARK">
         <component>SPARK_JOBHISTORYSERVER</component>
         <component>SPARK_CLIENT</component>
@@ -399,6 +409,7 @@
     <group xsi:type="restart" name="ATLAS" title="Atlas">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="ATLAS">
         <component>ATLAS_SERVER</component>
       </service>
@@ -431,6 +442,7 @@
     <group xsi:type="restart" name="OOZIE" title="Oozie">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="OOZIE">
         <component>OOZIE_SERVER</component>
         <component>OOZIE_CLIENT</component>
@@ -440,6 +452,7 @@
     <group xsi:type="restart" name="FALCON" title="Falcon">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="FALCON">
         <component>FALCON_SERVER</component>
         <component>FALCON_CLIENT</component>
@@ -449,6 +462,7 @@
     <group xsi:type="restart" name="KAFKA" title="Kafka">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="KAFKA">
         <component>KAFKA_BROKER</component>
       </service>
@@ -457,6 +471,7 @@
     <group xsi:type="restart" name="KNOX" title="Knox">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="KNOX">
         <component>KNOX_GATEWAY</component>
       </service>
@@ -465,6 +480,7 @@
     <group xsi:type="restart" name="STORM" title="Storm">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="STORM">
         <component>NIMBUS</component>
         <component>SUPERVISOR</component>
@@ -483,6 +499,7 @@
     <group xsi:type="restart" name="SLIDER" title="Slider">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="SLIDER">
         <component>SLIDER</component>
       </service>
@@ -491,6 +508,7 @@
     <group xsi:type="restart" name="FLUME" title="Flume">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="FLUME">
         <component>FLUME_HANDLER</component>
       </service>
@@ -499,6 +517,7 @@
     <group xsi:type="restart" name="ACCUMULO" title="Accumulo">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="ACCUMULO">
         <component>ACCUMULO_MASTER</component>
         <component>ACCUMULO_TSERVER</component>

ambari-server/src/test/java/org/apache/ambari/server/state/stack/UpgradePackTest.java

@@ -40,6 +40,7 @@ import org.apache.ambari.server.state.stack.upgrade.ConfigureTask;
 import org.apache.ambari.server.state.stack.upgrade.ConfigUpgradeChangeDefinition.Transfer;
 import org.apache.ambari.server.state.stack.upgrade.Direction;
 import org.apache.ambari.server.state.stack.upgrade.Grouping;
+import org.apache.ambari.server.state.stack.upgrade.ParallelScheduler;
 import org.apache.ambari.server.state.stack.upgrade.RestartGrouping;
 import org.apache.ambari.server.state.stack.upgrade.RestartTask;
 import org.apache.ambari.server.state.stack.upgrade.StopGrouping;
@@ -339,51 +340,65 @@ public class UpgradePackTest {
     assertEquals(ClusterGrouping.class, group.getClass());
     clusterGroup = (ClusterGrouping) group;
     assertEquals("Prepare Upgrade", clusterGroup.title);
+    assertNull(clusterGroup.parallelScheduler);
 
     group = groups.get(1);
     assertEquals(StopGrouping.class, group.getClass());
     stopGroup = (StopGrouping) group;
     assertEquals("Stop Daemons for High-Level Services", stopGroup.title);
+    assertNotNull(stopGroup.parallelScheduler);
+    assertEquals(ParallelScheduler.DEFAULT_MAX_DEGREE_OF_PARALLELISM, stopGroup.parallelScheduler.maxDegreeOfParallelism);
 
     group = groups.get(2);
     assertEquals(ClusterGrouping.class, group.getClass());
     clusterGroup = (ClusterGrouping) group;
     assertEquals("Take Backups", clusterGroup.title);
+    assertNull(clusterGroup.parallelScheduler);
 
     group = groups.get(3);
     assertEquals(StopGrouping.class, group.getClass());
     stopGroup = (StopGrouping) group;
     assertEquals("Stop Daemons for Low-Level Services", stopGroup.title);
+    assertNotNull(stopGroup.parallelScheduler);
+    assertEquals(ParallelScheduler.DEFAULT_MAX_DEGREE_OF_PARALLELISM, stopGroup.parallelScheduler.maxDegreeOfParallelism);
 
     group = groups.get(4);
     assertEquals(UpdateStackGrouping.class, group.getClass());
     updateStackGroup = (UpdateStackGrouping) group;
     assertEquals("Update Desired Stack Id", updateStackGroup.title);
+    assertNull(updateStackGroup.parallelScheduler);
 
     group = groups.get(5);
     assertEquals(ClusterGrouping.class, group.getClass());
     clusterGroup = (ClusterGrouping) group;
     assertEquals("Set Version On All Hosts", clusterGroup.title);
+    assertNull(clusterGroup.parallelScheduler);
 
     group = groups.get(6);
     assertEquals(RestartGrouping.class, group.getClass());
     restartGroup = (RestartGrouping) group;
     assertEquals("Zookeeper", restartGroup.title);
+    assertNull(restartGroup.parallelScheduler);
 
     group = groups.get(7);
     assertEquals(RestartGrouping.class, group.getClass());
     restartGroup = (RestartGrouping) group;
     assertEquals("HDFS", restartGroup.title);
+    assertNotNull(restartGroup.parallelScheduler);
+    assertEquals(2, restartGroup.parallelScheduler.maxDegreeOfParallelism);
 
     group = groups.get(8);
     assertEquals(RestartGrouping.class, group.getClass());
     restartGroup = (RestartGrouping) group;
     assertEquals("MR and YARN", restartGroup.title);
+    assertNotNull(restartGroup.parallelScheduler);
+    assertEquals(ParallelScheduler.DEFAULT_MAX_DEGREE_OF_PARALLELISM, restartGroup.parallelScheduler.maxDegreeOfParallelism);
 
     group = groups.get(9);
     assertEquals(ClusterGrouping.class, group.getClass());
     clusterGroup = (ClusterGrouping) group;
     assertEquals("Finalize {{direction.text.proper}}", clusterGroup.title);
+    assertNull(clusterGroup.parallelScheduler);
   }
 
   private int indexOf(Map<String, ?> map, String keyToFind) {

ambari-server/src/test/java/org/apache/ambari/server/state/stack/upgrade/StageWrapperBuilderTest.java

@@ -126,7 +126,7 @@ public class StageWrapperBuilderTest {
      */
     @Override
     public void add(UpgradeContext upgradeContext, HostsType hostsType, String service,
-        boolean clientOnly, ProcessingComponent pc, Map<String, String> params, boolean scheduleInParallel) {
+        boolean clientOnly, ProcessingComponent pc, Map<String, String> params) {
     }
 
     /**

ambari-server/src/test/java/org/apache/ambari/server/utils/SetUtilsTest.java

@@ -0,0 +1,101 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.utils;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.Collections;
+
+import junit.framework.Assert;
+import org.junit.Test;
+
+/**
+ * Tests {@link org.apache.ambari.server.utils.SetUtils}
+ */
+public class SetUtilsTest {
+
+  @Test
+  public void testSplit() {
+
+    try {
+      SetUtils.split(null, 0);
+      Assert.fail("Expected IllegalArgumentException.");
+    } catch (IllegalArgumentException e) {
+      // expected
+    }
+
+    int size = 10;
+    List<Set<Integer>> subsets = SetUtils.split(null, size);
+    Assert.assertEquals(0, subsets.size());
+
+    subsets = SetUtils.split(Collections.EMPTY_SET, size);
+    Assert.assertEquals(0, subsets.size());
+
+    subsets = SetUtils.split(Collections.singleton(0), size);
+    Assert.assertEquals(1, subsets.size());
+    Assert.assertEquals(1, subsets.get(0).size());
+
+    Set<Integer> set = new LinkedHashSet<Integer>(5);
+    for(int i = 0; i < 5; i++) {
+      set.add(i);
+    }
+    subsets = SetUtils.split(set, size);
+    Assert.assertEquals(1, subsets.size());
+    Assert.assertEquals(5, subsets.get(0).size());
+
+
+    set = new LinkedHashSet<Integer>(10);
+    for(int i = 0; i < 10; i++) {
+      set.add(i);
+    }
+    subsets = SetUtils.split(set, size);
+    Assert.assertEquals(1, subsets.size());
+    Assert.assertEquals(10, subsets.get(0).size());
+
+    set = new LinkedHashSet<Integer>(11);
+    for(int i = 0; i < 11; i++) {
+      set.add(i);
+    }
+    subsets = SetUtils.split(set, size);
+    Assert.assertEquals(2, subsets.size());
+    Assert.assertEquals(10, subsets.get(0).size());
+    Assert.assertEquals(1, subsets.get(1).size());
+
+    set = new LinkedHashSet<Integer>(20);
+    for(int i = 0; i < 20; i++) {
+      set.add(i);
+    }
+    subsets = SetUtils.split(set, size);
+    Assert.assertEquals(2, subsets.size());
+    Assert.assertEquals(10, subsets.get(0).size());
+    Assert.assertEquals(10, subsets.get(1).size());
+
+    set = new LinkedHashSet<Integer>(27);
+    for(int i = 0; i < 27; i++) {
+      set.add(i);
+    }
+    subsets = SetUtils.split(set, size);
+    Assert.assertEquals(3, subsets.size());
+    Assert.assertEquals(10, subsets.get(0).size());
+    Assert.assertEquals(10, subsets.get(1).size());
+    Assert.assertEquals(7, subsets.get(2).size());
+  }
+}

ambari-server/src/test/resources/stacks/HDP/2.1.1/upgrades/upgrade_test_nonrolling.xml

@@ -48,6 +48,7 @@
       <direction>UPGRADE</direction>
       <skippable>true</skippable>
       <service-check>false</service-check>
+      <parallel-scheduler></parallel-scheduler>
 
       <service name="YARN">
         <component>NODEMANAGER</component>
@@ -76,6 +77,7 @@
       <direction>UPGRADE</direction>
       <skippable>true</skippable>
       <service-check>false</service-check>
+      <parallel-scheduler></parallel-scheduler>
 
       <service name="HDFS">
         <component>DATANODE</component>
@@ -133,6 +135,9 @@
     <group xsi:type="restart" name="HDFS" title="HDFS">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler>
+        <max-degree-of-parallelism>2</max-degree-of-parallelism>
+      </parallel-scheduler>
       <service name="HDFS">
         <component>JOURNALNODE</component>
         <component>ZKFC</component>
@@ -146,6 +151,7 @@
     <group xsi:type="restart" name="MR and YARN" title="MR and YARN">
       <service-check>false</service-check>
       <skippable>true</skippable>
+      <parallel-scheduler></parallel-scheduler>
       <service name="MAPREDUCE2">
         <component>HISTORYSERVER</component>
         <component>MAPREDUCE2_CLIENT</component>