AMBARI-7825. Rolling Upgrades - hdfs:///apps/tez/tez.tar.gz needs to be versioned (alejandro)

ambari-common/src/main/python/resource_management/libraries/functions/dynamic_variable_interpretation.py

@@ -0,0 +1,311 @@
+#!/usr/bin/env python
+"""
+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.
+
+"""
+
+__all__ = ["copy_tarballs_to_hdfs", "interpret_dynamic_version_property"]
+import os
+import glob
+import re
+
+from resource_management.libraries.functions.default import default
+from resource_management.libraries.resources.copy_from_local import CopyFromLocal
+from resource_management.libraries.resources.execute_hadoop import ExecuteHadoop
+from resource_management.core.exceptions import Fail
+from resource_management.core.logger import Logger
+
+"""
+This file provides helper methods needed for the versioning of RPMs. Specifically, it does dynamic variable
+interpretation to replace strings like {{ hdp_stack_version }} and {{ component_version }} where the value of the
+variables cannot be determined ahead of time, but rather, depends on what files are found.
+
+It assumes that {{ hdp_stack_version }} is constructed as ${major.minor.patch.rev}-${build_number}
+E.g., 998.2.2.1.0-998
+Whereas {{ component_version }} is up to the Component to define, may be 3.0.1 or 301.
+"""
+
+# These values must be the suffix of the properties in cluster-env.xml
+TAR_SOURCE_SUFFIX = "_tar_source"
+TAR_DESTINATION_FOLDER_SUFFIX = "_tar_destination_folder"
+
+
+def __contains_dynamic_variable(string):
+  """
+  :param string: Input string to check
+  :return: Returns True if the string contains any dynamic variables to be interpreted, otherwise False.
+  """
+  return "{{ component_version }}" in string or "{{ hdp_stack_version }}" in string
+
+
+def __get_tar_source_and_dest_folder(tarball_prefix):
+  """
+  :param tarball_prefix: Prefix of the tarball must be one of tez, hive, mr, pig
+  :return: Returns a tuple of (x, y) after verifying the properties
+  """
+  component_tar_source_file = default("/configurations/cluster-env/%s%s" % (tarball_prefix.lower(), TAR_SOURCE_SUFFIX), None)
+  # E.g., /usr/hdp/current/hadoop-client/tez-{{ component_version }}.{{ hdp_stack_version }}.tar.gz
+
+  component_tar_destination_folder = default("/configurations/cluster-env/%s%s" % (tarball_prefix.lower(), TAR_DESTINATION_FOLDER_SUFFIX), None)
+  # E.g., hdfs:///hdp/apps/{{ hdp_stack_version }}/mapreduce/
+
+  if not component_tar_source_file or not component_tar_destination_folder:
+    Logger.warning("Did not find %s tar source file and destination folder properties in cluster-env.xml" %
+                   tarball_prefix)
+    return None, None
+
+  if component_tar_source_file.find("/") == -1:
+    Logger.warning("The tar file path %s is not valid" % str(component_tar_source_file))
+    return None, None
+
+  if not component_tar_destination_folder.endswith("/"):
+    component_tar_destination_folder = component_tar_destination_folder + "/"
+
+  if not component_tar_destination_folder.startswith("hdfs://"):
+    return None, None
+
+  return component_tar_source_file, component_tar_destination_folder
+
+
+def __create_regex_pattern(file_path, rpm_version):
+  """
+  :param file_path: Input file path
+  :param rpm_version: HDP rpm version, such as 2.2.0.0
+  :return: Returns an expression that uses file system regex that can be used with ls and hadoop fs -ls
+  """
+  # Perform the variable interpretation
+  file_path_pattern = file_path
+  if "{{ component_version }}" in file_path_pattern:
+    file_path_pattern = file_path_pattern.replace("{{ component_version }}", "*")
+
+  # IMPORTANT, the build version was used in HDP 2.2, but may not be needed in future versions.
+  if "{{ hdp_stack_version }}" in file_path_pattern:
+    file_path_pattern = file_path_pattern.replace("{{ hdp_stack_version }}", rpm_version + "-*")   # the trailing "-*" is the build number
+  return file_path_pattern
+
+
+def __populate_source_and_dests(tarball_prefix, source_file_pattern, component_tar_destination_folder, rpm_version):
+  """
+  :param tarball_prefix: Prefix of the tarball must be one of tez, hive, mr, pig
+  :param source_file_pattern: Regex pattern of the source file from the local file system
+  :param component_tar_destination_folder: Destination folder to copy the file to in HDFS
+  :param rpm_version: Stack version number without the build version. E.g., 2.2.0.0
+  :return: Returns a list of tuples (x, y), where x is the source file in the local file system,
+  and y is the destination file path in HDFS
+  """
+  source_and_dest_pairs = []
+
+  for file in glob.glob(source_file_pattern):
+    file_base_name = os.path.basename(file)
+    component_version = None
+    hdp_build_version = None
+
+    # Attempt to retrieve the hdp_build_version and component_version.
+    # In case the build number has dots, attempt to match as many as possible.
+    pattern = "%s-(.*)\\.%s-([0-9\\.]*)\\..*" % (tarball_prefix, str(rpm_version).replace(".", "\\."))
+    m = re.search(pattern, file_base_name)
+    if m and len(m.groups()) == 2:
+      component_version = str(m.group(1))
+      hdp_build_version = str(m.group(2))
+
+    missing_a_variable = False
+    # The destination_file_path will be interpreted as well.
+    destination_file_path = os.path.join(component_tar_destination_folder, file_base_name)
+
+    if "{{ component_version }}" in destination_file_path:
+      if component_version:
+        destination_file_path = destination_file_path.replace("{{ component_version }}", component_version)
+      else:
+        missing_a_variable = True
+
+    if "{{ hdp_stack_version }}" in destination_file_path:
+      if hdp_build_version :
+        destination_file_path = destination_file_path.replace("{{ hdp_stack_version }}", "%s-%s" % (rpm_version, hdp_build_version))
+      else:
+        missing_a_variable = True
+
+    if missing_a_variable:
+      print("WARNING. Could not identify HDP stack version or Component version in file %s , "
+            "so will not copy to HDFS." % str(file))
+    else:
+      source_and_dest_pairs.append((file, destination_file_path))
+  return source_and_dest_pairs
+
+
+def __copy_files(source_and_dest_pairs, file_owner, kinit_if_needed):
+  """
+  :param source_and_dest_pairs: List of tuples (x, y), where x is the source file in the local file system,
+  and y is the destination file path in HDFS
+  :param file_owner: Owner to set for the file copied to HDFS
+  :param kinit_if_needed: kinit command if it is needed, otherwise an empty string
+  :return: Returns 0 if at least one file was copied and no exceptions occurred, and 1 otherwise.
+
+  Must kinit before calling this function.
+  """
+  import params
+
+  return_value = 1
+  if source_and_dest_pairs and len(source_and_dest_pairs) > 0:
+    return_value = 0
+    for (source, destination) in source_and_dest_pairs:
+      try:
+        destination_dir = os.path.dirname(destination)
+        create_dir_cmd = "dfs -mkdir -p %s" % destination_dir
+        test_dir_exists = "dfs -test -e %s" % destination_dir
+
+        params.HdfsDirectory(destination_dir,
+                             action="create",
+                             owner=file_owner,
+                             mode=0777
+        )
+
+        CopyFromLocal(source,
+                      mode=0755,
+                      owner=file_owner,
+                      dest_dir=destination_dir,
+                      kinnit_if_needed=kinit_if_needed,
+                      hdfs_user=params.hdfs_user,
+                      hadoop_bin_dir=params.hadoop_bin_dir,
+                      hadoop_conf_dir=params.hadoop_conf_dir
+        )
+      except:
+        return_value = 1
+  return return_value
+
+
+def copy_tarballs_to_hdfs(tarball_prefix, component_user, file_owner):
+  """
+  :param tarball_prefix: Prefix of the tarball must be one of tez, hive, mr, pig
+  :param component_user: User that will execute the Hadoop commands
+  :param file_owner: Owner of the files copied to HDFS
+  :return: Returns 0 on success, 1 if no files were copied, and in some cases may raise an exception.
+
+  In order to call this function, params.py must have all of the following,
+  rpm_version, kinit_path_local, security_enabled, hdfs_user, hdfs_principal_name, hdfs_user_keytab,
+  hadoop_bin_dir, hadoop_conf_dir, and HdfsDirectory as a partial function.
+  """
+  import params
+
+  if not hasattr(params, "rpm_version") or params.rpm_version is None:
+    Logger.warning("cluster-env.xml does not have rpm_version")
+    return 1
+
+  component_tar_source_file, component_tar_destination_folder = __get_tar_source_and_dest_folder(tarball_prefix)
+  if not component_tar_source_file or not component_tar_destination_folder:
+    return 1
+
+  source_file_pattern = __create_regex_pattern(component_tar_source_file, params.rpm_version)
+  # This is just the last segment
+  file_name_pattern = source_file_pattern.split('/')[-1:][0]
+  tar_destination_folder_pattern = __create_regex_pattern(component_tar_destination_folder, params.rpm_version)
+
+  # Pattern for searching the file in HDFS. E.g. value, hdfs:///hdp/apps/2.2.0.0-*/tez/tez-*.2.2.0.0-*.tar.gz
+  hdfs_file_pattern = os.path.join(tar_destination_folder_pattern, file_name_pattern)
+  does_hdfs_file_exist_cmd = "fs -ls %s" % hdfs_file_pattern
+
+  kinit_if_needed = ""
+  if params.security_enabled:
+    kinit_if_needed = format("{kinit_path_local} -kt {hdfs_user_keytab} {hdfs_principal_name};")
+
+  if kinit_if_needed:
+    Execute(kinit_if_needed,
+            user=component_user,
+            path='/bin'
+    )
+
+  does_hdfs_file_exist = False
+  try:
+    ExecuteHadoop(does_hdfs_file_exist_cmd,
+                  user=component_user,
+                  logoutput=True,
+                  conf_dir=params.hadoop_conf_dir,
+                  bin_dir=params.hadoop_bin_dir
+    )
+    does_hdfs_file_exist = True
+  except Fail:
+    pass
+
+  if not does_hdfs_file_exist:
+    source_and_dest_pairs = __populate_source_and_dests(tarball_prefix, source_file_pattern,
+                                                        component_tar_destination_folder, params.rpm_version)
+    return __copy_files(source_and_dest_pairs, file_owner, kinit_if_needed)
+  return 1
+
+
+def __map_local_file_to_hdfs_file(tarball_prefix):
+  """
+  :param tarball_prefix: Prefix of the tarball must be one of tez, hive, mr, pig
+  :return: Using the source tarball file pattern, it finds the corresponding file in the local filesystem, and
+  maps it to its corresponding location in HDFS, while substituting the dynamic variables like {{ hdp_stack_version }}
+  and {{ component_version }}
+  """
+  import params
+
+  if not hasattr(params, "rpm_version") or params.rpm_version is None:
+    Logger.warning("cluster-env.xml does not have rpm_version")
+    return 1
+
+  component_tar_source_file, component_tar_destination_folder = __get_tar_source_and_dest_folder(tarball_prefix)
+  if not component_tar_source_file or not component_tar_destination_folder:
+    return 1
+
+  source_file_pattern = __create_regex_pattern(component_tar_source_file, params.rpm_version)
+  source_and_dest_pairs = __populate_source_and_dests(tarball_prefix, source_file_pattern, component_tar_destination_folder, params.rpm_version)
+  if source_and_dest_pairs and len(source_and_dest_pairs) == 1:
+    return source_and_dest_pairs[0][1]
+
+  return ""
+
+
+def interpret_dynamic_version_property(property_value, tarball_prefix, delimiter=","):
+  """
+  :param property_value: Value to scan for dynamic variables
+  :param tarball_prefix:  Prefix of the tarball must be one of tez, hive, mr, pig
+  :param delimiter: Delimiter character used in the property value, typically a comma or colon
+  :return: Returns a tuple of (x, y), where x is a bool indicating if at least one variable was substituted, and y
+  is the interpretation of the property value if an interpretation could be done, otherwise it remains unchanged.
+
+  Notice that params must have the rpm_version attribute.
+  """
+  import params
+
+  found_at_least_one_replacement = False
+  versioned_tarball = __map_local_file_to_hdfs_file(tarball_prefix)
+  if versioned_tarball and versioned_tarball != "":
+    # We expect to find a file in HDFS, and must substitute it for its regex equivalent in the property inside *-site.xml
+    property_value = "" if property_value is None or property_value.strip() == "" else property_value
+
+    if property_value:
+      elements = []
+
+      for elem in property_value.split(delimiter):
+        elem = elem.strip()
+        if __contains_dynamic_variable(elem):
+          # Need to do dynamic interpretation, and slight regex escaping. Must not escape " " since it is used in
+          # the dynamic variable string.
+          elem_pattern = __create_regex_pattern(elem, params.rpm_version).replace(".", "\\.").replace("*", ".*")
+          p = re.compile(elem_pattern)
+          m = p.match(versioned_tarball)
+          if m:
+            elements.append(versioned_tarball)
+            found_at_least_one_replacement = True
+        else:
+          elements.append(elem)
+
+      if found_at_least_one_replacement:
+        property_value = ",".join(elements)
+  return found_at_least_one_replacement, property_value

ambari-common/src/main/python/resource_management/libraries/script/config_dictionary.py

@@ -24,18 +24,22 @@ class ConfigDictionary(dict):
   Immutable config dictionary
   """
   
-  def __init__(self, dictionary):
+  def __init__(self, dictionary, allow_overwrite=False):
     """
     Recursively turn dict to ConfigDictionary
     """
+    self.__allow_overwrite = allow_overwrite
     for k, v in dictionary.iteritems():
       if isinstance(v, dict):
-        dictionary[k] = ConfigDictionary(v)
+        dictionary[k] = ConfigDictionary(v, allow_overwrite=allow_overwrite)
         
     super(ConfigDictionary, self).__init__(dictionary)
 
   def __setitem__(self, name, value):
-    raise Fail("Configuration dictionary is immutable!")
+    if self.__allow_overwrite:
+      super(ConfigDictionary, self).__setitem__(name, value)
+    else:
+      raise Fail("Configuration dictionary is immutable!")
 
   def __getitem__(self, name):
     """
@@ -63,7 +67,22 @@ class ConfigDictionary(dict):
     
     return value
   
-  
+
+class MutableConfigDictionary(ConfigDictionary):
+  """
+  Mutable Configuration Dictionary
+  """
+  def __init__(self, dictionary):
+    d = dict()
+    for k, v in dictionary.iteritems():
+      if isinstance(v, dict):
+        d[k] = MutableConfigDictionary(v)
+      else:
+        d[k] = v
+
+    super(MutableConfigDictionary, self).__init__(d, allow_overwrite=True)
+
+
 class UnknownConfiguration():
   """
   Lazy failing for unknown configs.

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HIVE/package/scripts/hive_server.py

@@ -21,7 +21,7 @@ limitations under the License.
 from resource_management import *
 from hive import hive
 from hive_service import hive_service
-from install_jars import install_tez_jars
+from resource_management.libraries.functions.dynamic_variable_interpretation import copy_tarballs_to_hdfs
 
 class HiveServer(Script):
 
@@ -39,8 +39,9 @@ class HiveServer(Script):
     import params
     env.set_params(params)
     self.configure(env) # FOR SECURITY
-    
-    install_tez_jars() # Put tez jars in hdfs
+
+    # This function is needed in HDP 2.2, but it is safe to call in earlier versions.
+    copy_tarballs_to_hdfs('tez', params.tez_user, params.hdfs_user)
 
     hive_service( 'hiveserver2',
                   action = 'start'

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HIVE/package/scripts/install_jars.py

@@ -1,108 +0,0 @@
-#!/usr/bin/env python
-"""
-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.
-
-"""
-
-from resource_management import *
-import os
-import fnmatch
-
-def install_tez_jars():
-  import params
-  
-  destination_hdfs_dirs = get_tez_hdfs_dir_paths(params.tez_lib_uris)
-
-  # If tez libraries are to be stored in hdfs
-  if destination_hdfs_dirs:
-    for hdfs_dir in destination_hdfs_dirs:
-      params.HdfsDirectory(hdfs_dir,
-                          action="create_delayed",
-                          owner=params.tez_user,
-                          mode=0755
-      )
-    pass
-    params.HdfsDirectory(None, action="create")
-
-    if params.security_enabled:
-      kinit_if_needed = format("{kinit_path_local} -kt {hdfs_user_keytab} {hdfs_principal_name};")
-    else:
-      kinit_if_needed = ""
-
-    if kinit_if_needed:
-      Execute(kinit_if_needed,
-              user=params.tez_user,
-              path='/bin'
-      )
-    pass
-
-    app_dir_path = None
-    lib_dir_path = None
-
-    if len(destination_hdfs_dirs) > 0:
-      for path in destination_hdfs_dirs:
-        if 'lib' in path:
-          lib_dir_path = path
-        else:
-          app_dir_path = path
-        pass
-      pass
-    pass
-
-    if app_dir_path:
-      for scr_file, dest_file in params.app_dir_files.iteritems():
-        CopyFromLocal(scr_file,
-                      mode=0755,
-                      owner=params.tez_user,
-                      dest_dir=app_dir_path,
-                      dest_file=dest_file,
-                      kinnit_if_needed=kinit_if_needed,
-                      hdfs_user=params.hdfs_user,
-                      hadoop_bin_dir=params.hadoop_bin_dir,
-                      hadoop_conf_dir=params.hadoop_conf_dir
-        )
-
-    if lib_dir_path:
-      CopyFromLocal(params.tez_local_lib_jars,
-                    mode=0755,
-                    owner=params.tez_user,
-                    dest_dir=lib_dir_path,
-                    kinnit_if_needed=kinit_if_needed,
-                    hdfs_user=params.hdfs_user,
-                    hadoop_bin_dir=params.hadoop_bin_dir,
-                    hadoop_conf_dir=params.hadoop_conf_dir
-      )
-    pass
-
-
-def get_tez_hdfs_dir_paths(tez_lib_uris = None):
-  hdfs_path_prefix = 'hdfs://'
-  lib_dir_paths = []
-  if tez_lib_uris and tez_lib_uris.strip().find(hdfs_path_prefix, 0) != -1:
-    dir_paths = tez_lib_uris.split(',')
-    for path in dir_paths:
-      if not "tez.tar.gz" in path:
-        lib_dir_path = path.replace(hdfs_path_prefix, '')
-        lib_dir_path = lib_dir_path if lib_dir_path.endswith(os.sep) else lib_dir_path + os.sep
-        lib_dir_paths.append(lib_dir_path)
-      else:
-        lib_dir_path = path.replace(hdfs_path_prefix, '')
-        lib_dir_paths.append(os.path.dirname(lib_dir_path))
-    pass
-  pass
-
-  return lib_dir_paths

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HIVE/package/scripts/params.py

@@ -31,16 +31,15 @@ rpm_version = default("/configurations/cluster-env/rpm_version", None)
 
 hdp_stack_version = config['hostLevelParams']['stack_version']
 
-#hadoop params
+# Hadoop params
+# TODO, this logic assumes that the existence of rpm_version => HDP version is 2.2 or greater.
+# Instead, it should initialize these parameters in a file inside the HDP 2.2 stack.
 if rpm_version:
   hadoop_bin_dir = "/usr/hdp/current/hadoop-client/bin"
   hadoop_home = '/usr/hdp/current/hadoop-client'
   hadoop_streeming_jars = "/usr/hdp/current/hadoop-mapreduce-client/hadoop-streaming-*.jar"
   hive_bin = '/usr/hdp/current/hive-client/bin'
   hive_lib = '/usr/hdp/current/hive-client/lib'
-  tez_local_api_jars = '/usr/hdp/current/tez-client/tez*.jar'
-  tez_local_lib_jars = '/usr/hdp/current/tez-client/lib/*.jar'
-  tez_tar_file = "/usr/hdp/current/tez-client/lib/tez*.tar.gz"
   pig_tar_file = '/usr/hdp/current/pig-client/pig.tar.gz'
   hive_tar_file = '/usr/hdp/current/hive-client/hive.tar.gz'
   sqoop_tar_file = '/usr/hdp/current/sqoop-client/sqoop*.tar.gz'
@@ -54,9 +53,6 @@ else:
   hadoop_streeming_jars = '/usr/lib/hadoop-mapreduce/hadoop-streaming-*.jar'
   hive_bin = '/usr/lib/hive/bin'
   hive_lib = '/usr/lib/hive/lib/'
-  tez_local_api_jars = '/usr/lib/tez/tez*.jar'
-  tez_local_lib_jars = '/usr/lib/tez/lib/*.jar'
-  tez_tar_file = "/usr/lib/tez/tez*.tar.gz"
   pig_tar_file = '/usr/share/HDP-webhcat/pig.tar.gz'
   hive_tar_file = '/usr/share/HDP-webhcat/hive.tar.gz'
   sqoop_tar_file = '/usr/share/HDP-webhcat/sqoop*.tar.gz'
@@ -217,8 +213,7 @@ hostname = config["hostname"]
 hdfs_user_keytab = config['configurations']['hadoop-env']['hdfs_user_keytab']
 hdfs_principal_name = config['configurations']['hadoop-env']['hdfs_principal_name']
 
-# Tez libraries
-tez_lib_uris = default("/configurations/tez-site/tez.lib.uris", None)
+# Tez-related properties
 tez_user = config['configurations']['tez-env']['tez_user']
 
 if System.get_instance().os_family == "ubuntu":
@@ -262,14 +257,6 @@ webhcat_hdfs_user_mode = 0755
 #for create_hdfs_directory
 security_param = "true" if security_enabled else "false"
 
-if str(hdp_stack_version).startswith('2.0') or str(hdp_stack_version).startswith('2.1'):
-  app_dir_files = {tez_local_api_jars:None}
-else:
-  app_dir_files = {
-              tez_local_api_jars:None,
-              tez_tar_file:"tez.tar.gz"
-  }
-
 import functools
 #create partial functions with common arguments for every HdfsDirectory call
 #to create hdfs directory we need to call params.HdfsDirectory in code

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HIVE/package/scripts/webhcat.py

@@ -87,6 +87,10 @@ def webhcat():
             path='/bin'
     )
 
+  # TODO, fix this in AMBARI-7842.
+  # The source of these tarballs will no longer be deterministic, since the build version is not known until HDP 2.2
+  # is deployed.
+  # Also, this logic should be specific to HDP 2.2 stack.
   CopyFromLocal(params.hadoop_streeming_jars,
                 owner=params.webhcat_user,
                 mode=0755,

ambari-server/src/main/resources/stacks/HDP/2.1/services/TEZ/package/scripts/params.py

@@ -23,6 +23,20 @@ from resource_management import *
 # server configurations
 config = Script.get_config()
 
+# RPM versioning support
+rpm_version = default("/configurations/cluster-env/rpm_version", None)
+if rpm_version:
+  hadoop_bin_dir = "/usr/hdp/current/hadoop-client/bin"
+else:
+  hadoop_bin_dir = "/usr/bin"
+hadoop_conf_dir = "/etc/hadoop/conf"
+
+kinit_path_local = functions.get_kinit_path(["/usr/bin", "/usr/kerberos/bin", "/usr/sbin"])
+security_enabled = config['configurations']['cluster-env']['security_enabled']
+hdfs_user = config['configurations']['hadoop-env']['hdfs_user']
+hdfs_principal_name = config['configurations']['hadoop-env']['hdfs_principal_name']
+hdfs_user_keytab = config['configurations']['hadoop-env']['hdfs_user_keytab']
+
 config_dir = "/etc/tez/conf"
 
 hadoop_home = '/usr'
@@ -30,4 +44,17 @@ java64_home = config['hostLevelParams']['java_home']
 
 tez_user = config['configurations']['tez-env']['tez_user']
 user_group = config['configurations']['cluster-env']['user_group']
-tez_env_sh_template = config['configurations']['tez-env']['content']
+tez_env_sh_template = config['configurations']['tez-env']['content']
+
+import functools
+# Create partial functions with common arguments for every HdfsDirectory call
+# to create hdfs directory we need to call params.HdfsDirectory in code
+HdfsDirectory = functools.partial(
+  HdfsDirectory,
+  conf_dir=hadoop_conf_dir,
+  hdfs_user=hdfs_principal_name if security_enabled else hdfs_user,
+  security_enabled=security_enabled,
+  keytab=hdfs_user_keytab,
+  kinit_path_local=kinit_path_local,
+  bin_dir=hadoop_bin_dir
+)

ambari-server/src/main/resources/stacks/HDP/2.1/services/TEZ/package/scripts/tez.py

@@ -20,30 +20,39 @@ Ambari Agent
 """
 
 from resource_management import *
+from resource_management.libraries.script.config_dictionary import MutableConfigDictionary
+from resource_management.libraries.functions.dynamic_variable_interpretation import interpret_dynamic_version_property
+
 
 def tez():
   import params
 
   Directory(params.config_dir,
-    owner = params.tez_user,
-    group = params.user_group,
-    recursive = True
+            owner = params.tez_user,
+            group = params.user_group,
+            recursive = True
   )
 
-  # TODO, tez-site.xml needs to use versioned tez.tar.gz file to allow currently running jobs to finish during a rolling upgrade.
+  mutable_configs = MutableConfigDictionary(params.config)
+  tez_lib_uris = params.config['configurations']['tez-site']['tez.lib.uris']
+  if tez_lib_uris:
+    found_at_least_one_replacement, new_tez_lib_uris = interpret_dynamic_version_property(tez_lib_uris, "tez", ",")
+    if found_at_least_one_replacement:
+      mutable_configs['configurations']['tez-site']['tez.lib.uris'] = new_tez_lib_uris
+
   XmlConfig( "tez-site.xml",
-            conf_dir = params.config_dir,
-            configurations = params.config['configurations']['tez-site'],
-            configuration_attributes=params.config['configuration_attributes']['tez-site'],
-            owner = params.tez_user,
-            group = params.user_group,
-            mode = 0664
+             conf_dir = params.config_dir,
+             configurations = mutable_configs['configurations']['tez-site'],
+             configuration_attributes=params.config['configuration_attributes']['tez-site'],
+             owner = params.tez_user,
+             group = params.user_group,
+             mode = 0664
   )
-  
+
   File(format("{config_dir}/tez-env.sh"),
-    owner=params.tez_user,
-    content=InlineTemplate(params.tez_env_sh_template)
-  )  
+       owner=params.tez_user,
+       content=InlineTemplate(params.tez_env_sh_template)
+  )
 
 
 def tez_TemplateConfig(name):
@@ -54,6 +63,6 @@ def tez_TemplateConfig(name):
 
   for x in name:
     TemplateConfig(format("{config_dir}/{x}"),
-        owner = params.tez_user
+                   owner = params.tez_user
     )
 

ambari-server/src/main/resources/stacks/HDP/2.2/configuration/cluster-env.xml

@@ -26,4 +26,24 @@
     <value>2.2.0.0</value>
     <description>Hadoop RPM version</description>
   </property>
+
+  <!-- The properties that end in tar_source describe the pattern of where the tar.gz files come from.
+  They will replace {{ hdp_stack_version }} with the rpm_version value followed by -* (which is the build number in HDP 2.2),
+  and treat {{ component_version }} as a wildcard.
+  When copying those tarballs, Ambari will look up the corresponding tar_destination_folder property to know where it
+  should be copied to.
+  All of the destination folders must begin with hdfs://
+  Please note that the spaces inside of {{ ... }} are important.
+  -->
+  <property>
+    <name>tez_tar_source</name>
+    <value>/usr/hdp/current/tez-client/lib/tez-{{ component_version }}.{{ hdp_stack_version }}.tar.gz</value>
+    <description></description>
+  </property>
+  <property>
+    <name>tez_tar_destination_folder</name>
+    <value>hdfs:///hdp/apps/{{ hdp_stack_version }}/tez/</value>
+    <description></description>
+  </property>
+
 </configuration>

ambari-server/src/main/resources/stacks/HDP/2.2/services/TEZ/configuration/tez-site.xml

@@ -21,7 +21,7 @@
 
   <property>
     <name>tez.lib.uris</name>
-    <value>hdfs:///apps/tez/tez.tar.gz</value>  <!-- TODO, this needs to be a versioned file to allow currently running jobs to finish during a rolling upgrade. -->
+    <value>hdfs:///hdp/apps/{{ hdp_stack_version }}/tez/tez-{{ component_version }}.{{ hdp_stack_version }}.tar.gz</value>
     <description>Comma-delimited list of the location of the Tez libraries which will be localized for DAGs.
       Specifying a single .tar.gz or .tgz assumes that a compressed version of the tez libs is being used. This is uncompressed into a tezlibs directory when running containers, and tezlibs/;tezlibs/lib/ are added to the classpath (after . and .*).
       If multiple files are specified - files are localized as regular files, contents of directories are localized as regular files (non-recursive).

ambari-server/src/test/python/stacks/2.0.6/HIVE/test_hive_server.py

@@ -47,59 +47,6 @@ class TestHiveServer(RMFTestCase):
     )
 
     self.assert_configure_default()
-    self.assertResourceCalled('HdfsDirectory', '/apps/tez/',
-                              action = ['create_delayed'],
-                              mode = 0755,
-                              owner = 'tez',
-                              security_enabled = False,
-                              keytab = UnknownConfigurationMock(),
-                              conf_dir = '/etc/hadoop/conf',
-                              hdfs_user = 'hdfs',
-                              bin_dir = '/usr/bin',
-                              kinit_path_local = "/usr/bin/kinit"
-    )
-
-    self.assertResourceCalled('HdfsDirectory', '/apps/tez/lib/',
-                              action = ['create_delayed'],
-                              mode = 0755,
-                              owner = 'tez',
-                              security_enabled = False,
-                              keytab = UnknownConfigurationMock(),
-                              conf_dir = '/etc/hadoop/conf',
-                              hdfs_user = 'hdfs',
-                              bin_dir = '/usr/bin',
-                              kinit_path_local = "/usr/bin/kinit"
-    )
-    self.assertResourceCalled('HdfsDirectory', None,
-                              security_enabled = False,
-                              keytab = UnknownConfigurationMock(),
-                              conf_dir = '/etc/hadoop/conf',
-                              hdfs_user = 'hdfs',
-                              kinit_path_local = '/usr/bin/kinit',
-                              bin_dir = '/usr/bin',
-                              action = ['create']
-                              )
-
-    self.assertResourceCalled('CopyFromLocal', '/usr/lib/tez/tez*.jar',
-                              mode=0755,
-                              owner='tez',
-                              dest_dir='/apps/tez/',
-                              kinnit_if_needed='',
-                              hadoop_conf_dir='/etc/hadoop/conf',
-                              hadoop_bin_dir='/usr/bin',
-                              hdfs_user='hdfs',
-                              dest_file=None
-    )
-
-    self.assertResourceCalled('CopyFromLocal', '/usr/lib/tez/lib/*.jar',
-                              mode=0755,
-                              owner='tez',
-                              dest_dir='/apps/tez/lib/',
-                              kinnit_if_needed='',
-                              hadoop_bin_dir='/usr/bin',
-                              hadoop_conf_dir='/etc/hadoop/conf',
-                              hdfs_user='hdfs'
-    )
 
     self.assertResourceCalled('Execute', 'env JAVA_HOME=/usr/jdk64/jdk1.7.0_45 /tmp/start_hiveserver2_script /var/log/hive/hive-server2.out /var/log/hive/hive-server2.log /var/run/hive/hive-server.pid /etc/hive/conf.server /var/log/hive',
                               not_if = 'ls /var/run/hive/hive-server.pid >/dev/null 2>&1 && ps `cat /var/run/hive/hive-server.pid` >/dev/null 2>&1',

ambari-server/src/test/python/stacks/2.2/SLIDER/test_slider_client.py

@@ -79,7 +79,7 @@ class TestSliderClient(RMFTestCase):
                        config_file="default.json"
     )
 
-    self.assertResourceCalled('Execute', ' /usr/lib/slider/bin/slider list',
+    self.assertResourceCalled('Execute', ' /usr/hdp/current/slider-client/bin/slider list',
                               logoutput=True,
                               tries=3,
                               user='ambari-qa',

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

@@ -45,6 +45,7 @@
             "yarn.resourcemanager.scheduler.address": "c6401.ambari.apache.org:8030"
         },
         "cluster-env": {
+            "rpm_version": "2.2.0.0",
             "security_enabled": "false",
             "ignore_groupsusers_create": "false",
             "smokeuser": "ambari-qa",