AMBARI-22628 - YARN Shuffle Service Can't Be Found On Client-Only Nodes After New Cluster Install (jonathanhurley)

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

@@ -29,8 +29,6 @@ import org.apache.ambari.annotations.ExperimentalFeature;
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.RoleCommand;
 import org.apache.ambari.server.state.Cluster;
-import org.apache.ambari.server.state.Service;
-import org.apache.ambari.server.state.ServiceComponent;
 import org.apache.ambari.server.state.UpgradeContext.UpgradeSummary;
 import org.apache.ambari.server.utils.StageUtils;
 import org.slf4j.Logger;
@@ -560,30 +558,21 @@ public class ExecutionCommand extends AgentCommand {
   }
 
   /**
-   * Used to set a map of {service -> { component -> version}}.  This is necessary when performing
-   * an upgrade to correct build paths of required binaries.
-   * @param cluster the cluster from which to build the map
+   * Used to set a map of {service -> { component -> version}}. This is
+   * necessary when performing an upgrade to correct build paths of required
+   * binaries. This method will only set the version information for a component
+   * if:
+   * <ul>
+   * <li>The component advertises a version</li>
+   * <li>The repository for the component has been resolved and the version can
+   * be trusted</li>
+   * </ul>
+   *
+   * @param cluster
+   *          the cluster from which to build the map
    */
   public void setComponentVersions(Cluster cluster) throws AmbariException {
-    Map<String, Map<String, String>> componentVersionMap = new HashMap<>();
-
-    for (Service service : cluster.getServices().values()) {
-      Map<String, String> componentMap = new HashMap<>();
-
-      boolean shouldSet = false;
-      for (ServiceComponent component : service.getServiceComponents().values()) {
-        if (component.isVersionAdvertised()) {
-          shouldSet = true;
-          componentMap.put(component.getName(), component.getDesiredVersion());
-        }
-      }
-
-      if (shouldSet) {
-        componentVersionMap.put(service.getName(), componentMap);
-      }
-    }
-
-    this.componentVersionMap = componentVersionMap;
+    componentVersionMap = cluster.getComponentVersionMap();
   }
 
   /**

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

@@ -448,6 +448,8 @@ public class ClientConfigResourceProvider extends AbstractControllerResourceProv
         jsonContent.put("clusterName", cluster.getClusterName());
         jsonContent.put("serviceName", serviceName);
         jsonContent.put("role", componentName);
+        jsonContent.put("componentVersionMap", cluster.getComponentVersionMap());
+
         jsonConfigurations = gson.toJson(jsonContent);
 
         File tmpDirectory = new File(TMP_PATH);

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

@@ -678,4 +678,13 @@ public interface Cluster {
    */
   void addSuspendedUpgradeParameters(Map<String, String> commandParams,
       Map<String, String> roleParams);
+
+  /**
+   * Gets a mapping of service to component/version for every installed
+   * component in the cluster which advertises a version and for which the
+   * repository has been resolved.
+   *
+   * @return a mapping of service to component version, or an empty map.
+   */
+  Map<String, Map<String, String>> getComponentVersionMap();
 }

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

@@ -2808,4 +2808,35 @@ public class ClusterImpl implements Cluster {
     // suspended goes in role params
     roleParams.put(KeyNames.UPGRADE_SUSPENDED, Boolean.TRUE.toString().toLowerCase());
   }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public Map<String, Map<String, String>> getComponentVersionMap() {
+    Map<String, Map<String, String>> componentVersionMap = new HashMap<>();
+
+    for (Service service : getServices().values()) {
+      Map<String, String> componentMap = new HashMap<>();
+      for (ServiceComponent component : service.getServiceComponents().values()) {
+        // skip components which don't advertise a version
+        if (!component.isVersionAdvertised()) {
+          continue;
+        }
+
+        // if the repo isn't resolved, then we can't trust the version
+        if (!component.getDesiredRepositoryVersion().isResolved()) {
+          continue;
+        }
+
+        componentMap.put(component.getName(), component.getDesiredVersion());
+      }
+
+      if (!componentMap.isEmpty()) {
+        componentVersionMap.put(service.getName(), componentMap);
+      }
+    }
+
+    return componentVersionMap;
+  }
 }

ambari-server/src/main/resources/common-services/YARN/2.1.0.2.0/package/scripts/params_linux.py

@@ -21,6 +21,7 @@ Ambari Agent
 import os
 
 from resource_management.core import sudo
+from resource_management.core.logger import Logger
 from resource_management.libraries.script.script import Script
 from resource_management.libraries.resources.hdfs_resource import HdfsResource
 from resource_management.libraries.functions import component_version
@@ -84,11 +85,45 @@ stack_supports_timeline_state_store = check_stack_feature(StackFeature.TIMELINE_
 # It cannot be used during the initial Cluser Install because the version is not yet known.
 version = default("/commandParams/version", None)
 
+def get_spark_version(service_name, component_name, yarn_version):
+  """
+  Attempts to calculate the correct version placeholder value for spark or spark2 based on
+  what is installed in the cluster. If Spark is not installed, then this value will need to be
+  that of YARN so it can still find the correct shuffle class.
+
+  On cluster installs, we have not yet calcualted any versions and all known values could be None.
+  This doesn't affect daemons, but it does affect client-only hosts where they will never receive
+  a start command after install. Therefore, this function will attempt to use stack-select as a
+  last resort to get a value value.
+  :param service_name:  the service name (SPARK, SPARK2, etc)
+  :param component_name:  the component name (SPARK_CLIENT, etc)
+  :param yarn_version:  the default version of Yarn to use if no spark is installed
+  :return:  a value for the version placeholder in shuffle classpath properties
+  """
+  # start off seeing if we need to populate a default value for YARN
+  if yarn_version is None:
+    yarn_version = component_version.get_component_repository_version(service_name = "YARN",
+      component_name = "YARN_CLIENT")
+
+  # now try to get the version of spark/spark2, defaulting to the version if YARN
+  shuffle_classpath_version = component_version.get_component_repository_version(service_name = service_name,
+    component_name = component_name, default_value = yarn_version)
+
+  # even with the default of using YARN's version, on an install this might be None since we haven't
+  # calculated the version of YARN yet - use stack_select as a last ditch effort
+  if shuffle_classpath_version is None:
+    try:
+      shuffle_classpath_version = stack_select.get_role_component_current_stack_version()
+    except:
+      Logger.exception("Unable to query for the correct shuffle classpath")
+
+  return shuffle_classpath_version
+
 # these are used to render the classpath for picking up Spark classes
 # in the event that spark is not installed, then we must default to the vesrion of YARN installed
 # since it will still load classes from its own spark version
-spark_version = component_version.get_component_repository_version(service_name =  "SPARK", component_name = "SPARK_CLIENT", default_value = version)
-spark2_version = component_version.get_component_repository_version(service_name = "SPARK2", component_name = "SPARK2_CLIENT", default_value = version)
+spark_version = get_spark_version("SPARK", "SPARK_CLIENT", version)
+spark2_version = get_spark_version("SPARK2", "SPARK2_CLIENT", version)
 
 stack_supports_ranger_kerberos = check_stack_feature(StackFeature.RANGER_KERBEROS_SUPPORT, version_for_stack_feature_checks)
 stack_supports_ranger_audit_db = check_stack_feature(StackFeature.RANGER_AUDIT_DB_SUPPORT, version_for_stack_feature_checks)
@@ -541,5 +576,3 @@ if enable_ranger_yarn and is_supported_yarn_ranger:
 
 # need this to capture cluster name from where ranger yarn plugin is enabled
 cluster_name = config['clusterName']
-
-# ranger yarn plugin end section

ambari-server/src/test/java/org/apache/ambari/server/actionmanager/ExecutionCommandWrapperTest.java

@@ -291,7 +291,12 @@ public class ExecutionCommandWrapperTest {
     Cluster cluster = clusters.getCluster(CLUSTER1);
 
     StackId stackId = cluster.getDesiredStackVersion();
+    
+    // set the repo version resolved state to verify that the version is not sent
     RepositoryVersionEntity repositoryVersion = ormTestHelper.getOrCreateRepositoryVersion(stackId, "0.1-0000");
+    repositoryVersion.setResolved(false);
+    ormTestHelper.repositoryVersionDAO.merge(repositoryVersion);
+
     Service service = cluster.getService("HDFS");
     service.setDesiredRepositoryVersion(repositoryVersion);
 

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

@@ -564,6 +564,52 @@ public class AmbariCustomCommandExecutionHelperTest {
     Assert.assertTrue(command.getComponentVersionMap().containsKey("ZOOKEEPER"));
   }
 
+  /**
+   * Tests that if a component's repository is not resolved, then the repo
+   * version map does not get populated.
+   *
+   * @throws Exception
+   */
+  @Test
+  public void testAvailableServicesMapIsEmptyWhenRepositoriesNotResolved() throws Exception {
+
+    // set all repos to resolve=false to verify that we don't get a
+    // component version map
+    RepositoryVersionDAO repositoryVersionDAO = injector.getInstance(RepositoryVersionDAO.class);
+    List<RepositoryVersionEntity> repoVersions = repositoryVersionDAO.findAll();
+    for (RepositoryVersionEntity repoVersion : repoVersions) {
+      repoVersion.setResolved(false);
+      repositoryVersionDAO.merge(repoVersion);
+    }
+
+    Map<String, String> requestProperties = new HashMap<String, String>() {
+      {
+        put(REQUEST_CONTEXT_PROPERTY, "Refresh YARN Capacity Scheduler");
+        put("command", "REFRESHQUEUES");
+      }
+    };
+
+    ExecuteActionRequest actionRequest = new ExecuteActionRequest("c1", "REFRESHQUEUES",
+        new HashMap<String, String>() {
+          {
+            put("forceRefreshConfigTags", "capacity-scheduler");
+          }
+        }, false);
+
+    actionRequest.getResourceFilters().add(new RequestResourceFilter("YARN", "RESOURCEMANAGER",
+        Collections.singletonList("c1-c6401")));
+
+    EasyMock.replay(hostRoleCommand, actionManager, configHelper);
+
+    ambariManagementController.createAction(actionRequest, requestProperties);
+    Request request = requestCapture.getValue();
+    Stage stage = request.getStages().iterator().next();
+    List<ExecutionCommandWrapper> commands = stage.getExecutionCommands("c1-c6401");
+    ExecutionCommand command = commands.get(0).getExecutionCommand();
+
+    Assert.assertTrue(MapUtils.isEmpty(command.getComponentVersionMap()));
+  }
+
   @Test
   public void testCommandRepository() throws Exception {
     Cluster cluster = clusters.getCluster("c1");

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

@@ -674,6 +674,9 @@ public class OrmTestHelper {
 
         repositoryVersion = repositoryVersionDAO.create(stackEntity, version,
             String.valueOf(System.currentTimeMillis()) + uniqueCounter.incrementAndGet(), operatingSystems);
+
+        repositoryVersion.setResolved(true);
+        repositoryVersion = repositoryVersionDAO.merge(repositoryVersion);
       } catch (Exception ex) {
         LOG.error("Caught exception", ex);