AMBARI-13595. Stop-and-Start Upgrade: Handle Storm in all upgrade paths (alejandro)

ambari-server/src/main/java/org/apache/ambari/server/checks/CheckDescription.java

@@ -72,6 +72,12 @@ public enum CheckDescription {
         put(AbstractCheckDescriptor.DEFAULT, "The SNameNode component must be deleted from host: {{fails}}.");
       }}),
 
+  STORM_REST_API_MUST_BE_DELETED(PrereqCheckType.SERVICE,
+      "The STORM_REST_API component will no longer be available and must be deleted from the cluster before upgrading. The same functionality is now provided by STORM_UI_SERVER. First, stop the component. Next, delete it using the API, e.g., curl -u $user:$password -X DELETE -H 'X-Requested-By:admin' http://$server:8080/api/v1/clusters/$name/services/STORM/components/STORM_REST_API ",
+      new HashMap<String, String>() {{
+        put(AbstractCheckDescriptor.DEFAULT, "The following component must be deleted from the cluster: {{fails}}.");
+      }}),
+
   SERVICES_HIVE_MULTIPLE_METASTORES(PrereqCheckType.SERVICE,
       "Hive Metastore Availability",
       new HashMap<String, String>() {{

ambari-server/src/main/java/org/apache/ambari/server/checks/StormRestAPIDeletedCheck.java

@@ -0,0 +1,108 @@
+/*
+ * 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.checks;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.ServiceComponentNotFoundException;
+import org.apache.ambari.server.ServiceNotFoundException;
+import org.apache.ambari.server.controller.PrereqCheckRequest;
+import org.apache.ambari.server.orm.dao.HostComponentStateDAO;
+import org.apache.ambari.server.orm.entities.HostComponentStateEntity;
+import org.apache.ambari.server.state.Cluster;
+import org.apache.ambari.server.state.ServiceComponent;
+import org.apache.ambari.server.state.stack.PrereqCheckStatus;
+import org.apache.ambari.server.state.stack.PrerequisiteCheck;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * The {@link StormRestAPIDeletedCheck}
+ * checks that STORM_REST_API Component is deleted when upgrading from HDP 2.1 to 2.2 or higher.
+ */
+@Singleton
+@UpgradeCheck(group = UpgradeCheckGroup.TOPOLOGY, order = 1.0f, required= false)
+public class StormRestAPIDeletedCheck extends AbstractCheckDescriptor {
+
+  @Inject
+  HostComponentStateDAO hostComponentStateDao;
+
+  /**
+   * Constructor.
+   */
+  public StormRestAPIDeletedCheck() {
+    super(CheckDescription.STORM_REST_API_MUST_BE_DELETED);
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public boolean isApplicable(PrereqCheckRequest request) throws AmbariException {
+    if (!super.isApplicable(request, Arrays.asList("STORM"), true)) {
+      return false;
+    }
+
+    return true;
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public void perform(PrerequisiteCheck prerequisiteCheck, PrereqCheckRequest request) throws AmbariException {
+    Set<String> hosts = new HashSet<String>();
+    final String STORM = "STORM";
+    final String STORM_REST_API = "STORM_REST_API";
+
+    final String clusterName = request.getClusterName();
+    final Cluster cluster = clustersProvider.get().getCluster(clusterName);
+    try {
+      ServiceComponent serviceComponent = cluster.getService(STORM).getServiceComponent(STORM_REST_API);
+      if (serviceComponent != null) {
+        hosts = serviceComponent.getServiceComponentHosts().keySet();
+      }
+    } catch (ServiceNotFoundException err) {
+      // This exception can be ignored if the component doesn't exist because it is a best-attempt at finding it.
+      ;
+    } catch (ServiceComponentNotFoundException err) {
+      // This exception can be ignored if the component doesn't exist because it is a best-attempt at finding it.
+      ;
+    }
+
+    // Try another method to find references to STORM_REST_API
+    if (hosts.isEmpty()) {
+      List<HostComponentStateEntity> allHostComponents = hostComponentStateDao.findAll();
+      for (HostComponentStateEntity hc : allHostComponents) {
+        if (hc.getServiceName().equalsIgnoreCase(STORM) && hc.getComponentName().equalsIgnoreCase(STORM_REST_API)) {
+          hosts.add(hc.getHostName());
+        }
+      }
+    }
+
+    if (!hosts.isEmpty()) {
+      prerequisiteCheck.getFailedOn().add(STORM_REST_API);
+      prerequisiteCheck.setStatus(PrereqCheckStatus.FAIL);
+      prerequisiteCheck.setFailReason(getFailReason(prerequisiteCheck, request));
+    }
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/checks/UpgradeCheckGroup.java

@@ -42,27 +42,32 @@ public enum UpgradeCheckGroup {
    */
   NAMENODE_HA(3.0f),
 
+  /**
+   * Checks for the topology of service and components.
+   */
+  TOPOLOGY(4.0f),
+
   /**
    * Checks for the state of a host or service being alive and responsive.
    */
-  LIVELINESS(4.0f),
+  LIVELINESS(5.0f),
 
   /**
    * Checks for the client retry properties to be set in clients that support
    * this.
    */
-  CLIENT_RETRY_PROPERTY(5.0f),
+  CLIENT_RETRY_PROPERTY(6.0f),
 
   /**
    * Checks for various HA components, such as multiple metastores, are
    * available.
    */
-  MULTIPLE_COMPONENT_WARNING(6.0f),
+  MULTIPLE_COMPONENT_WARNING(7.0f),
 
   /**
    * A general group for warning about configuration properties.
    */
-  CONFIGURATION_WARNING(7.0f),
+  CONFIGURATION_WARNING(8.0f),
 
   /**
    * All other checks.

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/drpc_server.py

@@ -48,7 +48,7 @@ class DrpcServer(Script):
 
     storm()
 
-  def pre_rolling_restart(self, env):
+  def pre_upgrade_restart(self, env, upgrade_type=None):
     import params
     env.set_params(params)
 
@@ -56,14 +56,14 @@ class DrpcServer(Script):
       conf_select.select(params.stack_name, "storm", params.version)
       hdp_select.select("storm-client", params.version)
 
-  def start(self, env, rolling_restart=False):
+  def start(self, env, upgrade_type=None):
     import params
     env.set_params(params)
     self.configure(env)
 
     service("drpc", action="start")
 
-  def stop(self, env, rolling_restart=False):
+  def stop(self, env, upgrade_type=None):
     import params
     env.set_params(params)
 

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/nimbus.py

@@ -37,6 +37,9 @@ from ambari_commons.os_family_impl import OsFamilyImpl
 from resource_management.core.resources.service import Service
 
 class Nimbus(Script):
+  def get_stack_to_component(self):
+    return {"HDP": "storm-nimbus"}
+
   def install(self, env):
     self.install_packages(env)
     self.configure(env)
@@ -49,10 +52,8 @@ class Nimbus(Script):
 
 @OsFamilyImpl(os_family=OsFamilyImpl.DEFAULT)
 class NimbusDefault(Nimbus):
-  def get_stack_to_component(self):
-    return {"HDP": "storm-nimbus"}
 
-  def pre_rolling_restart(self, env):
+  def pre_upgrade_restart(self, env, upgrade_type=None):
     import params
     env.set_params(params)
     if params.version and compare_versions(format_hdp_stack_version(params.version), '2.2.0.0') >= 0:
@@ -61,15 +62,15 @@ class NimbusDefault(Nimbus):
       hdp_select.select("storm-nimbus", params.version)
 
 
-  def start(self, env, rolling_restart=False):
+  def start(self, env, upgrade_type=None):
     import params
     env.set_params(params)
     self.configure(env)
-    setup_ranger_storm(rolling_upgrade=rolling_restart)
+    setup_ranger_storm(upgrade_type=upgrade_type)
     service("nimbus", action="start")
 
 
-  def stop(self, env, rolling_restart=False):
+  def stop(self, env, upgrade_type=None):
     import params
     env.set_params(params)
     service("nimbus", action="stop")

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/nimbus_prod.py

@@ -43,7 +43,7 @@ class Nimbus(Script):
 
     storm()
 
-  def pre_rolling_restart(self, env):
+  def pre_upgrade_restart(self, env, upgrade_type=None):
     import params
     env.set_params(params)
 
@@ -52,14 +52,14 @@ class Nimbus(Script):
       hdp_select.select("storm-client", params.version)
       hdp_select.select("storm-nimbus", params.version)
 
-  def start(self, env, rolling_restart=False):
+  def start(self, env, upgrade_type=None):
     import params
     env.set_params(params)
     self.configure(env)
 
     supervisord_service("nimbus", action="start")
 
-  def stop(self, env, rolling_restart=False):
+  def stop(self, env, upgrade_type=None):
     import params
     env.set_params(params)
 

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/rest_api.py

@@ -33,6 +33,11 @@ from service_check import ServiceCheck
 
 
 class StormRestApi(Script):
+  """
+  Storm REST API.
+  It was available in HDP 2.0 and 2.1.
+  In HDP 2.2, it was removed since the functionality was moved to Storm UI Server.
+  """
 
   def get_stack_to_component(self):
     return {"HDP": "storm-client"}
@@ -47,22 +52,14 @@ class StormRestApi(Script):
 
     storm()
 
-  def pre_rolling_restart(self, env):
-    import params
-    env.set_params(params)
-
-    if params.version and compare_versions(format_hdp_stack_version(params.version), '2.2.0.0') >= 0:
-      conf_select.select(params.stack_name, "storm", params.version)
-      hdp_select.select("storm-client", params.version)
-
-  def start(self, env, rolling_restart=False):
+  def start(self, env, upgrade_type=None):
     import params
     env.set_params(params)
     self.configure(env)
 
     service("rest_api", action="start")
 
-  def stop(self, env, rolling_restart=False):
+  def stop(self, env, upgrade_type=None):
     import params
     env.set_params(params)
 

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/service.py

@@ -18,6 +18,8 @@ limitations under the License.
 
 """
 
+import os
+
 from resource_management.core.resources import Execute
 from resource_management.core.resources import File
 from resource_management.core.shell import as_user
@@ -76,14 +78,15 @@ def service(name, action = 'start'):
 
   elif action == "stop":
     process_dont_exist = format("! ({no_op_test})")
-    pid = get_user_call_output.get_user_call_output(format("! test -f {pid_file} ||  cat {pid_file}"), user=params.storm_user)[1]
+    if os.path.exists(pid_file):
+      pid = get_user_call_output.get_user_call_output(format("! test -f {pid_file} ||  cat {pid_file}"), user=params.storm_user)[1]
 
-    Execute(format("{sudo} kill {pid}"),
-      not_if = process_dont_exist)
+      Execute(format("{sudo} kill {pid}"),
+        not_if = process_dont_exist)
 
-    Execute(format("{sudo} kill -9 {pid}"),
-      not_if = format(
-        "sleep 2; {process_dont_exist} || sleep 20; {process_dont_exist}"),
-      ignore_failures = True)
+      Execute(format("{sudo} kill -9 {pid}"),
+        not_if = format(
+          "sleep 2; {process_dont_exist} || sleep 20; {process_dont_exist}"),
+        ignore_failures = True)
 
-    File(pid_file, action = "delete")
+      File(pid_file, action = "delete")

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/setup_ranger_storm.py

@@ -19,7 +19,10 @@ limitations under the License.
 """
 from resource_management.core.logger import Logger
 
-def setup_ranger_storm(rolling_upgrade = False):
+def setup_ranger_storm(upgrade_type=None):
+  """
+  :param upgrade_type: Upgrade Type such as "rolling" or "nonrolling"
+  """
   import params
 
   if params.has_ranger_admin and params.security_enabled:
@@ -30,7 +33,7 @@ def setup_ranger_storm(rolling_upgrade = False):
       from resource_management.libraries.functions.setup_ranger_plugin import setup_ranger_plugin
     
     hdp_version = None
-    if rolling_upgrade:
+    if upgrade_type is not None:
       hdp_version = params.version
 
     if params.retryAble:

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

@@ -108,6 +108,13 @@ def storm(name=None):
   )
   
   if params.storm_logs_supported:
+    Directory(params.log4j_dir,
+              owner=params.storm_user,
+              group=params.user_group,
+              mode=0755,
+              recursive=True
+    )
+    
     File(format("{log4j_dir}/cluster.xml"),
       owner=params.storm_user,
       content=InlineTemplate(params.storm_cluster_log4j_content)

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/storm_upgrade.py

@@ -30,9 +30,13 @@ from resource_management.libraries.functions.default import default
 
 class StormUpgrade(Script):
   """
+  Applies to Rolling/Express Upgrade from HDP 2.1 or 2.2 to 2.3 or higher.
+
+  Requirements: Needs to run from a host with ZooKeeper Client.
+
   This class helps perform some of the upgrade tasks needed for Storm during
-  a non-rolling upgrade. Storm writes data to disk locally and to ZooKeeper.
-  If any HDP 2.2 bits exist in these directories when an HDP 2.3 instance
+  a Rolling or Express upgrade. Storm writes data to disk locally and to ZooKeeper.
+  If any HDP 2.1 or 2.2 bits exist in these directories when an HDP 2.3 instance
   starts up, it will fail to start properly. Because the upgrade framework in
   Ambari doesn't yet have a mechanism to say "stop all" before starting to
   upgrade each component, we need to rely on a Storm trick to bring down
@@ -54,16 +58,16 @@ class StormUpgrade(Script):
     if storm_zookeeper_root_dir is None:
       raise Fail("The storm ZooKeeper directory specified by storm-site/storm.zookeeper.root must be specified")
 
-    # the zookeeper client must be given a zookeeper host to contact
+    # The zookeeper client must be given a zookeeper host to contact. Guaranteed to have at least one host.
     storm_zookeeper_server_list = yaml_utils.get_values_from_yaml_array(params.storm_zookeeper_servers)
     if storm_zookeeper_server_list is None:
       Logger.info("Unable to extract ZooKeeper hosts from '{0}', assuming localhost").format(params.storm_zookeeper_servers)
       storm_zookeeper_server_list = ["localhost"]
 
-    # for every zk server, try to remove /storm
+    # For every zk server, try to remove /storm
     zookeeper_data_cleared = False
     for storm_zookeeper_server in storm_zookeeper_server_list:
-      # determine where the zkCli.sh shell script is
+      # Determine where the zkCli.sh shell script is
       zk_command_location = "/usr/hdp/current/zookeeper-client/bin/zkCli.sh"
       if params.version is not None:
         zk_command_location = "/usr/hdp/{0}/zookeeper/bin/zkCli.sh".format(params.version)
@@ -135,28 +139,27 @@ class StormUpgrade(Script):
           existing_json_map = json.load(file_pointer)
 
         if cmp(json_map, existing_json_map) == 0:
-          Logger.info("The storm upgrade has already removed the local directories for {0}-{1} for request {2} and stage {3}".format(
-            stack_name, stack_version, request_id, stage_id))
+          Logger.info("The storm upgrade has already removed the local directories for {0}-{1} for "
+                      "request {2} and stage {3}. Nothing else to do.".format(stack_name, stack_version, request_id, stage_id))
 
-          # nothing else to do here for this as it appears to have already been
+          # Nothing else to do here for this as it appears to have already been
           # removed by another component being upgraded
           return
-
-      except:
-        Logger.error("The upgrade file {0} appears to be corrupt; removing...".format(upgrade_file))
+      except Exception, e:
+        Logger.error("The upgrade file {0} appears to be corrupt; removing it. Error: {1}".format(upgrade_file, str(e)))
         File(upgrade_file, action="delete")
     else:
-      # delete the upgrade file since it does not match
-      File(upgrade_file, action="delete")
+      Logger.info('The upgrade file {0} does not exist, so will attempt to delete local Storm directory if it exists.')
 
-    # delete from local directory
-    Directory(storm_local_directory, action="delete", recursive=True)
+    # Delete from local directory
+    if os.path.isdir(storm_local_directory):
+      Directory(storm_local_directory, action="delete", recursive=True)
 
-    # recreate storm local directory
-    Directory(storm_local_directory, mode=0755, owner = params.storm_user,
-      group = params.user_group, recursive = True)
+    # Recreate storm local directory
+    Directory(storm_local_directory, mode=0755, owner=params.storm_user,
+      group=params.user_group, recursive=True)
 
-    # the file doesn't exist, so create it
+    # The file doesn't exist, so create it
     with open(upgrade_file, 'w') as file_pointer:
       json.dump(json_map, file_pointer, indent=2)
 

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/supervisor.py

@@ -34,6 +34,9 @@ from resource_management.core.resources.service import Service
 
 
 class Supervisor(Script):
+  def get_stack_to_component(self):
+    return {"HDP": "storm-supervisor"}
+
   def install(self, env):
     self.install_packages(env)
     self.configure(env)
@@ -66,10 +69,8 @@ class SupervisorWindows(Supervisor):
 
 @OsFamilyImpl(os_family=OsFamilyImpl.DEFAULT)
 class SupervisorDefault(Supervisor):
-  def get_stack_to_component(self):
-    return {"HDP": "storm-supervisor"}
 
-  def pre_rolling_restart(self, env):
+  def pre_upgrade_restart(self, env, upgrade_type=None):
     import params
     env.set_params(params)
 
@@ -78,7 +79,7 @@ class SupervisorDefault(Supervisor):
       hdp_select.select("storm-client", params.version)
       hdp_select.select("storm-supervisor", params.version)
 
-  def start(self, env, rolling_restart=False):
+  def start(self, env, upgrade_type=None):
     import params
     env.set_params(params)
     self.configure(env)
@@ -86,7 +87,7 @@ class SupervisorDefault(Supervisor):
     service("supervisor", action="start")
     service("logviewer", action="start")
 
-  def stop(self, env, rolling_restart=False):
+  def stop(self, env, upgrade_type=None):
     import params
     env.set_params(params)
 

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/supervisor_prod.py

@@ -44,7 +44,7 @@ class Supervisor(Script):
     env.set_params(params)
     storm()
 
-  def pre_rolling_restart(self, env):
+  def pre_upgrade_restart(self, env, upgrade_type=None):
     import params
     env.set_params(params)
 
@@ -53,7 +53,7 @@ class Supervisor(Script):
       hdp_select.select("storm-client", params.version)
       hdp_select.select("storm-supervisor", params.version)
 
-  def start(self, env, rolling_restart=False):
+  def start(self, env, upgrade_type=None):
     import params
     env.set_params(params)
     self.configure(env)
@@ -61,7 +61,7 @@ class Supervisor(Script):
     supervisord_service("supervisor", action="start")
     service("logviewer", action="start")
 
-  def stop(self, env, rolling_restart=False):
+  def stop(self, env, upgrade_type=None):
     import params
     env.set_params(params)
 

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

@@ -39,6 +39,10 @@ from resource_management.core.resources.service import Service
 
 
 class UiServer(Script):
+
+  def get_stack_to_component(self):
+    return {"HDP": "storm-client"}
+
   def install(self, env):
     self.install_packages(env)
     self.configure(env)
@@ -70,24 +74,22 @@ class UiServerWindows(UiServer):
 
 @OsFamilyImpl(os_family=OsFamilyImpl.DEFAULT)
 class UiServerDefault(UiServer):
-  def get_stack_to_component(self):
-    return {"HDP": "storm-client"}
 
-  def pre_rolling_restart(self, env):
+  def pre_upgrade_restart(self, env, upgrade_type=None):
     import params
     env.set_params(params)
     if params.version and compare_versions(format_hdp_stack_version(params.version), '2.2.0.0') >= 0:
       conf_select.select(params.stack_name, "storm", params.version)
       hdp_select.select("storm-client", params.version)
 
-  def start(self, env, rolling_restart=False):
+  def start(self, env, upgrade_type=None):
     import params
     env.set_params(params)
     self.configure(env)
-    setup_ranger_storm(rolling_upgrade=rolling_restart)
+    setup_ranger_storm(upgrade_type=upgrade_type)
     service("ui", action="start")
 
-  def stop(self, env, rolling_restart=False):
+  def stop(self, env, upgrade_type=None):
     import params
     env.set_params(params)
     service("ui", action="stop")

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

@@ -23,6 +23,7 @@
   <type>NON_ROLLING</type>
   <downgrade-allowed>false</downgrade-allowed>
   <prerequisite-checks>
+    <check>org.apache.ambari.server.checks.StormRestAPIDeletedCheck</check>
   </prerequisite-checks>
   <upgrade-path>
     <intermediate-stack version="2.2"/>

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

@@ -66,7 +66,6 @@
         <component>DRPC_SERVER</component>
         <component>STORM_UI_SERVER</component>
         <component>SUPERVISOR</component>
-        <component>STORM_REST_API</component>
         <component>NIMBUS</component>
       </service>
 
@@ -415,7 +414,6 @@
       <skippable>true</skippable>
       <service name="STORM">
         <component>NIMBUS</component>
-        <component>STORM_REST_API</component>
         <component>SUPERVISOR</component>
         <component>STORM_UI_SERVER</component>
         <component>DRPC_SERVER</component>
@@ -742,12 +740,6 @@
         </upgrade>
       </component>
 
-      <component name="STORM_REST_API">
-        <upgrade>
-          <task xsi:type="restart-task"/>
-        </upgrade>
-      </component>
-
       <component name="SUPERVISOR">
         <upgrade>
           <task xsi:type="restart-task"/>

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

@@ -96,7 +96,6 @@
         <component>DRPC_SERVER</component>
         <component>STORM_UI_SERVER</component>
         <component>SUPERVISOR</component>
-        <component>STORM_REST_API</component>
         <component>NIMBUS</component>
       </service>
 
@@ -665,7 +664,6 @@
       <skippable>true</skippable>
       <service name="STORM">
         <component>NIMBUS</component>
-        <component>STORM_REST_API</component>
         <component>SUPERVISOR</component>
         <component>STORM_UI_SERVER</component>
         <component>DRPC_SERVER</component>
@@ -1004,12 +1002,6 @@
         </upgrade>
       </component>
 
-      <component name="STORM_REST_API">
-        <upgrade>
-          <task xsi:type="restart-task"/>
-        </upgrade>
-      </component>
-
       <component name="SUPERVISOR">
         <upgrade>
           <task xsi:type="restart-task"/>

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

@@ -267,7 +267,6 @@
       <skippable>true</skippable>
       <service name="STORM">
         <component>NIMBUS</component>
-        <component>STORM_REST_API</component>
         <component>SUPERVISOR</component>
         <component>STORM_UI_SERVER</component>
         <component>DRPC_SERVER</component>
@@ -619,11 +618,6 @@
           <task xsi:type="restart-task"/>
         </upgrade>
       </component>
-      <component name="STORM_REST_API">
-        <upgrade>
-          <task xsi:type="restart-task"/>
-        </upgrade>
-      </component>
       <component name="SUPERVISOR">
         <upgrade>
           <task xsi:type="restart-task"/>

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

@@ -297,7 +297,6 @@
       <skippable>true</skippable>
       <service name="STORM">
         <component>NIMBUS</component>
-        <component>STORM_REST_API</component>
         <component>SUPERVISOR</component>
         <component>STORM_UI_SERVER</component>
         <component>DRPC_SERVER</component>
@@ -814,18 +813,6 @@
         </upgrade>
       </component>
 
-      <component name="STORM_REST_API">
-        <pre-upgrade>
-          <task xsi:type="execute" summary="Removing local Storm data">
-            <script>scripts/storm_upgrade.py</script>
-            <function>delete_storm_local_data</function>
-          </task>
-        </pre-upgrade>
-        <upgrade>
-          <task xsi:type="restart-task"/>
-        </upgrade>
-      </component>
-
       <component name="SUPERVISOR">
         <pre-upgrade>
           <task xsi:type="execute" summary="Removing local Storm data">

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

@@ -74,7 +74,6 @@
         <component>DRPC_SERVER</component>
         <component>STORM_UI_SERVER</component>
         <component>SUPERVISOR</component>
-        <component>STORM_REST_API</component>
         <component>NIMBUS</component>
       </service>
 
@@ -456,7 +455,6 @@
       <skippable>true</skippable>
       <service name="STORM">
         <component>NIMBUS</component>
-        <component>STORM_REST_API</component>
         <component>SUPERVISOR</component>
         <component>STORM_UI_SERVER</component>
         <component>DRPC_SERVER</component>
@@ -818,12 +816,6 @@
         </upgrade>
       </component>
 
-      <component name="STORM_REST_API">
-        <upgrade>
-          <task xsi:type="restart-task"/>
-        </upgrade>
-      </component>
-
       <component name="SUPERVISOR">
         <upgrade>
           <task xsi:type="restart-task"/>

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

@@ -300,7 +300,6 @@
       <skippable>true</skippable>
       <service name="STORM">
         <component>NIMBUS</component>
-        <component>STORM_REST_API</component>
         <component>SUPERVISOR</component>
         <component>STORM_UI_SERVER</component>
         <component>DRPC_SERVER</component>
@@ -688,11 +687,6 @@
           <task xsi:type="restart-task"/>
         </upgrade>
       </component>
-      <component name="STORM_REST_API">
-        <upgrade>
-          <task xsi:type="restart-task"/>
-        </upgrade>
-      </component>
       <component name="SUPERVISOR">
         <upgrade>
           <task xsi:type="restart-task"/>

ambari-server/src/test/python/stacks/2.1/STORM/test_storm_drpc_server.py

@@ -66,7 +66,10 @@ class TestStormDrpcServer(TestStormBase):
 
     self.assertNoMoreResources()
 
-  def test_stop_default(self):
+  @patch("os.path.exists")
+  def test_stop_default(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [True, ]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/drpc_server.py",
                        classname = "DrpcServer",
                        command = "stop",
@@ -123,7 +126,10 @@ class TestStormDrpcServer(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_stop_secured(self):
+  @patch("os.path.exists")
+  def test_stop_secured(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [True, ]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/drpc_server.py",
                        classname = "DrpcServer",
                        command = "stop",
@@ -143,17 +149,17 @@ class TestStormDrpcServer(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_pre_rolling_restart(self):
+  def test_pre_upgrade_restart(self):
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/drpc_server.py",
                      classname = "DrpcServer",
-                     command = "pre_rolling_restart",
+                     command = "pre_upgrade_restart",
                      config_file="default.json",
                      hdp_stack_version = self.STACK_VERSION,
                      target = RMFTestCase.TARGET_COMMON_SERVICES)
 
     self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-client', '2.2.1.0-2067'), sudo=True)
 
-  def test_pre_rolling_restart_23(self):
+  def test_pre_upgrade_restart_23(self):
     config_file = self.get_src_folder()+"/test/python/stacks/2.1/configs/default.json"
     with open(config_file, "r") as f:
       json_content = json.load(f)
@@ -163,7 +169,7 @@ class TestStormDrpcServer(TestStormBase):
     mocks_dict = {}
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/drpc_server.py",
                      classname = "DrpcServer",
-                     command = "pre_rolling_restart",
+                     command = "pre_upgrade_restart",
                      config_dict = json_content,
                      hdp_stack_version = self.STACK_VERSION,
                      target = RMFTestCase.TARGET_COMMON_SERVICES,

ambari-server/src/test/python/stacks/2.1/STORM/test_storm_nimbus.py

@@ -64,7 +64,10 @@ class TestStormNimbus(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_stop_default(self):
+  @patch("os.path.exists")
+  def test_stop_default(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [False, False, True]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/nimbus.py",
                        classname = "Nimbus",
                        command = "stop",
@@ -121,7 +124,10 @@ class TestStormNimbus(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_stop_secured(self):
+  @patch("os.path.exists")
+  def test_stop_secured(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [False, False, True]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/nimbus.py",
                        classname = "Nimbus",
                        command = "stop",
@@ -141,10 +147,10 @@ class TestStormNimbus(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_pre_rolling_restart(self):
+  def test_pre_upgrade_restart(self):
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/nimbus.py",
                        classname = "Nimbus",
-                       command = "pre_rolling_restart",
+                       command = "pre_upgrade_restart",
                        config_file="default.json",
                        hdp_stack_version = self.STACK_VERSION,
                        target = RMFTestCase.TARGET_COMMON_SERVICES)
@@ -152,7 +158,7 @@ class TestStormNimbus(TestStormBase):
     self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-client', '2.2.1.0-2067'), sudo=True)
     self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-nimbus', '2.2.1.0-2067'), sudo=True)
 
-  def test_pre_rolling_restart_23(self):
+  def test_pre_upgrade_restart_23(self):
     config_file = self.get_src_folder()+"/test/python/stacks/2.1/configs/default.json"
     with open(config_file, "r") as f:
       json_content = json.load(f)
@@ -162,7 +168,7 @@ class TestStormNimbus(TestStormBase):
     mocks_dict = {}
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/nimbus.py",
                      classname = "Nimbus",
-                     command = "pre_rolling_restart",
+                     command = "pre_upgrade_restart",
                      config_dict = json_content,
                      hdp_stack_version = self.STACK_VERSION,
                      target = RMFTestCase.TARGET_COMMON_SERVICES,

ambari-server/src/test/python/stacks/2.1/STORM/test_storm_nimbus_prod.py

@@ -53,7 +53,10 @@ class TestStormNimbus(TestStormBase):
 
     self.assertNoMoreResources()
 
-  def test_stop_default(self):
+  @patch("os.path.exists")
+  def test_stop_default(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [False, False, True]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/nimbus_prod.py",
                        classname = "Nimbus",
                        command = "stop",
@@ -93,7 +96,10 @@ class TestStormNimbus(TestStormBase):
 
     self.assertNoMoreResources()
 
-  def test_stop_secured(self):
+  @patch("os.path.exists")
+  def test_stop_secured(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [False, False, True]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/nimbus_prod.py",
                        classname = "Nimbus",
                        command = "stop",
@@ -106,10 +112,10 @@ class TestStormNimbus(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_pre_rolling_restart(self):
+  def test_pre_upgrade_restart(self):
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/nimbus_prod.py",
                        classname = "Nimbus",
-                       command = "pre_rolling_restart",
+                       command = "pre_upgrade_restart",
                        config_file="default.json",
                        hdp_stack_version = self.STACK_VERSION,
                        target = RMFTestCase.TARGET_COMMON_SERVICES)
@@ -117,7 +123,7 @@ class TestStormNimbus(TestStormBase):
     self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-client', '2.2.1.0-2067'), sudo=True)
     self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-nimbus', '2.2.1.0-2067'), sudo=True)
 
-  def test_pre_rolling_restart_23(self):
+  def test_pre_upgrade_restart_23(self):
     config_file = self.get_src_folder()+"/test/python/stacks/2.1/configs/default.json"
     with open(config_file, "r") as f:
       json_content = json.load(f)
@@ -127,7 +133,7 @@ class TestStormNimbus(TestStormBase):
     mocks_dict = {}
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/nimbus_prod.py",
                      classname = "Nimbus",
-                     command = "pre_rolling_restart",
+                     command = "pre_upgrade_restart",
                      config_dict = json_content,
                      hdp_stack_version = self.STACK_VERSION,
                      target = RMFTestCase.TARGET_COMMON_SERVICES,

ambari-server/src/test/python/stacks/2.1/STORM/test_storm_rest_api_service.py

@@ -65,7 +65,10 @@ class TestStormRestApi(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_stop_default(self):
+  @patch("os.path.exists")
+  def test_stop_default(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [True, ]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/rest_api.py",
                        classname = "StormRestApi",
                        command = "stop",
@@ -122,7 +125,10 @@ class TestStormRestApi(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_stop_secured(self):
+  @patch("os.path.exists")
+  def test_stop_secured(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [True, ]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/rest_api.py",
                        classname = "StormRestApi",
                        command = "stop",
@@ -140,42 +146,4 @@ class TestStormRestApi(TestStormBase):
     self.assertResourceCalled('File', '/var/run/storm/restapi.pid',
         action = ['delete'],
     )
-    self.assertNoMoreResources()
-
-  def test_pre_rolling_restart(self):
-    self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/rest_api.py",
-                       classname = "StormRestApi",
-                       command = "pre_rolling_restart",
-                       config_file="default.json",
-                       hdp_stack_version = self.STACK_VERSION,
-                       target = RMFTestCase.TARGET_COMMON_SERVICES)
-
-    self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-client', '2.2.1.0-2067'), sudo=True)
-
-  def test_pre_rolling_restart_23(self):
-    config_file = self.get_src_folder()+"/test/python/stacks/2.1/configs/default.json"
-    with open(config_file, "r") as f:
-      json_content = json.load(f)
-    version = '2.3.0.0-1234'
-    json_content['commandParams']['version'] = version
-
-    mocks_dict = {}
-    self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/rest_api.py",
-                     classname = "StormRestApi",
-                     command = "pre_rolling_restart",
-                     config_dict = json_content,
-                     hdp_stack_version = self.STACK_VERSION,
-                     target = RMFTestCase.TARGET_COMMON_SERVICES,
-                     call_mocks = [(0, None), (0, None)],
-                     mocks_dict = mocks_dict)
-
-    self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-client', '2.3.0.0-1234'), sudo=True)
-
-    self.assertEquals(1, mocks_dict['call'].call_count)
-    self.assertEquals(1, mocks_dict['checked_call'].call_count)
-    self.assertEquals(
-      ('ambari-python-wrap', '/usr/bin/conf-select', 'set-conf-dir', '--package', 'storm', '--stack-version', '2.3.0.0-1234', '--conf-version', '0'),
-       mocks_dict['checked_call'].call_args_list[0][0][0])
-    self.assertEquals(
-      ('ambari-python-wrap', '/usr/bin/conf-select', 'create-conf-dir', '--package', 'storm', '--stack-version', '2.3.0.0-1234', '--conf-version', '0'),
-       mocks_dict['call'].call_args_list[0][0][0])
+    self.assertNoMoreResources()

ambari-server/src/test/python/stacks/2.1/STORM/test_storm_supervisor.py

@@ -77,7 +77,10 @@ class TestStormSupervisor(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_stop_default(self):
+  @patch("os.path.exists")
+  def test_stop_default(self, path_exists_mock):
+    # Third bool is for the pid file
+    path_exists_mock.side_effect = [False, False, True, True]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/supervisor.py",
                        classname = "Supervisor",
                        command = "stop",
@@ -157,7 +160,10 @@ class TestStormSupervisor(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_stop_secured(self):
+  @patch("os.path.exists")
+  def test_stop_secured(self, path_exists_mock):
+    # Third bool is for the pid file
+    path_exists_mock.side_effect = [False, False, True, True]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/supervisor.py",
                        classname = "Supervisor",
                        command = "stop",
@@ -187,10 +193,10 @@ class TestStormSupervisor(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_pre_rolling_restart(self):
+  def test_pre_upgrade_restart(self):
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/supervisor.py",
                        classname = "Supervisor",
-                       command = "pre_rolling_restart",
+                       command = "pre_upgrade_restart",
                        config_file="default.json",
                        hdp_stack_version = self.STACK_VERSION,
                        target = RMFTestCase.TARGET_COMMON_SERVICES)
@@ -198,7 +204,7 @@ class TestStormSupervisor(TestStormBase):
     self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-client', '2.2.1.0-2067'), sudo=True)
     self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-supervisor', '2.2.1.0-2067'), sudo=True)
 
-  def test_pre_rolling_restart_23(self):
+  def test_pre_upgrade_restart_23(self):
     config_file = self.get_src_folder()+"/test/python/stacks/2.1/configs/default.json"
     with open(config_file, "r") as f:
       json_content = json.load(f)
@@ -208,7 +214,7 @@ class TestStormSupervisor(TestStormBase):
     mocks_dict = {}
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/supervisor.py",
                      classname = "Supervisor",
-                     command = "pre_rolling_restart",
+                     command = "pre_upgrade_restart",
                      config_dict = json_content,
                      hdp_stack_version = self.STACK_VERSION,
                      target = RMFTestCase.TARGET_COMMON_SERVICES,

ambari-server/src/test/python/stacks/2.1/STORM/test_storm_supervisor_prod.py

@@ -67,7 +67,10 @@ class TestStormSupervisor(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_stop_default(self):
+  @patch("os.path.exists")
+  def test_stop_default(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [True, ]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/supervisor_prod.py",
                        classname = "Supervisor",
                        command = "stop",
@@ -130,7 +133,10 @@ class TestStormSupervisor(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_stop_secured(self):
+  @patch("os.path.exists")
+  def test_stop_secured(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [True, ]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/supervisor_prod.py",
                        classname = "Supervisor",
                        command = "stop",
@@ -156,10 +162,10 @@ class TestStormSupervisor(TestStormBase):
 
     self.assertNoMoreResources()
 
-  def test_pre_rolling_restart(self):
+  def test_pre_upgrade_restart(self):
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/supervisor_prod.py",
                        classname = "Supervisor",
-                       command = "pre_rolling_restart",
+                       command = "pre_upgrade_restart",
                        config_file="default.json",
                        hdp_stack_version = self.STACK_VERSION,
                        target = RMFTestCase.TARGET_COMMON_SERVICES)
@@ -167,7 +173,7 @@ class TestStormSupervisor(TestStormBase):
     self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-client', '2.2.1.0-2067'), sudo=True)
     self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-supervisor', '2.2.1.0-2067'), sudo=True)
 
-  def test_pre_rolling_restart_23(self):
+  def test_pre_upgrade_restart_23(self):
     config_file = self.get_src_folder()+"/test/python/stacks/2.1/configs/default.json"
     with open(config_file, "r") as f:
       json_content = json.load(f)
@@ -177,7 +183,7 @@ class TestStormSupervisor(TestStormBase):
     mocks_dict = {}
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/supervisor_prod.py",
                      classname = "Supervisor",
-                     command = "pre_rolling_restart",
+                     command = "pre_upgrade_restart",
                      config_dict = json_content,
                      hdp_stack_version = self.STACK_VERSION,
                      target = RMFTestCase.TARGET_COMMON_SERVICES,

ambari-server/src/test/python/stacks/2.1/STORM/test_storm_ui_server.py

@@ -63,8 +63,11 @@ class TestStormUiServer(TestStormBase):
         try_sleep = 10,
     )
     self.assertNoMoreResources()
-  
-  def test_stop_default(self):
+
+  @patch("os.path.exists")
+  def test_stop_default(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [False, False, True]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/ui_server.py",
                        classname = "UiServer",
                        command = "stop",
@@ -121,7 +124,10 @@ class TestStormUiServer(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_stop_secured(self):
+  @patch("os.path.exists")
+  def test_stop_secured(self, path_exists_mock):
+    # Last bool is for the pid file
+    path_exists_mock.side_effect = [False, False, True]
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/ui_server.py",
                        classname = "UiServer",
                        command = "stop",
@@ -141,17 +147,17 @@ class TestStormUiServer(TestStormBase):
     )
     self.assertNoMoreResources()
 
-  def test_pre_rolling_restart(self):
+  def test_pre_upgrade_restart(self):
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/ui_server.py",
                        classname = "UiServer",
-                       command = "pre_rolling_restart",
+                       command = "pre_upgrade_restart",
                        config_file="default.json",
                        hdp_stack_version = self.STACK_VERSION,
                        target = RMFTestCase.TARGET_COMMON_SERVICES)
 
     self.assertResourceCalled("Execute", ('ambari-python-wrap', '/usr/bin/hdp-select', 'set', 'storm-client', '2.2.1.0-2067'), sudo=True)
 
-  def test_pre_rolling_restart_23(self):
+  def test_pre_upgrade_restart_23(self):
     config_file = self.get_src_folder()+"/test/python/stacks/2.1/configs/default.json"
     with open(config_file, "r") as f:
       json_content = json.load(f)
@@ -161,7 +167,7 @@ class TestStormUiServer(TestStormBase):
     mocks_dict = {}
     self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/ui_server.py",
                      classname = "UiServer",
-                     command = "pre_rolling_restart",
+                     command = "pre_upgrade_restart",
                      config_dict = json_content,
                      hdp_stack_version = self.STACK_VERSION,
                      target = RMFTestCase.TARGET_COMMON_SERVICES,