<p dir="auto"><strong>Apache Airflow version</strong>: 2.1.2 and also previous versions<br>
I find this relevant for investigation: <code class="notranslate">apache-airflow-providers-apache-spark==1.0.3</code></p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>): NA</p>
<p dir="auto"><strong>Environment</strong>: linux</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>: NA</li>
<li><strong>OS</strong> (e.g. from /etc/os-release): NA</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>): NA</li>
<li><strong>Install tools</strong>: systemd to manage worker lifecycle</li>
<li><strong>Others</strong>:</li>
</ul>
<p dir="auto"><strong>What happened</strong>:<br>
We launch all our spark processes in cluster mode on YARN. This helps us in keeping resource hungry driver processes on YARN and airflow workers usually stay with deterministic workloads. During maintenance, we need to restart workers. Our users reported to have duplicate data in some instances and the time period for the respective job runs coincides with worker restarts. On investigation, I found out that airflow jobs are marked as failed due to SIGTERM during worker restart but spark job kept running on cluster. Due to retry policy, airflow launched another instance of exactly same spark job.</p>
<p dir="auto"><strong>What you expected to happen</strong>:<br>
Spark job should have been killed before worker exit. The implementation is there but it seems to be not doing the trick.</p>
<p dir="auto"><strong>How to reproduce it</strong>:</p>
<ul dir="auto">
<li>Use spark connection with yarn cluster mode.</li>
<li>With spark operator, submit a spark job. It can even be a sleep of significant time. Keep number of retries for this task more than 1.</li>
<li>Restart worker.</li>
<li>You should see airflow launching another task</li>
<li>Check yarn, there should be two instances of spark job instead of one.</li>
</ul>
<p dir="auto"><strong>Anything else we need to know</strong>:<br>
TLDR; There are couple of odd things with <code class="notranslate">on_kill</code> in spark hook.</p>
<ul dir="auto">
<li>Same airflow task runner seems to be getting multiple SIGTERMs.</li>
<li><code class="notranslate">subprocess.wait</code> can not complete. It might be due to multiple SIGTERMs.</li>
<li>Not executing renewer block and removing wait from subsequent subprocess call successfully kills the spark process from YARN.</li>
</ul>
<p dir="auto">I did some investigation on the issue. The issue is around multiple <code class="notranslate">subprocess.wait</code> calls in <code class="notranslate">on_kill</code> calls of the spark submit hook. I had to bypass the kerberos renewer block to get following exception while in <code class="notranslate">wait</code>:</p>
<pre class="notranslate">Traceback (most recent call last):
  File "/xxx/lib64/python3.6/site-packages/airflow/providers/apache/spark/hooks/spark_submit.py", line 675, in on_kill
    self.log.info("YARN app killed with return code: %s", yarn_kill.wait())
  File "/usr/lib64/python3.6/subprocess.py", line 1477, in wait
    (pid, sts) = self._try_wait(0)
  File "/usr/lib64/python3.6/subprocess.py", line 1424, in _try_wait
    (pid, sts) = os.waitpid(self.pid, wait_flags)
  File "/xxx/lib64/python3.6/site-packages/airflow/models/taskinstance.py", line 1286, in signal_handler
    raise AirflowException("Task received SIGTERM signal")
airflow.exceptions.AirflowException: Task received SIGTERM signal
</pre>
<p dir="auto">It seems like kerberos renewer block is silently swallowing that exception <em>and</em> at the same time not letting yarn kill block get invoked.<br>
Continuing on this path, removing wait from the yarn_kill command, invoked yarn kill successfully.  I also added <code class="notranslate">trap '' SIGTERM;</code> for yarn_kill just to be sure that <code class="notranslate">SIGTERM</code> is not propagated to yarn_kill command. [Essentially we do want to kill running spark job at any cost]</p>
<p dir="auto">How often does this problem occur?<br>
Pretty much on each worker restart.</p>
<p dir="auto">Any relevant logs to include? Put them here in side a detail tag:</p>
<details><summary>What happens</summary>
<pre class="notranslate">[2021-08-04 22:41:06,515] {local_task_job.py:77} ERROR - Received SIGTERM. Terminating subprocesses
[2021-08-04 22:41:06,521] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 24135
[2021-08-04 22:41:06,522] {taskinstance.py:1284} ERROR - Received SIGTERM. Terminating subprocesses.
[2021-08-04 22:41:06,523] {spark_submit.py:657} INFO - Sending kill signal to spark-submit
[2021-08-04 22:41:06,523] {spark_submit.py:660} INFO - Yarn application id: application_1627904414005_0022
[2021-08-04 22:41:06,523] {kerberos.py:74} INFO - Re-initialising kerberos from keytab: kinit -r 3600m -k -t /xxx/airflow.keytab -c /xxx/krb5cc_airflow airflow
[2021-08-04 22:41:07,034] {process_utils.py:66} INFO - Process psutil.Process(pid=26346, status='terminated') (26346) terminated with exit code None
[2021-08-04 22:41:07,221] {local_task_job.py:77} ERROR - Received SIGTERM. Terminating subprocesses
[2021-08-04 22:41:07,222] {taskinstance.py:1284} ERROR - Received SIGTERM. Terminating subprocesses.
[2021-08-04 22:41:07,228] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 24135
[2021-08-04 22:41:07,228] {taskinstance.py:1284} ERROR - Received SIGTERM. Terminating subprocesses.
[2021-08-04 22:41:07,235] {process_utils.py:66} INFO - Process psutil.Process(pid=26433, status='terminated') (26433) terminated with exit code None
[2021-08-04 22:41:07,235] {process_utils.py:66} INFO - Process psutil.Process(pid=24135, status='terminated') (24135) terminated with exit code 1
</pre>
Essentially, we don't see following block:
<pre class="notranslate">[2021-08-04 22:45:51,092] {kerberos.py:74} INFO - Re-initialising kerberos from keytab: kinit -r 3600m -k -t /xxx/airflow.keytab -c /xxx/krb5cc_airflow airflow
[2021-08-04 22:45:51,596] {process_utils.py:66} INFO - Process psutil.Process(pid=32619, status='terminated') (32619) terminated with exit code None
[2021-08-04 22:45:52,604] {kerberos.py:126} INFO - Renewing kerberos ticket to work around kerberos 1.8.1: kinit -c /xxx/krb5cc_airflow -R
[2021-08-04 22:45:53,975] {spark_submit.py:681} INFO - YARN app killed with return code: 0
</pre>
</details>
<p dir="auto">What works:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="            if self._yarn_application_id:
                yarn_trap_kill_cmd = f&quot;trap '' SIGTERM; yarn application -kill {self._yarn_application_id}&quot;
                kill_cmd = [&quot;bash&quot;, &quot;-c&quot;, yarn_trap_kill_cmd]
                env = {**os.environ, **(self._env or {})}
                # self._skip_renew_keytab is set to False for backward compat.
                if not self._skip_renew_keytab and self._keytab is not None and self._principal is not None:
                    # we are ignoring renewal failures from renew_from_kt
                    # here as the failure could just be due to a non-renewable ticket,
                    # we still attempt to kill the yarn application
                    renew_from_kt(self._principal, self._keytab, exit_on_fail=False)
                    env = os.environ.copy()
                    env[&quot;KRB5CCNAME&quot;] = airflow_conf.get('kerberos', 'ccache')

                subprocess.Popen(kill_cmd, env=env, start_new_session=True)"><pre class="notranslate">            <span class="pl-k">if</span> <span class="pl-s1">self</span>.<span class="pl-s1">_yarn_application_id</span>:
                <span class="pl-s1">yarn_trap_kill_cmd</span> <span class="pl-c1">=</span> <span class="pl-s">f"trap '' SIGTERM; yarn application -kill <span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">self</span>.<span class="pl-s1">_yarn_application_id</span><span class="pl-kos">}</span></span>"</span>
                <span class="pl-s1">kill_cmd</span> <span class="pl-c1">=</span> [<span class="pl-s">"bash"</span>, <span class="pl-s">"-c"</span>, <span class="pl-s1">yarn_trap_kill_cmd</span>]
                <span class="pl-s1">env</span> <span class="pl-c1">=</span> {<span class="pl-c1">**</span><span class="pl-s1">os</span>.<span class="pl-s1">environ</span>, <span class="pl-c1">**</span>(<span class="pl-s1">self</span>.<span class="pl-s1">_env</span> <span class="pl-c1">or</span> {})}
                <span class="pl-c"># self._skip_renew_keytab is set to False for backward compat.</span>
                <span class="pl-k">if</span> <span class="pl-c1">not</span> <span class="pl-s1">self</span>.<span class="pl-s1">_skip_renew_keytab</span> <span class="pl-c1">and</span> <span class="pl-s1">self</span>.<span class="pl-s1">_keytab</span> <span class="pl-c1">is</span> <span class="pl-c1">not</span> <span class="pl-c1">None</span> <span class="pl-c1">and</span> <span class="pl-s1">self</span>.<span class="pl-s1">_principal</span> <span class="pl-c1">is</span> <span class="pl-c1">not</span> <span class="pl-c1">None</span>:
                    <span class="pl-c"># we are ignoring renewal failures from renew_from_kt</span>
                    <span class="pl-c"># here as the failure could just be due to a non-renewable ticket,</span>
                    <span class="pl-c"># we still attempt to kill the yarn application</span>
                    <span class="pl-en">renew_from_kt</span>(<span class="pl-s1">self</span>.<span class="pl-s1">_principal</span>, <span class="pl-s1">self</span>.<span class="pl-s1">_keytab</span>, <span class="pl-s1">exit_on_fail</span><span class="pl-c1">=</span><span class="pl-c1">False</span>)
                    <span class="pl-s1">env</span> <span class="pl-c1">=</span> <span class="pl-s1">os</span>.<span class="pl-s1">environ</span>.<span class="pl-en">copy</span>()
                    <span class="pl-s1">env</span>[<span class="pl-s">"KRB5CCNAME"</span>] <span class="pl-c1">=</span> <span class="pl-s1">airflow_conf</span>.<span class="pl-en">get</span>(<span class="pl-s">'kerberos'</span>, <span class="pl-s">'ccache'</span>)

                <span class="pl-s1">subprocess</span>.<span class="pl-v">Popen</span>(<span class="pl-s1">kill_cmd</span>, <span class="pl-s1">env</span><span class="pl-c1">=</span><span class="pl-s1">env</span>, <span class="pl-s1">start_new_session</span><span class="pl-c1">=</span><span class="pl-c1">True</span>)</pre></div>
<p dir="auto"><strong>Apache Airflow version</strong>: 1.7.1.2</p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>):</p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>:</li>
<li><strong>OS</strong> (e.g. from /etc/os-release):</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>):</li>
<li><strong>Install tools</strong>:</li>
<li><strong>Others</strong>:<br>
<strong>What happened</strong>:</li>
</ul>
<p dir="auto">Since the rework of a rolling webserver restart the "airflow webserver -D" does not detach from the console anymore</p>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<p dir="auto"><strong>How to reproduce it</strong>:</p>
<p dir="auto"><strong>Anything else we need to know</strong>:</p>
<p dir="auto">Moved here from <a href="https://issues.apache.org/jira/browse/AIRFLOW-486" rel="nofollow">https://issues.apache.org/jira/browse/AIRFLOW-486</a></p>