SQLExecuteQueryOperator not timing out within expected timeframe

Apache Airflow Provider(s)

common-sql

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==9.0.0 apache-airflow-providers-common-sql==1.19.0 apache-airflow-providers-mysql==5.7.3

Apache Airflow version

2.10.3

Operating System

Amazon Linux 2023

Deployment

Amazon (AWS) MWAA

Deployment details

The issue happens on a production Amazon (AWS) MWAA environment, and is also reproducible locally using the MWAA local runner (deployed on a Mac).

What happened

The SQLExecuteQueryOperator fails with timeout, but only after the query has completed.

This is the workflow currently happening: T0: Start -> T1: Run query -> T2: Expected timeout failure -> T3: Query finishes -> T4: Operator fails on timeout

What you think should happen instead

Once the execution timeout is reached, the SQLExecuteQueryOperator should kill the query and then fail.

This is the expected workflow: T0: Start -> T1: Run query -> T2: Expected timeout, operator kills query and fails

How to reproduce

The issue is reproducible both on a production Amazon (AWS) MWAA environment, and locally using the MWAA local runner (deployed on a Mac).

The DAG below contains two tasks:

1. The first one executes a long-running query on a MySQL database using the SQLExecuteQueryOperator.
2. The second one executes a long-running process using the PythonOperator.

The PythonOperator fails its execution exactly after one minute (as expected), the SQLExecuteQueryOperator takes a few minutes until the query completes (not as expected).

import os
import time

from datetime import timedelta

from airflow.decorators import dag
from airflow.models.connection import Connection
from airflow.operators.python import PythonOperator
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator


@dag(schedule=None)
def test_sql_operator_timeout_dag():

    # SQL operator test
    connection = Connection(
        conn_id="mysql_connection",
        conn_type="mysql",
        host="xyz",
        port="3306",
        schema="test",
        login="xyz",
        password="xyz",
    )

    env_key = f"AIRFLOW_CONN_{connection.conn_id.upper()}"
    connection_uri = connection.get_uri()
    os.environ[env_key] = connection_uri
    
    SQLExecuteQueryOperator(
        task_id="mysql_long_query",
        conn_id=connection.conn_id,
        sql="select benchmark(1200000000, md5('when will it end?'));",
        execution_timeout=timedelta(minutes=1),
    )


    # Python operator test
    def sleeping_function():
        time.sleep(120)

    PythonOperator(
        task_id="python_long_task",
        python_callable=sleeping_function,
        execution_timeout=timedelta(minutes=1),
    )

test_sql_operator_timeout_dag = test_sql_operator_timeout_dag()

Anything else

No response

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