AMBARI-17658 Storm nimbus server fails to come up with CNF backtype.storm.metric.IClusterReporter error (dsen)

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

@@ -25,6 +25,7 @@ import java.util.Map;
 import java.util.Set;
 
 import org.apache.ambari.server.RoleCommand;
+import org.apache.ambari.server.state.ServiceInfo;
 import org.apache.ambari.server.utils.StageUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -112,7 +113,7 @@ public class ExecutionCommand extends AgentCommand {
   private Set<String> localComponents = new HashSet<String>();
 
   @SerializedName("availableServices")
-  private Set<String> availableServices = new HashSet<String>();
+  private Map<String, String> availableServices = new HashMap<>();
 
   public String getCommandId() {
     return commandId;
@@ -264,12 +265,16 @@ public class ExecutionCommand extends AgentCommand {
     this.localComponents = localComponents;
   }
 
-  public Set<String> getAvailableServices() {
+  public Map<String, String> getAvailableServices() {
     return availableServices;
   }
 
-  public void setAvailableServices(Set<String> availableServices) {
-    this.availableServices = availableServices;
+  public void setAvailableServicesFromServiceInfoMap(Map<String, ServiceInfo> serviceInfoMap) {
+    Map<String, String> serviceVersionMap = new HashMap<>();
+    for (Map.Entry<String, ServiceInfo> entry : serviceInfoMap.entrySet()) {
+      serviceVersionMap.put(entry.getKey(), entry.getValue().getVersion());
+    }
+    this.availableServices = serviceVersionMap;
   }
 
   public Map<String, Map<String, Map<String, String>>> getConfigurationAttributes() {

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

@@ -367,7 +367,7 @@ public class AmbariCustomCommandExecutionHelper {
         execCmd.setForceRefreshConfigTagsBeforeExecution(parseAndValidateComponentsMapping(actionExecutionContext.getParameters().get(KeyNames.REFRESH_CONFIG_TAGS_BEFORE_EXECUTION)));
       }
 
-      execCmd.setAvailableServices(ambariMetaInfo.getServices(stackId.getStackName(), stackId.getStackVersion()).keySet());
+      execCmd.setAvailableServicesFromServiceInfoMap(ambariMetaInfo.getServices(stackId.getStackName(), stackId.getStackVersion()));
 
       Map<String, String> hostLevelParams = new TreeMap<String, String>();
 

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

@@ -2309,7 +2309,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
 
     execCmd.setRoleParams(roleParams);
     
-    execCmd.setAvailableServices(ambariMetaInfo.getServices(stackId.getStackName(), stackId.getStackVersion()).keySet());
+    execCmd.setAvailableServicesFromServiceInfoMap(ambariMetaInfo.getServices(stackId.getStackName(), stackId.getStackVersion()));
 
     if ((execCmd != null) && (execCmd.getConfigurationTags().containsKey("cluster-env"))) {
       LOG.info("AmbariManagementControllerImpl.createHostAction: created ExecutionCommand for host {}, role {}, roleCommand {}, and command ID {}, with cluster-env tags {}",

ambari-server/src/main/resources/common-services/STORM/0.9.1/package/scripts/params_linux.py

@@ -341,6 +341,8 @@ if has_ranger_admin:
 namenode_hosts = default("/clusterHostInfo/namenode_host", [])
 has_namenode = not len(namenode_hosts) == 0
 
+availableServices = config['availableServices']
+
 hdfs_user = config['configurations']['hadoop-env']['hdfs_user'] if has_namenode else None
 hdfs_user_keytab = config['configurations']['hadoop-env']['hdfs_user_keytab'] if has_namenode else None
 hdfs_principal_name = config['configurations']['hadoop-env']['hdfs_principal_name'] if has_namenode else None

ambari-server/src/main/resources/common-services/STORM/0.9.1/package/scripts/storm.py

@@ -117,10 +117,10 @@ def storm(name=None):
     # On old HDP 2.1 versions, this symlink may also exist and break EU to newer versions
     Link("/usr/lib/storm/lib/ambari-metrics-storm-sink.jar", action="delete")
 
-    if Script.get_stack_name() == "HDP" and Script.is_stack_greater_or_equal("2.5"):
-      sink_jar = params.metric_collector_sink_jar
-    else:
+    if params.availableServices.has_key("STORM") and params.availableServices.get("STORM").startswith("0"):
       sink_jar = params.metric_collector_legacy_sink_jar
+    else:
+      sink_jar = params.metric_collector_sink_jar
 
     Execute(format("{sudo} ln -s {sink_jar} {storm_lib_dir}/ambari-metrics-storm-sink.jar"),
             not_if=format("ls {storm_lib_dir}/ambari-metrics-storm-sink.jar"),

ambari-server/src/main/resources/common-services/STORM/0.9.1/package/scripts/ui_server.py

@@ -93,10 +93,10 @@ class UiServerDefault(UiServer):
     # On old HDP 2.1 versions, this symlink may also exist and break EU to newer versions
     Link("/usr/lib/storm/lib/ambari-metrics-storm-sink.jar", action="delete")
 
-    if Script.get_stack_name() == "HDP" and Script.is_stack_greater_or_equal("2.5"):
-      sink_jar = params.metric_collector_sink_jar
-    else:
+    if params.availableServices.has_key("STORM") and params.availableServices.get("STORM").startswith("0"):
       sink_jar = params.metric_collector_legacy_sink_jar
+    else:
+      sink_jar = params.metric_collector_sink_jar
 
     Execute(format("{sudo} ln -s {sink_jar} {storm_lib_dir}/ambari-metrics-storm-sink.jar"),
             not_if=format("ls {storm_lib_dir}/ambari-metrics-storm-sink.jar"),

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

@@ -51,6 +51,8 @@ import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.HostState;
 import org.apache.ambari.server.state.MaintenanceState;
 import org.apache.ambari.server.state.SecurityType;
+import org.apache.ambari.server.state.ServiceInfo;
+import org.apache.ambari.server.state.StackId;
 import org.apache.ambari.server.state.State;
 import org.apache.ambari.server.topology.TopologyManager;
 import org.apache.ambari.server.utils.StageUtils;
@@ -385,6 +387,36 @@ public class AmbariCustomCommandExecutionHelperTest {
     Assert.assertFalse(helper.isTopologyRefreshRequired("STOP", "c2", "HDFS"));
   }
 
+  @Test
+  public void testAvailableServicesMapContainsVersions() throws Exception {
+
+    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);
+
+    ambariManagementController.createAction(actionRequest, requestProperties);
+    StackId stackId = clusters.getCluster("c1").getDesiredStackVersion();
+    Map<String, ServiceInfo> services = ambariManagementController.getAmbariMetaInfo().getServices(stackId.getStackName(), stackId.getStackVersion());
+    Request request = requestCapture.getValue();
+    Stage stage = request.getStages().iterator().next();
+    List<ExecutionCommandWrapper> commands = stage.getExecutionCommands("c1-c6401");
+    ExecutionCommand command = commands.get(0).getExecutionCommand();
+    for (String service : services.keySet()) {
+      Assert.assertEquals(command.getAvailableServices().get(service), services.get(service).getVersion());
+    }
+  }
+
   private void createClusterFixture(String clusterName, String stackVersion, String hostPrefix) throws AmbariException, AuthorizationException {
     createCluster(clusterName, stackVersion);
     addHost(hostPrefix + "-c6401", clusterName);

ambari-server/src/test/python/stacks/2.1/configs/default.json

@@ -922,7 +922,10 @@
             "tag": "version1"
         }
     }, 
-    "commandId": "7-1", 
+    "commandId": "7-1",
+    "availableServices" : {
+        "STORM" : "0.9.1"
+    },
     "clusterHostInfo": {
         "ambari_server_host": [
             "c6401.ambari.apache.org"

ambari-server/src/test/python/stacks/2.1/configs/secured.json

@@ -910,7 +910,10 @@
             "tag": "version1389972945450"
         }
     }, 
-    "commandId": "9-1", 
+    "commandId": "9-1",
+    "availableServices" : {
+      "STORM" : "0.9.1"
+    },
     "clusterHostInfo": {
         "ambari_server_host": [
             "c6401.ambari.apache.org"