[airflow] branch master updated: Enforce READ COMMITTED isolation when using mysql (#15714)

[po...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

potiuk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/master by this push:
     new 231d104  Enforce READ COMMITTED isolation when using mysql (#15714)
231d104 is described below

commit 231d104e37da57aa097e5f726fe6d3031ad04c52
Author: Sam Wheating <sa...@shopify.com>
AuthorDate: Sun May 9 04:45:16 2021 -0400

    Enforce READ COMMITTED isolation when using mysql (#15714)
    
    * Enforce READ COMMITTED isolation when using mysql
    
    * Fixing up tests, removing indentation
    
    * Fixing test
---
 airflow/settings.py                  | 9 +++++++++
 tests/core/test_sqlalchemy_config.py | 5 +++++
 tests/www/test_app.py                | 2 ++
 3 files changed, 16 insertions(+)

diff --git a/airflow/settings.py b/airflow/settings.py
index 0db4a0a..580aa97 100644
--- a/airflow/settings.py
+++ b/airflow/settings.py
@@ -287,6 +287,15 @@ def prepare_engine_args(disable_connection_pool=False):
         engine_args['pool_recycle'] = pool_recycle
         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'
+
     return engine_args
 
 
diff --git a/tests/core/test_sqlalchemy_config.py b/tests/core/test_sqlalchemy_config.py
index c4c909b..de62141 100644
--- a/tests/core/test_sqlalchemy_config.py
+++ b/tests/core/test_sqlalchemy_config.py
@@ -57,6 +57,7 @@ class TestSqlAlchemySettings(unittest.TestCase):
             pool_pre_ping=True,
             pool_recycle=1800,
             pool_size=5,
+            isolation_level='READ COMMITTED',
         )
 
     @patch('airflow.settings.setup_event_handlers')
@@ -75,11 +76,15 @@ class TestSqlAlchemySettings(unittest.TestCase):
         }
         with conf_vars(config):
             settings.configure_orm()
+            engine_args = {}
+            if settings.SQL_ALCHEMY_CONN.startswith('mysql'):
+                engine_args['isolation_level'] = 'READ COMMITTED'
             mock_create_engine.assert_called_once_with(
                 settings.SQL_ALCHEMY_CONN,
                 connect_args=SQL_ALCHEMY_CONNECT_ARGS,
                 poolclass=NullPool,
                 encoding='utf-8',
+                **engine_args,
             )
 
     @patch('airflow.settings.setup_event_handlers')
diff --git a/tests/www/test_app.py b/tests/www/test_app.py
index 47b8f81..2e25798 100644
--- a/tests/www/test_app.py
+++ b/tests/www/test_app.py
@@ -223,6 +223,8 @@ class TestApp(unittest.TestCase):
     def test_should_set_sqlalchemy_engine_options(self):
         app = application.cached_app(testing=True)
         engine_params = {'pool_size': 3, 'pool_recycle': 120, 'pool_pre_ping': True, 'max_overflow': 5}
+        if app.config['SQLALCHEMY_DATABASE_URI'].startswith('mysql'):
+            engine_params['isolation_level'] = 'READ COMMITTED'
         assert app.config['SQLALCHEMY_ENGINE_OPTIONS'] == engine_params
 
     @conf_vars(