Deferred operator causing deadlock and Airflow falled down

[nesp159de]

Apache Airflow version

Other Airflow 2 version (please specify below)

If "Other Airflow 2 version" selected, which one?

2.6.3

What happened?

We used defered operator waiting for something in several instances during high mysql database server load.
We got following issue of airflow core two times:

Exception when executing TriggererJobRunner._run_trigger_loop
Traceback (most recent call last):
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1910, in _execute_context
    self.dialect.do_execute(
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/cursors.py", line 153, in execute
    result = self._query(query)
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/cursors.py", line 322, in _query
    conn.query(q)
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/connections.py", line 558, in query
    self._affected_rows = self._read_query_result(unbuffered=unbuffered)
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/connections.py", line 822, in _read_query_result
    result.read()
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/connections.py", line 1200, in read
    first_packet = self.connection._read_packet()
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/connections.py", line 772, in _read_packet
    packet.raise_for_error()
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/protocol.py", line 221, in raise_for_error
    err.raise_mysql_exception(self._data)
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/err.py", line 143, in raise_mysql_exception
    raise errorclass(errno, errval)
pymysql.err.OperationalError: (1213, 'Deadlock found when trying to get lock; try restarting transaction')

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/airflow/jobs/triggerer_job_runner.py", line 333, in _execute
    self._run_trigger_loop()
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/airflow/jobs/triggerer_job_runner.py", line 362, in _run_trigger_loop
    self.handle_events()
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/airflow/jobs/triggerer_job_runner.py", line 390, in handle_events
    Trigger.submit_event(trigger_id=trigger_id, event=event)
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/airflow/api_internal/internal_api_call.py", line 112, in wrapper
    return func(*args, **kwargs)
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, session=session, **kwargs)
  File "/opt/gts/3pp/python/lib/python3.9/contextlib.py", line 126, in __exit__
    next(self.gen)
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/airflow/utils/session.py", line 37, in create_session
    session.commit()
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 1454, in commit
    self._transaction.commit(_to_root=self.future)
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 832, in commit
    self._prepare_impl()
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 811, in _prepare_impl
    self.session.flush()
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3449, in flush
    self._flush(objects)
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3589, in _flush
    transaction.rollback(_capture_exception=True)
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3549, in _flush
    flush_context.execute()
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/orm/unitofwork.py", line 456, in execute
    rec.execute(self)
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/orm/unitofwork.py", line 630, in execute
    util.preloaded.orm_persistence.save_obj(
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py", line 237, in save_obj
    _emit_update_statements(
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py", line 1001, in _emit_update_statements
    c = connection._execute_20(
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1710, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/sql/elements.py", line 334, in _execute_on_connection
    return connection._execute_clauseelement(
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1577, in _execute_clauseelement
    ret = self._execute_context(
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1953, in _execute_context
    self._handle_dbapi_exception(
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 2134, in _handle_dbapi_exception
    util.raise_(
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1910, in _execute_context
    self.dialect.do_execute(
  File "/opt/gts/3pp/airflow/lib/python3.9/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/cursors.py", line 153, in execute
    result = self._query(query)
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/cursors.py", line 322, in _query
    conn.query(q)
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/connections.py", line 558, in query
    self._affected_rows = self._read_query_result(unbuffered=unbuffered)
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/connections.py", line 822, in _read_query_result
    result.read()
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/connections.py", line 1200, in read
    first_packet = self.connection._read_packet()
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/connections.py", line 772, in _read_packet
    packet.raise_for_error()
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/protocol.py", line 221, in raise_for_error
    err.raise_mysql_exception(self._data)
  File "/opt/gts/3pp/python/lib/python3.9/site-packages/pymysql/err.py", line 143, in raise_mysql_exception
    raise errorclass(errno, errval)
sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (1213, 'Deadlock found when trying to get lock; try restarting transaction')

[SQL: UPDATE task_instance SET state=%(state)s, updated_at=%(updated_at)s, 
      trigger_id=%(trigger_id)s, next_kwargs=%(next_kwargs)s 
      WHERE task_instance.dag_id = %(task_instance_dag_id)s 
        AND task_instance.task_id = %(task_instance_task_id)s 
        AND task_instance.run_id = %(task_instance_run_id)s 
        AND task_instance.map_index = %(task_instance_map_index)s]
[parameters: {
    'state': <TaskInstanceState.SCHEDULED: 'scheduled'>, 
    'updated_at': datetime.datetime(2024, 12, 12, 10, 37, 41, 847579), 
    'trigger_id': None, 
    'next_kwargs': '{"__var": {"event": {"__var": 1733999860.954081, "__type": "datetime"}}, "__type": "dict"}', 
    'task_instance_dag_id': 'AE_executor', 
    'task_instance_task_id': 'RFF.RFF_WAIT_FOR_IVS_DATA_CLEANSING_STATUS', 
    'task_instance_run_id': 'manual__2024-12-12T08:01:19+00:00', 
    'task_instance_map_index': -1
}]

What you think should happen instead?

Recommended from mysql is re-trigger the operation as is already implemented in airflow for example when fixing: #41428

How to reproduce

I created code to trigger the function same way as the code raised the issue:

from operators.reschedule import RescheduleOperator
from airflow.operators.python import BaseOperator
from airflow.operators.bash import BashOperator
from operators.prisma_empty import PrismaEmptyOperator
from airflow import DAG
from datetime import datetime, timedelta
from common.logger import task_logger
from airflow.triggers.temporal import TimeDeltaTrigger

class EmptyOperator(BashOperator):
    '''
    Empty operator does literally nothing. To be used for example as placeholder for
    not-yet-implemented job.
    '''

    def __init__(self, *args, **kwargs):
        super().__init__(
            bash_command="echo 'No action'",
            *args, **kwargs)


class WaitForOperator(BaseOperator):
    def __init__(self, *,
                 poll_interval = 0.5,
                 **kwargs):

        self.timeout_deadline = datetime.now() + timedelta(days = 1)
        self.poll_interval = timedelta(seconds=poll_interval)

        super().__init__(**kwargs)

    def execute(self, context):
        now = datetime.now()
        deadline = self.timeout_deadline
        if deadline < now:
            task_logger.info(f"Deadline time is in the past. Exiting.")
            return

        task_logger.info(f"Setting deadline to: {deadline}")
        time_diff = (deadline - now).total_seconds()
        self.xcom_push(context, key="timeout_rerun_sec", value=time_diff)
        self.xcom_push(context, key="start_time", value=now.isoformat())
        self.xcom_push(context, key="deadline", value=deadline.isoformat())

        self.deferred(context)

    def evaluate_condition(self):
        return False

    # pylint: disable=unused-argument
    def deferred(self, context, event=None):
        """deferred"""

        if self.evaluate_condition():
            return

        now = datetime.now()
        deadline = datetime.fromisoformat(self.xcom_pull(context, key="deadline"))
        task_logger.info(f"Woken from deferral; deadline: {deadline}")
        diff = deadline - now
        if diff >= timedelta(seconds=0):
            task_logger.info(f"Remaining time to deadline: {diff}")
        elif self.fail_on_timeout:
            raise RuntimeError(self.message_on_timeout)
        else:
            task_logger.info("Deadline reached, considering as SUCCESS")
            return

        defer_interval = min(self.poll_interval, deadline - now)
        task_logger.info(f"Deferring for {defer_interval} (until {now + defer_interval})")
        self.defer(trigger=TimeDeltaTrigger(defer_interval), method_name="deferred")


# This sample is to be deleted soon
with DAG(
    dag_id="wait_for_drill",
    default_args=None,
    description="Spawns several instances of Wait for in order to investigate PRISMA-10686",
    tags=["prisma-samples"],
) as dag:
    beg = EmptyOperator(dag=dag, task_id="BEGIN")
    end = EmptyOperator(dag=dag, task_id="END")

    for i in range(30):
        wait_for = WaitForOperator(dag=dag, task_id=f"WAITFOR_{i}")
        beg >> wait_for

        wait_for >> end

But even with thise example I was not able to reproduce it.
The issue probably relates huge complexity of our application involve more than 300 tasks spread in 30 dags.

Operating System

Red Hat Enterprise Linux 8.10 (Ootpa)

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

Local executor + Airflow is installed on Rhel machine.

Anything else?

We see the issue two times during one day of testing.
We are sharing mysql server for Airflow and application itself.
We are using two airflow instances (each one is having itself database) over SQL alchemy.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[bugraoz93]

Seems like AI, closing.

[potiuk]

I don't think so @bugraoz93 :). This one looks pretty legitimate (and sorry @nesp159de - we recently have been flooded with almost good looking issues generated by AI so we are pretty "sensitive" now).

But I am not sure we can do anything about it regardless. There are a number of places where mysql can deadlock - where it is completely unnecessary and we know it - the deadlocks are far less often happening on Postgres and our recommendation - for now - would be to switch to postgres.

There are few other things you can do:

• try to disable "schedule_after_task_execution" - which is an often source for deadlock
• upgrade to latest Airflow 2.10.* - you are running pretty old version of Airlfow and you miss more or less 1000 fixes and new features that have been implemented since. Among them - especially in some of the latest versions of Airlfow was limiting a number of deadlocks that you might have. If you look at the changelog you will be able to find them. Generally speaking "upgrade to latest version" and "check if the problems are still there" is absolutely recommended action before anyones time is spent on looking at this issue.
• Airlfow 3 (coming in 2-3 months) will have completely revamped architecture where removing database deadlocs and generally database pressure is one of the most important changes that drove the re-architecting. Stay tuned and upgrede/test it when ready

Converting it into discussion if more discussion is needed.

[msen-aiq]

I had a similar issue where Backfill tasks are deadlocked if any task is placed after deferrable tasks. (airflow 2.9.3)
airflow.exceptions.BackfillUnfinished: BackfillJob is deadlocked.
Disabling scheduler.schedule_after_task_execution resolved my problem.

[nesp159de]

Thanks for the info, in my case it does not help.

[bugraoz93]

My bad! Thanks @potiuk! :) It is hard to differentiate whether the account or the post looks like an AI in the latest events. The accounts I saw generally have zero interaction means either they had not pushed anything or had random hello world commits 5-6 months ago. There are even with the pictures which makes it even harder if the issues look legit.

Sorry, @nesp159de! Indeed, we are a bit sensitive at the moment. What makes it more sensitive and sadder/angrier is that everyone here is trying to help each other and serve the community while being part of it. That's why, please don't take it personally. I am glad to see this didn't fade away along with other generated ones.

[nesp159de]

Thank you for fast response.
I am still trying to reproduce the issue with the provided sample, but not succeeded yet. Now I see the same issue from our testing run in Airflow: 2.10.3.
The exception is still the same from realy easy task waiting for some external action using deferred infrastructure. Still related to UPDATE task_instance SET state=.

Related to your proposals:

• Move to postgres: I already tested it because of performance issues, I was a bit surprised it was even slower than mysql.
  This solution is quite heavy for us because we have currently running app with mysql and introduce new DB engine including setup users + installation will take serious time.

• try to disable "schedule_after_task_execution" - which is an often source for deadlock: I will test it.

• Airlfow 3 it is far future, maybe our application will be released sooner.

[nesp159de]

The call stack from the latest issue observation:

[nesp159de]

2025-01-27T16:19:38.995943+01:00 de875-xv9 AirflowCoreCE-1_041_wd default.INFO watchdogD.py 552 Daemon:0
Daemon() 0.930 proc stdout: [2025-01-27T16:19:38.986+0100] {triggerer_job_runner.py:348} ERROR -
Exception when executing TriggererJobRunner._run_trigger_loop

Traceback (most recent call last):
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1910, in _execute_context
self.dialect.do_execute(
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
cursor.execute(statement, parameters)
File "/opt/cpython-3.12.4-1/lib/python3.12/site-packages/pymysql/cursors.py", line 153, in execute
result = self._query(query)
File "/opt/cpython-3.12.4-1/lib/python3.12/site-packages/pymysql/cursors.py", line 322, in _query
conn.query(q)
File "/opt/cpython-3.12.4-1/lib/python3.12/site-packages/pymysql/connections.py", line 563, in query
self._affected_rows = self._read_query_result(unbuffered=unbuffered)
File "/opt/cpython-3.12.4-1/lib/python3.12/site-packages/pymysql/connections.py", line 825, in _read_query_result
result.read()
File "/opt/cpython-3.12.4-1/lib/python3.12/site-packages/pymysql/connections.py", line 1199, in read
first_packet = self.connection._read_packet()
File "/opt/cpython-3.12.4-1/lib/python3.12/site-packages/pymysql/connections.py", line 775, in _read_packet
packet.raise_for_error()
File "/opt/cpython-3.12.4-1/lib/python3.12/site-packages/pymysql/protocol.py", line 219, in raise_for_error
err.raise_mysql_exception(self._data)
File "/opt/cpython-3.12.4-1/lib/python3.12/site-packages/pymysql/err.py", line 150, in raise_mysql_exception
raise errorclass(errno, errval)

pymysql.err.OperationalError: (1213, 'Deadlock found when trying to get lock; try restarting transaction')

The above exception was probably the direct cause of the following exception:

Traceback (most recent call last):
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/airflow/jobs/triggerer_job_runner.py", line 346, in _execute
self._run_trigger_loop()
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/airflow/jobs/triggerer_job_runner.py", line 378, in _run_trigger_loop
self.handle_events()
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/airflow/traces/tracer.py", line 58, in wrapper
return func(*args, **kwargs)
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/airflow/jobs/triggerer_job_runner.py", line 409, in handle_events
Trigger.submit_event(trigger_id=trigger_id, event=event)
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/airflow/api_internal/internal_api_call.py", line 166, in wrapper
return func(*args, **kwargs)
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/airflow/utils/session.py", line 97, in wrapper
return func(*args, session=session, **kwargs)
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/airflow/models/trigger.py", line 208, in submit_event
event.handle_submit(task_instance=task_instance)
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/airflow/utils/session.py", line 96, in wrapper
with create_session() as session:
File "/opt/cpython-3.12.4-1/lib/python3.12/contextlib.py", line 144, in exit
next(self.gen)
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/airflow/utils/session.py", line 57, in create_session
session.commit()
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 1454, in commit
self._transaction.commit(_to_root=self.future)
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 832, in commit
self._prepare_impl()
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 811, in _prepare_impl
self.session.flush()
File "/opt/gts/3pp/airflow/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 3449, in flush
self._flush(objects)
...

pymysql.err.OperationalError: (1213, 'Deadlock found when trying to get lock; try restarting transaction')

[potiuk]

Actually, the way how to find out what happens - you will have to find and provide log from the server about the deadlock - because only the server log contains details about the deadlock.

[nesp159de]

I am attaching SHOW ENGINE INNODB STATUS results. Btw on 28.1 I was able to reproduce the issue even with the provided sample. (But after two days of runs).
deadlock.txt

[nesp159de]

We try to switch: try to switch "schedule_after_task_execution" to False, proposed solution did not help.
We can still reproduce the issue with the attached example after roughly 10 hours of run.

[nesp159de]

We finally fix the issue by remove deffered infrastructure from our code at all because the issue is still there.
We replace it by inheriting from: airflow.sensors.base import BaseSensorOperator (default configuration is using active sleep).