<p dir="auto"><strong>Apache Airflow version</strong>: 2.1.0</p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>): v1.18.17-gke.1901</p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>: Google Cloud</li>
<li><strong>OS</strong> (e.g. from /etc/os-release): Debian GNU/Linux 10 (buster)</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>): Linux airflow-scheduler-7697b66974-m6mct 5.4.89+ <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="69689814" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/1" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/1/hovercard" href="https://github.com/apache/airflow/pull/1">#1</a> SMP Sat Feb 13 19:45:14 PST 2021 x86_64 GNU/Linux</li>
<li><strong>Install tools</strong>:</li>
<li><strong>Others</strong>:</li>
</ul>
<p dir="auto"><strong>What happened</strong>:<br>
When the <code class="notranslate">scheduler</code> is restarted the currently running tasks are facing SIGTERM error.<br>
Every time the <code class="notranslate">scheduler</code> is restarted or re-deployed then the current <code class="notranslate">scheduler</code> is terminated and a new <code class="notranslate">scheduler</code> is created. If during this process exist tasks running the new <code class="notranslate">scheduler</code> will terminate these tasks with <code class="notranslate">complete</code> status and new tasks will be created to continue the work of the terminated ones. After few seconds the new tasks are terminated with <code class="notranslate">error</code> status and SIGTERM error.</p>
<details><summary>Error log</summary> [2021-07-07 14:59:49,024] {cursor.py:661} INFO - query execution done
[2021-07-07 14:59:49,025] {arrow_result.pyx:0} INFO - fetching data done
[2021-07-07 15:00:07,361] {local_task_job.py:196} WARNING - State of this instance has been externally set to failed. Terminating instance.
[2021-07-07 15:00:07,363] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 150
[2021-07-07 15:00:12,845] {taskinstance.py:1264} ERROR - Received SIGTERM. Terminating subprocesses.
[2021-07-07 15:00:12,907] {process_utils.py:66} INFO - Process psutil.Process(pid=150, status='terminated', exitcode=0, started='14:59:46') (150) terminated with exit code 0 </details>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<p dir="auto">The tasks currently running should be allowed to finish their process or the substitute tasks should execute their process with success.<br>
The new <code class="notranslate">scheduler</code> should interfere with the running tasks.</p>
<p dir="auto"><strong>How to reproduce it</strong>:</p>
<p dir="auto">To reproduce is necessary to start a DAG that has some task(s) that take some minutes to be completed. During this task(s) processing a new deploy for <code class="notranslate">scheduler</code> should be executed. During the re-deploy, the current <code class="notranslate">scheduler</code> will be terminated and a new one will be created. The current task(s) will be completed (without finish their processing) and substituted for new ones that will fail in seconds.</p>
<p dir="auto"><strong>Anything else we need to know</strong>:</p>
<p dir="auto">The problem was not happening with Airflow 1.10.15 and it started to happens after the upgrade to Airflow 2.1.0.</p>
<p dir="auto"><strong>Apache Airflow version</strong>: 2.1.0</p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>):</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Client Version: version.Info{Major:&quot;1&quot;, Minor:&quot;21&quot;, GitVersion:&quot;v1.21.0&quot;, GitCommit:&quot;cb303e613a121a29364f75cc67d3d580833a7479&quot;, GitTreeState:&quot;clean&quot;, BuildDate:&quot;2021-04-08T16:31:21Z&quot;, GoVersion:&quot;go1.16.1&quot;, Compiler:&quot;gc&quot;, Platform:&quot;linux/amd64&quot;}
Server Version: version.Info{Major:&quot;1&quot;, Minor:&quot;20&quot;, GitVersion:&quot;v1.20.7&quot;, GitCommit:&quot;6b3f9b283463c1d5a2455df301182805e65c7145&quot;, GitTreeState:&quot;clean&quot;, BuildDate:&quot;2021-05-19T22:28:47Z&quot;, GoVersion:&quot;go1.15.12&quot;, Compiler:&quot;gc&quot;, Platform:&quot;linux/amd64&quot;}"><pre class="notranslate"><code class="notranslate">Client Version: version.Info{Major:"1", Minor:"21", GitVersion:"v1.21.0", GitCommit:"cb303e613a121a29364f75cc67d3d580833a7479", GitTreeState:"clean", BuildDate:"2021-04-08T16:31:21Z", GoVersion:"go1.16.1", Compiler:"gc", Platform:"linux/amd64"}
Server Version: version.Info{Major:"1", Minor:"20", GitVersion:"v1.20.7", GitCommit:"6b3f9b283463c1d5a2455df301182805e65c7145", GitTreeState:"clean", BuildDate:"2021-05-19T22:28:47Z", GoVersion:"go1.15.12", Compiler:"gc", Platform:"linux/amd64"}
</code></pre></div>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>: Azure</li>
<li><strong>OS</strong> (e.g. from /etc/os-release): ubuntu 18.04</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>):</li>
<li><strong>Install tools</strong>:</li>
<li><strong>Others</strong>:</li>
</ul>
<p dir="auto"><strong>What happened</strong>:<br>
Since I launched airflow 2.1.0 on our cluster on Friday, the scheduler has failed 716 times stating "BrokenPipeError"</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2021-06-07 12:07:19,362] {scheduler_job.py:1205} INFO - Executor reports execution of demo_git_notebook_parameterized.demo_git_notebook_parameterized execution_date=2021-06-07 12:05:41.835167+00:00 exited with status None for try_number 1
[2021-06-07 12:07:22,798] {scheduler_job.py:748} INFO - Exiting gracefully upon receiving signal 15
[2021-06-07 12:07:23,800] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 55
[2021-06-07 12:07:24,154] {process_utils.py:207} INFO - Waiting up to 5 seconds for processes to exit...
[2021-06-07 12:07:24,211] {process_utils.py:207} INFO - Waiting up to 5 seconds for processes to exit...
[2021-06-07 12:07:24,265] {process_utils.py:66} INFO - Process psutil.Process(pid=55, status='terminated', exitcode=0, started='12:02:39') (55) terminated with exit code 0
[2021-06-07 12:07:24,266] {process_utils.py:66} INFO - Process psutil.Process(pid=7433, status='terminated', started='12:07:23') (7433) terminated with exit code None
[2021-06-07 12:07:24,266] {process_utils.py:66} INFO - Process psutil.Process(pid=7432, status='terminated', started='12:07:22') (7432) terminated with exit code None
[2021-06-07 12:07:24,266] {kubernetes_executor.py:759} INFO - Shutting down Kubernetes executor
[2021-06-07 12:07:24,266] {scheduler_job.py:1308} ERROR - Exception when executing Executor.end
Traceback (most recent call last):
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1286, in _execute
    self._run_scheduler_loop()
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1400, in _run_scheduler_loop
    time.sleep(min(self._processor_poll_interval, next_event))
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py&quot;, line 751, in _exit_gracefully
    sys.exit(os.EX_OK)
SystemExit: 0

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1306, in _execute
    self.executor.end()
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py&quot;, line 761, in end
    self._flush_task_queue()
  File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py&quot;, line 714, in _flush_task_queue
    self.log.debug('Executor shutting down, task_queue approximate size=%d', self.task_queue.qsize())
  File &quot;&lt;string&gt;&quot;, line 2, in qsize
  File &quot;/usr/local/lib/python3.8/multiprocessing/managers.py&quot;, line 834, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;/usr/local/lib/python3.8/multiprocessing/connection.py&quot;, line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File &quot;/usr/local/lib/python3.8/multiprocessing/connection.py&quot;, line 411, in _send_bytes
    self._send(header + buf)
  File &quot;/usr/local/lib/python3.8/multiprocessing/connection.py&quot;, line 368, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
[2021-06-07 12:07:24,268] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 55
[2021-06-07 12:07:24,268] {scheduler_job.py:1313} INFO - Exited execute loop"><pre class="notranslate"><code class="notranslate">[2021-06-07 12:07:19,362] {scheduler_job.py:1205} INFO - Executor reports execution of demo_git_notebook_parameterized.demo_git_notebook_parameterized execution_date=2021-06-07 12:05:41.835167+00:00 exited with status None for try_number 1
[2021-06-07 12:07:22,798] {scheduler_job.py:748} INFO - Exiting gracefully upon receiving signal 15
[2021-06-07 12:07:23,800] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 55
[2021-06-07 12:07:24,154] {process_utils.py:207} INFO - Waiting up to 5 seconds for processes to exit...
[2021-06-07 12:07:24,211] {process_utils.py:207} INFO - Waiting up to 5 seconds for processes to exit...
[2021-06-07 12:07:24,265] {process_utils.py:66} INFO - Process psutil.Process(pid=55, status='terminated', exitcode=0, started='12:02:39') (55) terminated with exit code 0
[2021-06-07 12:07:24,266] {process_utils.py:66} INFO - Process psutil.Process(pid=7433, status='terminated', started='12:07:23') (7433) terminated with exit code None
[2021-06-07 12:07:24,266] {process_utils.py:66} INFO - Process psutil.Process(pid=7432, status='terminated', started='12:07:22') (7432) terminated with exit code None
[2021-06-07 12:07:24,266] {kubernetes_executor.py:759} INFO - Shutting down Kubernetes executor
[2021-06-07 12:07:24,266] {scheduler_job.py:1308} ERROR - Exception when executing Executor.end
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1286, in _execute
    self._run_scheduler_loop()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1400, in _run_scheduler_loop
    time.sleep(min(self._processor_poll_interval, next_event))
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 751, in _exit_gracefully
    sys.exit(os.EX_OK)
SystemExit: 0

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1306, in _execute
    self.executor.end()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 761, in end
    self._flush_task_queue()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 714, in _flush_task_queue
    self.log.debug('Executor shutting down, task_queue approximate size=%d', self.task_queue.qsize())
  File "&lt;string&gt;", line 2, in qsize
  File "/usr/local/lib/python3.8/multiprocessing/managers.py", line 834, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 411, in _send_bytes
    self._send(header + buf)
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 368, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
[2021-06-07 12:07:24,268] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 55
[2021-06-07 12:07:24,268] {scheduler_job.py:1313} INFO - Exited execute loop
</code></pre></div>
<p dir="auto"><strong>What you expected to happen</strong>:<br>
For it to not do that.</p>
<p dir="auto"><strong>How to reproduce it</strong>:<br>
I'm not too sure. Could it be an issue with Airflow 2.1.0 itself, and it can be reproduced just by launching it in a cluster? Using KubernetesExecutor, no celery.<br>
Could it be an issue with Azure?</p>
<p dir="auto"><strong>Anything else we need to know</strong>:<br>
by my very rough calculations it happens every 6 minutes?</p>