Flink-stop added to stop the flink applications

Flink history server requires applications to be cancelled using "flink stop" or "flink cancel" command,
then only applications are getting added to flink history server.

PNDA-4796

Author: dharaneeshvrd

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

@@ -0,0 +1,29 @@
+#!/usr/bin/env python
+
+import commands
+import json
+import requests
+
+COMMAND_OUTPUT = commands.getoutput('yarn application -list')
+
+IS_RUNNING = False
+
+for line in COMMAND_OUTPUT.splitlines():
+    fields = line.split('\t')
+    if len(fields) >= 6:
+        app = fields[1].strip()
+        state = fields[5].strip()
+        if app == '${component_job_name}':
+            IS_RUNNING = True
+            yarn_app_id = fields[0].strip()
+            tracking_url = fields[8].strip()
+            break
+if IS_RUNNING:
+    URL = '%s/jobs' % tracking_url
+    FLINK_JOB_LIST = requests.get(URL)
+    FLINK_JOB_LIST = json.loads(FLINK_JOB_LIST.text)
+    FLINK_JOBID = FLINK_JOB_LIST['jobs-running'][0]
+    STATUS, OUT = commands.getstatusoutput('flink stop %s -yid %s' % (FLINK_JOBID, yarn_app_id))
+    if STATUS != 0:
+        commands.getoutput('flink cancel %s -yid %s' % (FLINK_JOBID, yarn_app_id))
+    commands.getoutput('yarn application -kill %s' % yarn_app_id)

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

@@ -79,7 +79,7 @@ def create_component(self, staged_component_path, application_name, user_name, c
             raise Exception('properties.json must contain "main_jar or main_py" for %s flink %s' % (application_name, component['component_name']))
 
         this_dir = os.path.dirname(os.path.realpath(__file__))
-        copy(os.path.join(this_dir, 'yarn-kill.py'), staged_component_path)
+        copy(os.path.join(this_dir, 'flink-stop.py'), staged_component_path)
         service_script = 'flink.systemd.service.tpl' if java_app else 'flink.systemd.service.py.tpl'
         service_script_install_path = '/usr/lib/systemd/system/%s.service' % service_name
         if 'component_respawn_type' not in properties:
@@ -98,7 +98,7 @@ def create_component(self, staged_component_path, application_name, user_name, c
 
         self._fill_properties(os.path.join(staged_component_path, service_script), properties)
         self._fill_properties(os.path.join(staged_component_path, 'application.properties'), properties)
-        self._fill_properties(os.path.join(staged_component_path, 'yarn-kill.py'), properties)
+        self._fill_properties(os.path.join(staged_component_path, 'flink-stop.py'), properties)
 
         mkdircommands = []
         mkdircommands.append('mkdir -p %s' % remote_component_tmp_path)
@@ -111,7 +111,7 @@ def create_component(self, staged_component_path, application_name, user_name, c
         commands = []
         commands.append('sudo cp %s/%s %s' % (remote_component_tmp_path, service_script, service_script_install_path))
         commands.append('sudo cp %s/* %s' % (remote_component_tmp_path, remote_component_install_path))
-        commands.append('sudo chmod a+x %s/yarn-kill.py' % (remote_component_install_path))
+        commands.append('sudo chmod a+x %s/flink-stop.py' % (remote_component_install_path))
 
         if 'component_main_jar' in properties:
             commands.append('cd %s && sudo jar uf %s application.properties' % (remote_component_install_path, properties['component_main_jar']))

api/src/main/resources/plugins/flink.systemd.service.py.tpl

@@ -5,8 +5,8 @@ Description=PNDA Application: ${component_application}-${component_name}
 Type=simple
 User=${application_user}
 WorkingDirectory=/opt/${environment_namespace}/${component_application}/${component_name}/
-ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
-ExecStopPost=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
+ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_name}/flink-stop.py
+ExecStop=/opt/${environment_namespace}/${component_application}/${component_name}/flink-stop.py
 Environment=FLINK_VERSION=${component_flink_version}
 ExecStart=${environment_flink} run -m  yarn-cluster ${component_flink_config_args} -ynm ${component_job_name} -v ${flink_python_jar} ${component_main_py} ${component_application_args}
 Restart=${component_respawn_type}

api/src/main/resources/plugins/flink.systemd.service.tpl

@@ -5,8 +5,8 @@ Description=PNDA Application: ${component_application}-${component_name}
 Type=simple
 User=${application_user}
 WorkingDirectory=/opt/${environment_namespace}/${component_application}/${component_name}/
-ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
-ExecStopPost=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
+ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_name}/flink-stop.py
+ExecStop=/opt/${environment_namespace}/${component_application}/${component_name}/flink-stop.py
 Environment=FLINK_VERSION=${component_flink_version}
 ExecStart=${environment_flink} run -m  yarn-cluster ${component_flink_config_args} -ynm ${component_job_name} --class ${component_main_class} ${component_main_jar} ${component_application_args}
 Restart=${component_respawn_type}