<p dir="auto"><strong>Apache Airflow version</strong>: <code class="notranslate">2.0.1</code></p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>):<br>
<code class="notranslate">Client Version: version.Info{Major:"1", Minor:"18", GitVersion:"v1.18.1", GitCommit:"7879fc12a63337efff607952a323df90cdc7a335", GitTreeState:"clean", BuildDate:"2020-04-10T21:53:58Z", GoVersion:"go1.14.2", Compiler:"gc", Platform:"darwin/amd64"}</code></p>
<p dir="auto"><code class="notranslate">Server Version: version.Info{Major:"1", Minor:"16+", GitVersion:"v1.16.15-gke.7800", GitCommit:"cef3156c566a1d1a4b23ee360a760f45bfbaaac1", GitTreeState:"clean", BuildDate:"2020-12-14T09:12:37Z", GoVersion:"go1.13.15b4", Compiler:"gc", Platform:"linux/amd64"}</code></p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>: <code class="notranslate">GKE</code></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>: We use scheduler HA with 2 instances.</li>
</ul>
<p dir="auto"><strong>What happened</strong>:</p>
<ul dir="auto">
<li>We are running Airflow 2.0.1 using KubernetesExecutor and PostgreSQL 9.6.2.</li>
<li>Task is <code class="notranslate">up_for_retry</code> after its worker pod fails to start</li>
<li>It gets stuck in <code class="notranslate">queued</code> state</li>
<li>It runs only after a scheduler restart.</li>
</ul>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<ul dir="auto">
<li>Task gets rescheduled and runs successfully.</li>
</ul>
<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">Logs below.</p>
<details>
<p dir="auto">Its upstream task succeeds and schedules it via fast follow.<br>
<code class="notranslate">[2021-03-01 17:08:51,306] {taskinstance.py:1166} INFO - Marking task as SUCCESS. dag_id=datalake_dag_id, task_id=processidlogs, execution_date=20210301T163000, start_date=20210301T170546, end_date=20210301T170851</code></p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:08:51,339] {taskinstance.py:1220} INFO - 1 downstream tasks scheduled from follow-on schedule check</code></p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:08:51,357] {local_task_job.py:146} INFO - Task exited with return code 0</code></p>
<p dir="auto">Task is attempted to be run.</p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:08:52,229] {scheduler_job.py:1105} INFO - Sending TaskInstanceKey(dag_id='datalake_dag_id', task_id='delta_id_logs', execution_date=datetime.datetime(2021, 3, 1, 16, 30, tzinfo=Timezone('UTC')), try_number=1) to executor with priority 1 and queue default</code></p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:08:52,308] {kubernetes_executor.py:306} DEBUG - Kubernetes running for command ['airflow', 'tasks', 'run', 'datalake_dag_id', 'delta_id_logs', '2021-03-01T16:30:00+00:00', '--local', '--pool', 'default_pool', '--subdir', '/opt/airflow/dags/data_lake/some_tasks/some_tasks.py']</code></p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:08:52,332] {scheduler_job.py:1206} INFO - Executor reports execution of datalake_dag_id.delta_id_logs execution_date=2021-03-01 16:30:00+00:00 exited with status queued for try_number 1</code></p>
<p dir="auto">Pod fails to start.</p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:12:17,319] {kubernetes_executor.py:197} INFO - Event: Failed to start pod datalakedagiddeltaidlogs.5fa98ae3856f4cb4b6c8810ac13e5c6a, will reschedule</code></p>
<p dir="auto">It is put as up_for_reschedule.</p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:12:23,912] {kubernetes_executor.py:343} DEBUG - Processing task ('datalakedagiddeltaidlogs.5fa98ae3856f4cb4b6c8810ac13e5c6a', 'prod', 'up_for_reschedule', {'dag_id': 'datalake_dag_id', 'task_id': 'delta_id_logs', 'execution_date': '2021-03-01T16:30:00+00:00', 'try_number': '1'}, '1172208829')</code></p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:12:23,930] {kubernetes_executor.py:528} INFO - Changing state of (TaskInstanceKey(dag_id='datalake_dag_id', task_id='delta_id_logs', execution_date=datetime.datetime(2021, 3, 1, 16, 30, tzinfo=tzlocal()), try_number=1), 'up_for_reschedule', 'datalakedagiddeltaidlogs.5fa98ae3856f4cb4b6c8810ac13e5c6a', 'prod', '1172208829') to up_for_reschedule</code></p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:12:23,941] {scheduler_job.py:1206} INFO - Executor reports execution of datalake_dag_id.delta_id_logs execution_date=2021-03-01 16:30:00+00:00 exited with status up_for_reschedule for try_number 1</code></p>
<p dir="auto">A few minutes later, another scheduler finds it in queued state.</p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:15:39,177] {taskinstance.py:851} DEBUG - Dependencies all met for &lt;TaskInstance: datalake_dag_id.delta_id_logs 2021-03-01 16:30:00+00:00 [queued]&gt;</code></p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:15:40,477] {taskinstance.py:866} DEBUG - &lt;TaskInstance: datalake_dag_id.delta_id_logs 2021-03-01 16:30:00+00:00 [queued]&gt; dependency 'Not In Retry Period' PASSED: True, The context specified that being in a retry period was permitted.</code></p>
<p dir="auto"><code class="notranslate">[2021-03-01 17:15:40,478] {taskinstance.py:866} DEBUG - &lt;TaskInstance: datalake_dag_id.delta_id_logs 2021-03-01 16:30:00+00:00 [queued]&gt; dependency 'Previous Dagrun State' PASSED: True, The task did not have depends_on_past set.</code></p>
<p dir="auto">It stays in that state for another hour and a half until the scheduler is restarted.</p>
<p dir="auto">Finally, it is rescheduled.<br>
<code class="notranslate">[2021-03-01 18:58:10,475] {kubernetes_executor.py:463} INFO - TaskInstance: &lt;TaskInstance: datalake_dag_id.delta_id_logs 2021-03-01 16:30:00+00:00 [queued]&gt; found in queued state but was not launched, rescheduling</code></p>
</details>
<p dir="auto">Other tasks run fine while that one is stuck in queued state. We have a cron job to restart the scheduler as a hack to recover from when such cases happen, but we would like to avoid it as much as possible.</p>
<p dir="auto">We run 60 DAGs with 50-100 tasks each every 30 minutes. We have been seeing this issue at least once daily since we upgraded to Airflow 2.0.1.</p>
<p dir="auto">I understand there are some open issues about the scheduler or tasks getting stuck. But I could not tell if this is related, since hundreds of other tasks run as expected. Apologies if this turns out to be a duplicate of an existing issue. Thank you.</p>
<p dir="auto"><strong>Apache Airflow version</strong>: 2.1.0</p>
<p dir="auto"><strong>What happened</strong>:</p>
<p dir="auto">When I tried to call <code class="notranslate">logging.exception()</code>, the task instances fails.</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="def func():
    import logging
    logging.exception(&quot;some error message&quot;)

with dag:
    p = PythonOperator(
        task_id='test',
        python_callable=func,
    )"><pre class="notranslate"><span class="pl-k">def</span> <span class="pl-en">func</span>():
    <span class="pl-k">import</span> <span class="pl-s1">logging</span>
    <span class="pl-s1">logging</span>.<span class="pl-en">exception</span>(<span class="pl-s">"some error message"</span>)

<span class="pl-k">with</span> <span class="pl-s1">dag</span>:
    <span class="pl-s1">p</span> <span class="pl-c1">=</span> <span class="pl-v">PythonOperator</span>(
        <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">'test'</span>,
        <span class="pl-s1">python_callable</span><span class="pl-c1">=</span><span class="pl-s1">func</span>,
    )</pre></div>
<details>
 <summary>task failure stack strace</summary>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Traceback (most recent call last):
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py&quot;, line 1138, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py&quot;, line 1311, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py&quot;, line 1341, in _execute_task
    result = task_copy.execute(context=context)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/operators/python.py&quot;, line 117, in execute
    return_value = self.execute_callable()
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/operators/python.py&quot;, line 128, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File &quot;/usr/local/airflow/src/__main__.py&quot;, line 12, in main
    src.file_get_sftp(sftp_site_folder,dest_dir)
  File &quot;/usr/local/airflow/src/controller/source.py&quot;, line 102, in file_get_sftp
    logging.exception(&quot;Destination directory doesn't exist&quot;)
  File &quot;/usr/local/lib/python3.7/logging/__init__.py&quot;, line 1967, in exception
    error(msg, *args, exc_info=exc_info, **kwargs)
  File &quot;/usr/local/lib/python3.7/logging/__init__.py&quot;, line 1959, in error
    root.error(msg, *args, **kwargs)
  File &quot;/usr/local/lib/python3.7/logging/__init__.py&quot;, line 1407, in error
    self._log(ERROR, msg, args, **kwargs)
  File &quot;/usr/local/lib/python3.7/logging/__init__.py&quot;, line 1514, in _log
    self.handle(record)
  File &quot;/usr/local/lib/python3.7/logging/__init__.py&quot;, line 1523, in handle
    if (not self.disabled) and self.filter(record):
  File &quot;/usr/local/lib/python3.7/logging/__init__.py&quot;, line 751, in filter
    result = f.filter(record)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py&quot;, line 161, in filter
    exc.args = (self.redact(v) for v in exc.args)
AttributeError: 'NoneType' object has no attribute 'args'"><pre class="notranslate"><code class="notranslate">Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1138, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1311, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1341, in _execute_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/python3.7/site-packages/airflow/operators/python.py", line 117, in execute
    return_value = self.execute_callable()
  File "/usr/local/lib/python3.7/site-packages/airflow/operators/python.py", line 128, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/usr/local/airflow/src/__main__.py", line 12, in main
    src.file_get_sftp(sftp_site_folder,dest_dir)
  File "/usr/local/airflow/src/controller/source.py", line 102, in file_get_sftp
    logging.exception("Destination directory doesn't exist")
  File "/usr/local/lib/python3.7/logging/__init__.py", line 1967, in exception
    error(msg, *args, exc_info=exc_info, **kwargs)
  File "/usr/local/lib/python3.7/logging/__init__.py", line 1959, in error
    root.error(msg, *args, **kwargs)
  File "/usr/local/lib/python3.7/logging/__init__.py", line 1407, in error
    self._log(ERROR, msg, args, **kwargs)
  File "/usr/local/lib/python3.7/logging/__init__.py", line 1514, in _log
    self.handle(record)
  File "/usr/local/lib/python3.7/logging/__init__.py", line 1523, in handle
    if (not self.disabled) and self.filter(record):
  File "/usr/local/lib/python3.7/logging/__init__.py", line 751, in filter
    result = f.filter(record)
  File "/usr/local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 161, in filter
    exc.args = (self.redact(v) for v in exc.args)
AttributeError: 'NoneType' object has no attribute 'args'
</code></pre></div>
</details>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<p dir="auto">I expect an exception message to be printed out.</p>
<p dir="auto"><strong>How to reproduce it</strong>:</p>
<ol dir="auto">
<li>Create a simple DAG that calls <code class="notranslate">logging.exception()</code></li>
<li>Execute the function</li>
</ol>
<p dir="auto"><strong>Anything else we need to know</strong>:</p>
<p dir="auto">This issue was created for the PR (<a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="900588486" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/16047" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/16047/hovercard" href="https://github.com/apache/airflow/pull/16047">#16047</a>).</p>