Handling wrong oozie spark version request

PNDA-4543

Author: dharaneeshvrd

api/src/main/resources/application_creator.py

@@ -46,6 +46,11 @@ def __init__(self, config, environment, service):
                                  environment['webhdfs_port'],
                                  'hdfs')
 
+    def assert_application_properties(self, override_properties, default_properties):
+        for component_type, component_properties in default_properties.iteritems():
+            creator = self._load_creator(component_type)
+            creator.assert_application_properties(override_properties.get(component_type, {}), component_properties)
+
     def create_application(self, package_data_path, package_metadata, application_name, property_overrides):
 
         logging.debug("create_application: %s", application_name)

api/src/main/resources/deployment_manager.py

@@ -464,6 +464,7 @@ def create_application(self, package, application, overrides, user_name):
             self._authorize(user_name, Resources.PACKAGE, package_owner, Actions.READ)
             self._authorize(user_name, Resources.APPLICATION, None, Actions.CREATE)
             defaults = self.get_package_info(package)['defaults']
+            self._application_creator.assert_application_properties(overrides, defaults)
             package_data_path = self._package_registrar.get_package_data(package)
             self._application_registrar.create_application(package, application, overrides, defaults)
             self._mark_creating(application)

api/src/main/resources/plugins/base_creator.py

@@ -106,6 +106,15 @@ def create_component(self, staged_component_path, application_name, user_name, c
         '''
         pass
 
+    def assert_application_properties(self, override_properties, default_properties):
+        '''
+        Assert application properties before creating the application
+
+        override_properties - properties overrided by user
+        default_properties - default properties present in package
+        '''
+        pass
+
     def destroy_component(self, application_name, create_data):
         '''
         Destroys component of the package of given component type

api/src/main/resources/plugins/flink.py

@@ -87,7 +87,7 @@ def create_component(self, staged_component_path, application_name, user_name, c
                 properties['component_respawn_type'] = 'no'
             else:
                 properties['component_respawn_type'] = 'always'
-        
+
         if 'component_respawn_timeout_sec' not in properties:
             if properties['component_flink_job_type'] == 'batch':
                 properties['component_respawn_timeout_sec'] = '0'

api/src/main/resources/plugins/oozie.py

@@ -34,7 +34,7 @@
 
 import deployer_utils
 from plugins.base_creator import Creator
-from exceptiondef import FailedCreation
+from exceptiondef import FailedCreation, FailedValidation
 
 class OozieCreator(Creator):
 
@@ -48,6 +48,17 @@ def validate_component(self, component):
     def get_component_type(self):
         return 'oozie'
 
+    def assert_application_properties(self, override_properties, default_properties):
+        # Assert application properties before creating the application
+        for component in default_properties:
+            properties = default_properties[component].copy()
+            properties.update(override_properties.get(component, {}))
+            if 'spark_version' in properties and properties['spark_version'] != self._config['oozie_spark_version']:
+                information = "Mismatch between cluster's oozie spark version (version = %s) and \
+requested spark version" % self._config['oozie_spark_version']
+                raise FailedValidation(json.dumps({"information": information}))
+            properties = None
+
     def destroy_component(self, application_name, create_data):
         logging.debug(
             "destroy_component: %s %s",
@@ -98,7 +109,7 @@ def create_component(self, staged_component_path, application_name, user_name, c
         # platform shared libs e.g. hbase
         properties['oozie.libpath'] = '/pnda/deployment/platform'
         # For spark2 add a special setting to select spark2
-        if 'component_spark_version' in properties and properties['component_spark_version'] == '2':
+        if self._config['oozie_spark_version'] == '2':
             properties['oozie.action.sharelib.for.spark'] = 'spark2'
 
         # insert default queue selection

api/src/main/resources/test_application_creator.py

@@ -31,7 +31,7 @@
 class ApplicationCreatorTests(unittest.TestCase):
 
     user = getpass.getuser()
-    config = {'stage_root': 'stage', 'plugins_path': 'plugins'}
+    config = {'stage_root': 'stage', 'plugins_path': 'plugins', 'oozie_spark_version': '1'}
     environment = {
         'webhdfs_host': 'webhdfshost',
         'webhdfs_port': 'webhdfsport',
@@ -239,10 +239,27 @@ class Resp(object):
             def json(self):
                 return {'id': 'someid'}
 
+        default_properties = {
+            "oozie": {
+                "componentA": {
+                    "spark_version": "1"
+                }
+            }
+        }
+        override_properties = {
+            "user": "root",
+            "oozie": {
+                "componentA": {
+                    "spark_version": "2"
+                }
+            }
+        }
+
         post_mock.return_value = Resp()
         cmd_mock.return_value = (0, 'dev')
         with patch("__builtin__.open", mock_open(read_data="[]")):
             creator = ApplicationCreator(self.config, self.environment, self.service)
+            self.assertRaises(FailedValidation, creator.assert_application_properties, override_properties, default_properties)
             self.assertRaises(FailedCreation, creator.create_application, 'abcd', self.package_metadata_2, 'aname', self.property_overrides)
 
     @patch('starbase.Connection')

api/src/main/resources/test_deployer_manager.py

@@ -75,7 +75,7 @@ def setUp(self):
             'queue_policy': 'echo dev',
             'namespace': 'mockspace'
         }
-        self.mock_config = {"deployer_thread_limit": 1, 'stage_root': 'stage', 'plugins_path': 'plugins'}
+        self.mock_config = {"deployer_thread_limit": 1, 'stage_root': 'stage', 'plugins_path': 'plugins', 'oozie_spark_version': '1'}
         # mock app registrar:
         mock_application_registar = Mock()
         application_data = {}
@@ -934,4 +934,3 @@ def expect_exception():
             deployment_manager.start_application(self.test_app_name, 'username2')
 
         self.assertRaises(Forbidden, expect_exception)
-