[AMBARI-25240] : Dynamically update Rolling Upgrade Batch size (#2926)

* [AMBARI-25240] : Dynamically update Rolling Upgrade Batch size using cluster env property

* minor code style fix

ambari-server/src/main/java/org/apache/ambari/server/stack/MasterHostResolver.java

@@ -318,6 +318,17 @@ private List<HostsType.HighAvailabilityHosts> nameSpaces(Set<String> componentHo
 
   }
 
+  /**
+   * Find Config value for current Cluster using configType and propertyName
+   *
+   * @param configType   Config Type
+   * @param propertyName Property Name
+   * @return Value of property if present else null
+   */
+  public String getValueFromDesiredConfigurations(final String configType, final String propertyName) {
+    return m_configHelper.getValueFromDesiredConfigurations(m_cluster, configType, propertyName);
+  }
+
   /**
    * Find the master and secondary namenode(s) based on JMX NameNodeStatus.
    */

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

@@ -31,7 +31,9 @@
 import org.apache.ambari.server.stack.upgrade.ServiceCheckGrouping;
 import org.apache.ambari.server.stack.upgrade.Task;
 import org.apache.ambari.server.stack.upgrade.UpgradePack.ProcessingComponent;
+import org.apache.ambari.server.state.ConfigHelper;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang.StringUtils;
 
 /**
  * Defines how to build stages for an Upgrade or Downgrade.
@@ -279,6 +281,11 @@ protected int getParallelHostCount(UpgradeContext ctx, int defaultValue) {
 
     if (m_grouping.parallelScheduler != null) {
       int taskParallelism = m_grouping.parallelScheduler.maxDegreeOfParallelism;
+      String maxDegreeFromClusterEnv =
+              ctx.getResolver().getValueFromDesiredConfigurations(ConfigHelper.CLUSTER_ENV, "max_degree_parallelism");
+      if (StringUtils.isNotEmpty(maxDegreeFromClusterEnv) && StringUtils.isNumeric(maxDegreeFromClusterEnv)) {
+        taskParallelism = Integer.parseInt(maxDegreeFromClusterEnv);
+      }
       if (taskParallelism == ParallelScheduler.DEFAULT_MAX_DEGREE_OF_PARALLELISM) {
         taskParallelism = ctx.getDefaultMaxDegreeOfParallelism();
       }