<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.5.0</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">A user with access to manually triggering DAGs can trigger a DAG. provide a run_id that matches the pattern used when creating scheduled runs and cause the scheduler to crash due to database unique key violation:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="2022-12-12 12:58:00,793] {scheduler_job.py:776} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File &quot;/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py&quot;, line 759, in _execute
    self._run_scheduler_loop()
  File &quot;/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py&quot;, line 885, in _run_scheduler_loop
    num_queued_tis = self._do_scheduling(session)
  File &quot;/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py&quot;, line 956, in _do_scheduling
    self._create_dagruns_for_dags(guard, session)
  File &quot;/usr/local/lib/python3.8/site-packages/airflow/utils/retries.py&quot;, line 78, in wrapped_function
    for attempt in run_with_db_retries(max_retries=retries, logger=logger, **retry_kwargs):
  File &quot;/usr/local/lib/python3.8/site-packages/tenacity/__init__.py&quot;, line 384, in __iter__
    do = self.iter(retry_state=retry_state)
  File &quot;/usr/local/lib/python3.8/site-packages/tenacity/__init__.py&quot;, line 351, in iter
    return fut.result()
  File &quot;/usr/local/lib/python3.8/concurrent/futures/_base.py&quot;, line 437, in result
    return self.__get_result()
  File &quot;/usr/local/lib/python3.8/concurrent/futures/_base.py&quot;, line 389, in __get_result
    raise self._exception
  File &quot;/usr/local/lib/python3.8/site-packages/airflow/utils/retries.py&quot;, line 87, in wrapped_function
    return func(*args, **kwargs)
  File &quot;/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1018, in _create_dagruns_for_dags
    query, dataset_triggered_dag_info = DagModel.dags_needing_dagruns(session)
  File &quot;/usr/local/lib/python3.8/site-packages/airflow/models/dag.py&quot;, line 3341, in dags_needing_dagruns
    for x in session.query(
  File &quot;/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/query.py&quot;, line 2773, in all
    return self._iter().all()
  File &quot;/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/query.py&quot;, line 2916, in _iter
    result = self.session.execute(
  File &quot;/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py&quot;, line 1713, in execute
    conn = self._connection_for_bind(bind)
  File &quot;/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py&quot;, line 1552, in _connection_for_bind
    return self._transaction._connection_for_bind(
  File &quot;/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py&quot;, line 721, in _connection_for_bind
    self._assert_active()
  File &quot;/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py&quot;, line 601, in _assert_active
    raise sa_exc.PendingRollbackError(
sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint &quot;dag_run_dag_id_run_id_key&quot;
DETAIL:  Key (dag_id, run_id)=(example_branch_dop_operator_v3, scheduled__2022-12-12T12:57:00+00:00) already exists.

[SQL: INSERT INTO dag_run (dag_id, queued_at, execution_date, start_date, end_date, state, run_id, creating_job_id, external_trigger, run_type, conf, data_interval_start, data_interval_end, last_scheduling_decision, dag_hash, log_template_id, updated_at) VALUES (%(dag_id)s, %(queued_at)s, %(execution_date)s, %(start_date)s, %(end_date)s, %(state)s, %(run_id)s, %(creating_job_id)s, %(external_trigger)s, %(run_type)s, %(conf)s, %(data_interval_start)s, %(data_interval_end)s, %(last_scheduling_decision)s, %(dag_hash)s, (SELECT max(log_template.id) AS max_1 
FROM log_template), %(updated_at)s) RETURNING dag_run.id]
[parameters: {'dag_id': 'example_branch_dop_operator_v3', 'queued_at': datetime.datetime(2022, 12, 12, 12, 58, 0, 435945, tzinfo=Timezone('UTC')), 'execution_date': DateTime(2022, 12, 12, 12, 57, 0, tzinfo=Timezone('UTC')), 'start_date': None, 'end_date': None, 'state': &lt;DagRunState.QUEUED: 'queued'&gt;, 'run_id': 'scheduled__2022-12-12T12:57:00+00:00', 'creating_job_id': 1, 'external_trigger': False, 'run_type': &lt;DagRunType.SCHEDULED: 'scheduled'&gt;, 'conf': &lt;psycopg2.extensions.Binary object at 0x7f283a82af60&gt;, 'data_interval_start': DateTime(2022, 12, 12, 12, 57, 0, tzinfo=Timezone('UTC')), 'data_interval_end': DateTime(2022, 12, 12, 12, 58, 0, tzinfo=Timezone('UTC')), 'last_scheduling_decision': None, 'dag_hash': '1653a588de69ed25c5b1dcfef928479c', 'updated_at': datetime.datetime(2022, 12, 12, 12, 58, 0, 436871, tzinfo=Timezone('UTC'))}]
(Background on this error at: https://sqlalche.me/e/14/gkpj) (Background on this error at: https://sqlalche.me/e/14/7s2a)"><pre class="notranslate"><code class="notranslate">2022-12-12 12:58:00,793] {scheduler_job.py:776} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 759, in _execute
    self._run_scheduler_loop()
  File "/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 885, in _run_scheduler_loop
    num_queued_tis = self._do_scheduling(session)
  File "/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 956, in _do_scheduling
    self._create_dagruns_for_dags(guard, session)
  File "/usr/local/lib/python3.8/site-packages/airflow/utils/retries.py", line 78, in wrapped_function
    for attempt in run_with_db_retries(max_retries=retries, logger=logger, **retry_kwargs):
  File "/usr/local/lib/python3.8/site-packages/tenacity/__init__.py", line 384, in __iter__
    do = self.iter(retry_state=retry_state)
  File "/usr/local/lib/python3.8/site-packages/tenacity/__init__.py", line 351, in iter
    return fut.result()
  File "/usr/local/lib/python3.8/concurrent/futures/_base.py", line 437, in result
    return self.__get_result()
  File "/usr/local/lib/python3.8/concurrent/futures/_base.py", line 389, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.8/site-packages/airflow/utils/retries.py", line 87, in wrapped_function
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1018, in _create_dagruns_for_dags
    query, dataset_triggered_dag_info = DagModel.dags_needing_dagruns(session)
  File "/usr/local/lib/python3.8/site-packages/airflow/models/dag.py", line 3341, in dags_needing_dagruns
    for x in session.query(
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/query.py", line 2773, in all
    return self._iter().all()
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/query.py", line 2916, in _iter
    result = self.session.execute(
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1713, in execute
    conn = self._connection_for_bind(bind)
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1552, in _connection_for_bind
    return self._transaction._connection_for_bind(
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 721, in _connection_for_bind
    self._assert_active()
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 601, in _assert_active
    raise sa_exc.PendingRollbackError(
sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "dag_run_dag_id_run_id_key"
DETAIL:  Key (dag_id, run_id)=(example_branch_dop_operator_v3, scheduled__2022-12-12T12:57:00+00:00) already exists.

[SQL: INSERT INTO dag_run (dag_id, queued_at, execution_date, start_date, end_date, state, run_id, creating_job_id, external_trigger, run_type, conf, data_interval_start, data_interval_end, last_scheduling_decision, dag_hash, log_template_id, updated_at) VALUES (%(dag_id)s, %(queued_at)s, %(execution_date)s, %(start_date)s, %(end_date)s, %(state)s, %(run_id)s, %(creating_job_id)s, %(external_trigger)s, %(run_type)s, %(conf)s, %(data_interval_start)s, %(data_interval_end)s, %(last_scheduling_decision)s, %(dag_hash)s, (SELECT max(log_template.id) AS max_1 
FROM log_template), %(updated_at)s) RETURNING dag_run.id]
[parameters: {'dag_id': 'example_branch_dop_operator_v3', 'queued_at': datetime.datetime(2022, 12, 12, 12, 58, 0, 435945, tzinfo=Timezone('UTC')), 'execution_date': DateTime(2022, 12, 12, 12, 57, 0, tzinfo=Timezone('UTC')), 'start_date': None, 'end_date': None, 'state': &lt;DagRunState.QUEUED: 'queued'&gt;, 'run_id': 'scheduled__2022-12-12T12:57:00+00:00', 'creating_job_id': 1, 'external_trigger': False, 'run_type': &lt;DagRunType.SCHEDULED: 'scheduled'&gt;, 'conf': &lt;psycopg2.extensions.Binary object at 0x7f283a82af60&gt;, 'data_interval_start': DateTime(2022, 12, 12, 12, 57, 0, tzinfo=Timezone('UTC')), 'data_interval_end': DateTime(2022, 12, 12, 12, 58, 0, tzinfo=Timezone('UTC')), 'last_scheduling_decision': None, 'dag_hash': '1653a588de69ed25c5b1dcfef928479c', 'updated_at': datetime.datetime(2022, 12, 12, 12, 58, 0, 436871, tzinfo=Timezone('UTC'))}]
(Background on this error at: https://sqlalche.me/e/14/gkpj) (Background on this error at: https://sqlalche.me/e/14/7s2a)
</code></pre></div>
<p dir="auto">Worse yet, the scheduler will keep crashing after a restart with the same exception.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">A user should not be able to crash the scheduler from the UI.<br>
I see 2 alternatives for solving this:</p>
<ol dir="auto">
<li>Reject custom run_id that would (or could) collide with a scheduled one, preventing this situation from happening.</li>
<li>Handle the database error and assign a different run_id to the scheduled run.</li>
</ol>
<h3 dir="auto">How to reproduce</h3>
<ol dir="auto">
<li>Find an unpaused DAG.</li>
<li>Trigger DAG w/ config, set the run id to something like scheduled__2022-11-21T12:00:00+00:00 (adjust the time to be in the future where there is no run yet).</li>
<li>Let the manual DAG run finish.</li>
<li>Wait for the scheduler to try to schedule another DAG run with the same run id.</li>
<li>💥</li>
<li>Attempt to restart the scheduler.</li>
<li><g-emoji class="g-emoji" alias="boom" fallback-src="https://github.githubassets.com/images/icons/emoji/unicode/1f4a5.png">💥</g-emoji></li>
</ol>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux 11 (bullseye)</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-postgres==5.3.1</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Docker-Compose</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">I'm using a Postgres docker container as a metadata database that is linked via docker networking to the scheduler and the rest of the components. Scheduler, workers and webserver are all running in separate containers (using CeleryExecutor backed by a Redis container), though I do not think it is relevant in this case.</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.1.4</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">After upgrading to airflow 1.10.10 to 2.1.4, I cannot see 'Last Run' column value on dags page on airflow web ui.</p>
<h3 dir="auto"></h3>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/30676005/151303105-a7c4b8fe-cdd5-4b84-817e-33318d6355cb.png"><img src="https://user-images.githubusercontent.com/30676005/151303105-a7c4b8fe-cdd5-4b84-817e-33318d6355cb.png" alt="image" style="max-width: 100%;"></a></p>
<p dir="auto">Why does this happen?</p>
<h3 dir="auto">What you expected to happen</h3>
<p dir="auto">It seems to be a problem with the version of the sources ​​related to airflow.</p>
<p dir="auto">Or, I think it may be the cause of not executing "airflow db init" while upgrading to 2.1.4.</p>
<h3 dir="auto">How to reproduce</h3>
<ol dir="auto">
<li>Install airflow version 1.10.10</li>
<li>Upgrade airflow to 2.1.4</li>
</ol>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Linux</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>