This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new ed918278f2f Use READ COMMITTED for async MySQL metadata sessions
(#76694)
ed918278f2f is described below
commit ed918278f2f09dc729a9fa3a45ad5ffba9a6b549
Author: Dev-iL <[email protected]>
AuthorDate: Thu Nov 12 20:37:26 2026 +0200
Use READ COMMITTED for async MySQL metadata sessions (#76694)
The async engine never got the isolation level the sync engine sets for
MySQL, so its sessions ran at the server default, REPEATABLE READ. A
query repeated in one async transaction could then read a stale snapshot
that the sync engine's sessions would not, and task heartbeats, audit
log rows and other writes that already go through the async engine
behaved differently from their sync counterparts.
---
airflow-core/src/airflow/settings.py | 24 +++++++++++++++++-------
airflow-core/tests/unit/core/test_settings.py | 12 ++++++++++++
2 files changed, 29 insertions(+), 7 deletions(-)
diff --git a/airflow-core/src/airflow/settings.py
b/airflow-core/src/airflow/settings.py
index f5a2c460268..d049df5520f 100644
--- a/airflow-core/src/airflow/settings.py
+++ b/airflow-core/src/airflow/settings.py
@@ -499,6 +499,7 @@ def _configure_async_session() -> None:
engine_args["pool_recycle"] = conf.getint("database",
"SQL_ALCHEMY_POOL_RECYCLE", fallback=1800)
engine_args["pool_pre_ping"] = conf.getboolean("database",
"SQL_ALCHEMY_POOL_PRE_PING", fallback=True)
engine_args["max_overflow"] = conf.getint("database",
"SQL_ALCHEMY_MAX_OVERFLOW", fallback=10)
+ engine_args.update(_get_isolation_level_args(SQL_ALCHEMY_CONN_ASYNC))
async_engine = create_async_metadata_engine(
SQL_ALCHEMY_CONN_ASYNC,
@@ -673,17 +674,26 @@ def prepare_engine_args(disable_connection_pool=False,
pool_class=None):
engine_args["pool_pre_ping"] = pool_pre_ping
engine_args["max_overflow"] = max_overflow
- # The default isolation level for MySQL (REPEATABLE READ) can introduce
inconsistencies when
- # running multiple schedulers, as repeated queries on the same session may
read from stale snapshots.
- # 'READ COMMITTED' is the default value for PostgreSQL.
- # More information here:
- #
https://dev.mysql.com/doc/refman/8.0/en/innodb-transaction-isolation-levels.html
- if SQL_ALCHEMY_CONN.startswith("mysql"):
- engine_args["isolation_level"] = "READ COMMITTED"
+ engine_args.update(_get_isolation_level_args(SQL_ALCHEMY_CONN))
return engine_args
+def _get_isolation_level_args(sql_alchemy_conn: str) -> dict[str, str]:
+ """
+ Return the isolation level engine argument for the metadata database
connection.
+
+ The default isolation level for MySQL (REPEATABLE READ) can introduce
inconsistencies when
+ running multiple schedulers, as repeated queries on the same session may
read from stale snapshots.
+ 'READ COMMITTED' is the default value for PostgreSQL.
+ More information here:
+
https://dev.mysql.com/doc/refman/8.0/en/innodb-transaction-isolation-levels.html
+ """
+ if sql_alchemy_conn.startswith("mysql"):
+ return {"isolation_level": "READ COMMITTED"}
+ return {}
+
+
def dispose_orm(do_log: bool = True):
"""Properly close pooled database connections."""
global Session, engine, NonScopedSession, async_engine, AsyncSession
diff --git a/airflow-core/tests/unit/core/test_settings.py
b/airflow-core/tests/unit/core/test_settings.py
index 28f9d074578..962aa08ad57 100644
--- a/airflow-core/tests/unit/core/test_settings.py
+++ b/airflow-core/tests/unit/core/test_settings.py
@@ -784,6 +784,18 @@ def
test_async_pool_is_closed_before_process_shutdown(driver):
pytest.param(
"mysql+asyncmy", "SELECT @@session.time_zone", "+00:00",
marks=pytest.mark.backend("mysql")
),
+ pytest.param(
+ "mysql+aiomysql",
+ "SELECT @@session.transaction_isolation",
+ "READ-COMMITTED",
+ marks=pytest.mark.backend("mysql"),
+ ),
+ pytest.param(
+ "mysql+asyncmy",
+ "SELECT @@session.transaction_isolation",
+ "READ-COMMITTED",
+ marks=pytest.mark.backend("mysql"),
+ ),
pytest.param("sqlite+aiosqlite", "PRAGMA foreign_keys", 1,
marks=pytest.mark.backend("sqlite")),
],
)
