<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.3</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">stuck in the middle of <code class="notranslate">airflow db upgrade</code> in one of my Airflow’s clusters.<br>
I am trying to upgrade from <code class="notranslate">2.2.3</code> --&gt; <code class="notranslate">2.3.3</code>, using MySql.<br>
Upgrade failed in <code class="notranslate">Running upgrade c97c2ab6aa23 -&gt; 4eaab2fe6582, Migrate RTIF to use run_id and map_index</code>.</p>
<p dir="auto">Actually, it also failed before that (on previous steps) and I had to run some commands manually to proceed until this point. Here is what I have done so far:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="USE airflow;
ALTER TABLE task_reschedule
ADD CONSTRAINT `task_reschedule_ti_fkey` FOREIGN KEY (`dag_id`, `task_id`, `run_id`) REFERENCES `task_instance` (`dag_id`, `task_id`, `run_id`) ON DELETE CASCADE;
ALTER TABLE task_reschedule
ADD KEY `idx_task_reschedule_dag_task_run` (`dag_id`,`task_id`,`run_id`);
ALTER TABLE `task_instance` 
ADD PRIMARY KEY (`dag_id`,`task_id`,`run_id`);
ALTER TABLE task_instance DROP COLUMN map_index;
ALTER TABLE task_reschedule DROP COLUMN map_index;
ALTER TABLE rendered_task_instance_fields DROP COLUMN map_index;
ALTER TABLE task_instance MODIFY `task_id` VARCHAR(250) CHARACTER SET utf8 COLLATE utf8_bin;
ALTER TABLE task_reschedule MODIFY `task_id` VARCHAR(250) CHARACTER SET utf8 COLLATE utf8_bin;
ALTER TABLE rendered_task_instance_fields MODIFY `dag_id` VARCHAR(250) CHARACTER SET utf8 COLLATE utf8_bin;"><pre class="notranslate"><code class="notranslate">USE airflow;
ALTER TABLE task_reschedule
ADD CONSTRAINT `task_reschedule_ti_fkey` FOREIGN KEY (`dag_id`, `task_id`, `run_id`) REFERENCES `task_instance` (`dag_id`, `task_id`, `run_id`) ON DELETE CASCADE;
ALTER TABLE task_reschedule
ADD KEY `idx_task_reschedule_dag_task_run` (`dag_id`,`task_id`,`run_id`);
ALTER TABLE `task_instance` 
ADD PRIMARY KEY (`dag_id`,`task_id`,`run_id`);
ALTER TABLE task_instance DROP COLUMN map_index;
ALTER TABLE task_reschedule DROP COLUMN map_index;
ALTER TABLE rendered_task_instance_fields DROP COLUMN map_index;
ALTER TABLE task_instance MODIFY `task_id` VARCHAR(250) CHARACTER SET utf8 COLLATE utf8_bin;
ALTER TABLE task_reschedule MODIFY `task_id` VARCHAR(250) CHARACTER SET utf8 COLLATE utf8_bin;
ALTER TABLE rendered_task_instance_fields MODIFY `dag_id` VARCHAR(250) CHARACTER SET utf8 COLLATE utf8_bin;
</code></pre></div>
<p dir="auto">The log in this step:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade c97c2ab6aa23 -&gt; 4eaab2fe6582, Migrate RTIF to use run_id and map_index
Traceback (most recent call last):
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1705, in _execute_context
    self.dialect.do_execute(
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/default.py&quot;, line 716, in do_execute
    cursor.execute(statement, parameters)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py&quot;, line 148, in execute
    result = self._query(query)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py&quot;, line 310, in _query
    conn.query(q)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py&quot;, line 548, in query
    self._affected_rows = self._read_query_result(unbuffered=unbuffered)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py&quot;, line 775, in _read_query_result
    result.read()
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py&quot;, line 1156, in read
    first_packet = self.connection._read_packet()
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py&quot;, line 725, in _read_packet
    packet.raise_for_error()
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/protocol.py&quot;, line 221, in raise_for_error
    err.raise_mysql_exception(self._data)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/err.py&quot;, line 143, in raise_mysql_exception
    raise errorclass(errno, errval)
pymysql.err.OperationalError: (1060, &quot;Duplicate column name 'run_id'&quot;)

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

Traceback (most recent call last):
  File &quot;/home/airflow/.local/bin/airflow&quot;, line 8, in &lt;module&gt;
    sys.exit(main())
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/__main__.py&quot;, line 38, in main
    args.func(args)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/cli_parser.py&quot;, line 51, in command
    return func(*args, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 99, in wrapper
    return f(*args, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/db_command.py&quot;, line 82, in upgradedb
    db.upgradedb(to_revision=to_revision, from_revision=from_revision, show_sql_only=args.show_sql_only)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 71, in wrapper
    return func(*args, session=session, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/db.py&quot;, line 1463, in upgradedb
    command.upgrade(config, revision=to_revision or 'heads')
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/alembic/command.py&quot;, line 322, in upgrade
    script.run_env()
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/alembic/script/base.py&quot;, line 569, in run_env
    util.load_python_file(self.dir, &quot;env.py&quot;)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/alembic/util/pyfiles.py&quot;, line 94, in load_python_file
    module = load_module_py(module_id, path)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/alembic/util/pyfiles.py&quot;, line 110, in load_module_py
    spec.loader.exec_module(module)  # type: ignore
  File &quot;&lt;frozen importlib._bootstrap_external&gt;&quot;, line 850, in exec_module
  File &quot;&lt;frozen importlib._bootstrap&gt;&quot;, line 228, in _call_with_frames_removed
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/migrations/env.py&quot;, line 107, in &lt;module&gt;
    run_migrations_online()
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/migrations/env.py&quot;, line 101, in run_migrations_online
    context.run_migrations()
  File &quot;&lt;string&gt;&quot;, line 8, in run_migrations
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/alembic/runtime/environment.py&quot;, line 853, in run_migrations
    self.get_context().run_migrations(**kw)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/alembic/runtime/migration.py&quot;, line 623, in run_migrations
    step.migration_fn(**kw)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/migrations/versions/0104_2_3_0_migrate_rtif_to_use_run_id_and_map_index.py&quot;, line 118, in upgrade
    batch_op.add_column(sa.Column('run_id', type_=StringID(), nullable=True))
  File &quot;/usr/local/lib/python3.9/contextlib.py&quot;, line 126, in __exit__
    next(self.gen)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/alembic/operations/base.py&quot;, line 376, in batch_alter_table
    impl.flush()
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/alembic/operations/batch.py&quot;, line 111, in flush
    fn(*arg, **kw)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/alembic/ddl/impl.py&quot;, line 322, in add_column
    self._exec(base.AddColumn(table_name, column, schema=schema))
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/alembic/ddl/impl.py&quot;, line 195, in _exec
    return conn.execute(construct, multiparams)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1200, in execute
    return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/sql/ddl.py&quot;, line 77, in _execute_on_connection
    return connection._execute_ddl(
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1290, in _execute_ddl
    ret = self._execute_context(
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1748, in _execute_context
    self._handle_dbapi_exception(
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1929, in _handle_dbapi_exception
    util.raise_(
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/util/compat.py&quot;, line 211, in raise_
    raise exception
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1705, in _execute_context
    self.dialect.do_execute(
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/default.py&quot;, line 716, in do_execute
    cursor.execute(statement, parameters)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py&quot;, line 148, in execute
    result = self._query(query)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py&quot;, line 310, in _query
    conn.query(q)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py&quot;, line 548, in query
    self._affected_rows = self._read_query_result(unbuffered=unbuffered)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py&quot;, line 775, in _read_query_result
    result.read()
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py&quot;, line 1156, in read
    first_packet = self.connection._read_packet()
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py&quot;, line 725, in _read_packet
    packet.raise_for_error()
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/protocol.py&quot;, line 221, in raise_for_error
    err.raise_mysql_exception(self._data)
  File &quot;/home/airflow/.local/lib/python3.9/site-packages/pymysql/err.py&quot;, line 143, in raise_mysql_exception
    raise errorclass(errno, errval)
sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (1060, &quot;Duplicate column name 'run_id'&quot;)
[SQL: ALTER TABLE rendered_task_instance_fields ADD COLUMN run_id VARCHAR(250) COLLATE utf8mb3_bin]
(Background on this error at: http://sqlalche.me/e/14/e3q8)"><pre class="notranslate"><code class="notranslate">INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade c97c2ab6aa23 -&gt; 4eaab2fe6582, Migrate RTIF to use run_id and map_index
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_context
    self.dialect.do_execute(
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute
    cursor.execute(statement, parameters)
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py", line 148, in execute
    result = self._query(query)
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py", line 310, in _query
    conn.query(q)
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 548, in query
    self._affected_rows = self._read_query_result(unbuffered=unbuffered)
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 775, in _read_query_result
    result.read()
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 1156, in read
    first_packet = self.connection._read_packet()
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 725, in _read_packet
    packet.raise_for_error()
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/protocol.py", line 221, in raise_for_error
    err.raise_mysql_exception(self._data)
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/err.py", line 143, in raise_mysql_exception
    raise errorclass(errno, errval)
pymysql.err.OperationalError: (1060, "Duplicate column name 'run_id'")

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

Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 8, in &lt;module&gt;
    sys.exit(main())
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/__main__.py", line 38, in main
    args.func(args)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/cli_parser.py", line 51, in command
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/cli.py", line 99, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/db_command.py", line 82, in upgradedb
    db.upgradedb(to_revision=to_revision, from_revision=from_revision, show_sql_only=args.show_sql_only)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/session.py", line 71, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/db.py", line 1463, in upgradedb
    command.upgrade(config, revision=to_revision or 'heads')
  File "/home/airflow/.local/lib/python3.9/site-packages/alembic/command.py", line 322, in upgrade
    script.run_env()
  File "/home/airflow/.local/lib/python3.9/site-packages/alembic/script/base.py", line 569, in run_env
    util.load_python_file(self.dir, "env.py")
  File "/home/airflow/.local/lib/python3.9/site-packages/alembic/util/pyfiles.py", line 94, in load_python_file
    module = load_module_py(module_id, path)
  File "/home/airflow/.local/lib/python3.9/site-packages/alembic/util/pyfiles.py", line 110, in load_module_py
    spec.loader.exec_module(module)  # type: ignore
  File "&lt;frozen importlib._bootstrap_external&gt;", line 850, in exec_module
  File "&lt;frozen importlib._bootstrap&gt;", line 228, in _call_with_frames_removed
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/migrations/env.py", line 107, in &lt;module&gt;
    run_migrations_online()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/migrations/env.py", line 101, in run_migrations_online
    context.run_migrations()
  File "&lt;string&gt;", line 8, in run_migrations
  File "/home/airflow/.local/lib/python3.9/site-packages/alembic/runtime/environment.py", line 853, in run_migrations
    self.get_context().run_migrations(**kw)
  File "/home/airflow/.local/lib/python3.9/site-packages/alembic/runtime/migration.py", line 623, in run_migrations
    step.migration_fn(**kw)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/migrations/versions/0104_2_3_0_migrate_rtif_to_use_run_id_and_map_index.py", line 118, in upgrade
    batch_op.add_column(sa.Column('run_id', type_=StringID(), nullable=True))
  File "/usr/local/lib/python3.9/contextlib.py", line 126, in __exit__
    next(self.gen)
  File "/home/airflow/.local/lib/python3.9/site-packages/alembic/operations/base.py", line 376, in batch_alter_table
    impl.flush()
  File "/home/airflow/.local/lib/python3.9/site-packages/alembic/operations/batch.py", line 111, in flush
    fn(*arg, **kw)
  File "/home/airflow/.local/lib/python3.9/site-packages/alembic/ddl/impl.py", line 322, in add_column
    self._exec(base.AddColumn(table_name, column, schema=schema))
  File "/home/airflow/.local/lib/python3.9/site-packages/alembic/ddl/impl.py", line 195, in _exec
    return conn.execute(construct, multiparams)
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1200, in execute
    return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/sql/ddl.py", line 77, in _execute_on_connection
    return connection._execute_ddl(
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1290, in _execute_ddl
    ret = self._execute_context(
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1748, in _execute_context
    self._handle_dbapi_exception(
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1929, in _handle_dbapi_exception
    util.raise_(
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_context
    self.dialect.do_execute(
  File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute
    cursor.execute(statement, parameters)
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py", line 148, in execute
    result = self._query(query)
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py", line 310, in _query
    conn.query(q)
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 548, in query
    self._affected_rows = self._read_query_result(unbuffered=unbuffered)
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 775, in _read_query_result
    result.read()
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 1156, in read
    first_packet = self.connection._read_packet()
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 725, in _read_packet
    packet.raise_for_error()
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/protocol.py", line 221, in raise_for_error
    err.raise_mysql_exception(self._data)
  File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/err.py", line 143, in raise_mysql_exception
    raise errorclass(errno, errval)
sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (1060, "Duplicate column name 'run_id'")
[SQL: ALTER TABLE rendered_task_instance_fields ADD COLUMN run_id VARCHAR(250) COLLATE utf8mb3_bin]
(Background on this error at: http://sqlalche.me/e/14/e3q8)
</code></pre></div>
<p dir="auto">The problem is that the table’s schema now looks like:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="CREATE TABLE `rendered_task_instance_fields` (
  `dag_id` varchar(250) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL,
  `task_id` varchar(250) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL,
  `rendered_fields` json NOT NULL,
  `k8s_pod_yaml` json DEFAULT NULL,
  `run_id` varchar(250) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL,
  PRIMARY KEY (`dag_id`,`task_id`,`run_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;"><pre class="notranslate"><code class="notranslate">CREATE TABLE `rendered_task_instance_fields` (
  `dag_id` varchar(250) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL,
  `task_id` varchar(250) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL,
  `rendered_fields` json NOT NULL,
  `k8s_pod_yaml` json DEFAULT NULL,
  `run_id` varchar(250) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL,
  PRIMARY KEY (`dag_id`,`task_id`,`run_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
</code></pre></div>
<p dir="auto">Seems like the <code class="notranslate">execution_date</code> was dropped from <code class="notranslate">rendered_task_instance_fields</code> table. I don't see this column now.<br>
To ensure it happened during the migration, I tried to bring up Airflow 2.2.5 in docker - and I see the  <code class="notranslate">execution_date</code>. Then, I tried to bring up Airflow 2.3.3 in docker - and I DON'T see this column.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">The migration should succeed smoothly as happened in the previous Airflow upgrades.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">I guess you should have data in MySql tables created by version 2.2.3 (and before, we are running Airflow since 1.10.x) and try to upgrade to 2.3.3</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">using docker image - apache/airflow:2.3.3-python3.9</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 3rd-party Helm chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">running MySql 8 in RDS (AWS)</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>
<p dir="auto"><strong>Apache Airflow version</strong>: 1.7.1.2</p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>):</p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>:</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>:<br>
<strong>What happened</strong>:</li>
</ul>
<p dir="auto">Environment setup<br>
We are running Airflow 1.7.1.2 with MySQL 5.6. The <code class="notranslate">wait_timeout</code> of MySQL is set at 300 seconds, which means that idle connections will go away after 300 seconds of inactivity. To reflect this, we set SQLAlchemy's <code class="notranslate">pool_recycle</code> in <code class="notranslate">airflow.cfg</code> to 290 seconds, which should force Airflow/SQLAlchemy to recycle/discard connections after 290 seconds. Thus Airflow shouldn't try to use an already-dead connection.<br>
Symptom<br>
When running a PythonOperator that takes more than 300 seconds to execute, the task would finish executing the Python callable, but ends up with error: <a href="https://gist.github.com/garthcn/cd7bcdec12748406506f2b0710655c8b">https://gist.github.com/garthcn/cd7bcdec12748406506f2b0710655c8b</a><br>
It seems that after the Python callable finishes executing, Airflow tries to push its return value to XCom. However, the SQL connection has gone away while Airflow/SQLAlchemy think it's still there.<br>
Hypothesis<br>
I did some investigation and think that it might be caused by not calling <code class="notranslate">session.commit()</code> or <code class="notranslate">session.close()</code> for the DB operations before the XCom push. As far as I know, in SQLAlchemy, if you don't close a connection and let it be in <code class="notranslate">checked-out</code> state, it won't be recycled by connection pool, and thus SQLAlchemy will try to use it again after &gt;300 seconds (which is the wait_timeout for MySQL in our case). This will result in a "MySQL connection has gone away" issue.</p>
<p dir="auto">It seems that Airflow codebase uses @provide_context decorator to help with session open/close, and my hunch is that some functions are not using it or misusing it.</p>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<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">Moved here from <a href="https://issues.apache.org/jira/browse/AIRFLOW-405" rel="nofollow">https://issues.apache.org/jira/browse/AIRFLOW-405</a></p>