BigQueryInsertJobOperator marked successful incorrectly

[set92]

What happened
It is true that the apache-airflow-providers-google version is 6.8.0 (apr 2022), and Airflow 2.5.0 2.4.2, so maybe updating the packages could fix the error.

The error happened with a long query (~ 3h), in which the task marked as successfully before the task really finished. In the logs we can see both started almost at the same time, but airflow thought it was already finished at 7:17 UTC, but Bigquery didn't really finish until 8:05 UTC. The query was the first step of a transactional query, in which it calculates some st_intersetcs and cross join between 2 tables.

The log from the task in Airflow:

*** Reading remote log from s3://.../dag_id=dag_name/run_id=manual__2023-05-03T14:36:11+00:00/task_id=task_1_name.insert_into_prod.insert_data_and_update_valid_dates/attempt=1.log.
[2023-05-04, 05:15:39 UTC] {taskinstance.py:1165} INFO - Dependencies all met for <TaskInstance: dag_name.task_1_name.insert_into_prod.insert_data_and_update_valid_dates manual__2023-05-03T14:36:11+00:00 [queued]>
[2023-05-04, 05:15:40 UTC] {taskinstance.py:1165} INFO - Dependencies all met for <TaskInstance: dag_name.task_1_name.insert_into_prod.insert_data_and_update_valid_dates manual__2023-05-03T14:36:11+00:00 [queued]>
[2023-05-04, 05:15:40 UTC] {taskinstance.py:1362} INFO - 
--------------------------------------------------------------------------------
[2023-05-04, 05:15:40 UTC] {taskinstance.py:1363} INFO - Starting attempt 1 of 1
[2023-05-04, 05:15:40 UTC] {taskinstance.py:1364} INFO - 
--------------------------------------------------------------------------------
[2023-05-04, 05:15:40 UTC] {taskinstance.py:1383} INFO - Executing <Task(BigQueryInsertParametrizedJobOperator): task_1_name.insert_into_prod.insert_data_and_update_valid_dates> on 2023-05-03 14:36:11+00:00
[2023-05-04, 05:15:40 UTC] {standard_task_runner.py:55} INFO - Started process 23 to run task
[2023-05-04, 05:15:40 UTC] {standard_task_runner.py:82} INFO - Running: ['***', 'tasks', 'run', 'dag_name', 'task_1_name.insert_into_prod.insert_data_and_update_valid_dates', 'manual__2023-05-03T14:36:11+00:00', '--job-id', '78337', '--raw', '--subdir', 'DAGS_FOLDER/master_dag_factory_dag_name.py', '--cfg-path', '/tmp/tmppm9tmkfj']
[2023-05-04, 05:15:40 UTC] {standard_task_runner.py:83} INFO - Job 78337: Subtask task_1_name.insert_into_prod.insert_data_and_update_valid_dates
[2023-05-04, 05:15:40 UTC] {task_command.py:376} INFO - Running <TaskInstance: dag_name.task_1_name.insert_into_prod.insert_data_and_update_valid_dates manual__2023-05-03T14:36:11+00:00 [running]> on host pod-c8ddc6bb711e4d1ab5f1a9feb9a16501
[2023-05-04, 05:15:40 UTC] {taskinstance.py:1590} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_ID=dag_name
AIRFLOW_CTX_TASK_ID=task_1_name.insert_into_prod.insert_data_and_update_valid_dates
AIRFLOW_CTX_EXECUTION_DATE=2023-05-03T14:36:11+00:00
AIRFLOW_CTX_TRY_NUMBER=1
AIRFLOW_CTX_DAG_RUN_ID=manual__2023-05-03T14:36:11+00:00
[2023-05-04, 05:15:42 UTC] {base.py:71} INFO - Using connection ID 'default_conn' for task execution.
[2023-05-04, 05:15:42 UTC] {bigquery.py:1554} INFO - Inserting job ***_transaction_dag_name_task_1_name_insert_into_prod_generate_job_id_1683124743200_1_472eb3e9b7aabcbdab449fbf7ca0ca0b
[2023-05-04, 07:16:41 UTC] {taskinstance.py:1401} INFO - Marking task as SUCCESS. dag_id=dag_name, task_id=task_1_name.insert_into_prod.insert_data_and_update_valid_dates, execution_date=20230503T143611, start_date=20230504T051539, end_date=20230504T071641
[2023-05-04, 07:16:44 UTC] {local_task_job.py:164} INFO - Task exited with return code 0

Bigquery Job information

Job ID -- XXX
User -- XXX
Location -- EU
Creation time -- May 4, 2023, 7:15:44 AM UTC+2
Start time -- May 4, 2023, 7:15:45 AM UTC+2
End time -- May 4, 2023, 10:05:20 AM UTC+2
Duration -- 2 hr 49 min
Bytes processed -- 236.05 GB
Bytes billed -- 236.05 GB
Job priority -- INTERACTIVE
Use legacy SQL -- false
Destination table -- XXX

The code used to run the operator:

        config = {
            "query": {
                "query": transaction_query,
                "useLegacySql": BigQueryToProdValues.USE_LEGACY_SQL,
                "allowLargeResults": BigQueryToProdValues.ALLOW_LARGE_RESULTS,
                "location": "eu",
            }
        }

        if self.query_params:
            config["query"]["queryParameters"] = self.query_params

        insert_data = BigQueryInsertParametrizedJobOperator(
            task_id=CommonTaskIds.INSERT_DATA_AND_UPDATE_VALID_DATES,
            task_group=self.task_group,
            gcp_conn_id=self.gcp_conn_id,
            force_rerun=False,
            reattach_states={"PENDING", "RUNNING", "DONE"},
            job_id=kwargs.pop("job_id"),
            configuration=config,
            pool=self.pool,
            trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
            **kwargs,
        )

        return insert_data, insert_data


class BigQueryInsertParametrizedJobOperator(BigQueryInsertJobOperator):
    def execute(self, context):
        query_params = self.configuration.get("query", {}).get("queryParameters", None)
        if query_params is not None:
            self.configuration["query"]["queryParameters"] = ast.literal_eval(query_params)
        super().execute(context)

Because of Airflow saying the data was already processed, the following task failed saying that the table was empty. I only had to wait for it to really finish and clear the failed tasks, but not sure why or when this can happen.

What you think should happen instead
I think the task should run until the task is finished, since in this version, there wasn't a deferrable parameter, but maybe it didn't work quite well, and it got detached at some moment, and thought it was completed because of something I'm not seeing.

How to reproduce
Didn't try to test it because it took too long to run. My idea was to ask if you knew about this or had some theory to test it further, and if not, upgrade apache-airflow-providers-google to 10.0.0 and hope it doesn't happen more, or only happens in really long task runs.

Operating System
Ubuntu 20.04

Versions of Apache Airflow Providers
apache-airflow-providers-amazon==6.0.0
apache-airflow-providers-cncf-kubernetes==4.4.0
apache-airflow-providers-common-sql==1.2.0
apache-airflow-providers-ftp==3.1.0
apache-airflow-providers-google==6.8.0
apache-airflow-providers-http==4.0.0
apache-airflow-providers-imap==3.0.0
apache-airflow-providers-postgres==5.2.2
apache-airflow-providers-sendgrid==3.0.0
apache-airflow-providers-slack==6.0.0
apache-airflow-providers-sqlite==3.2.1

[potiuk]

I do not think this is a known issue. And likely you should open a "Real issue" for - however it might be that this is simply a problem with the client (or maybe timeout is not propagated as error?). Hard to say - we have few thousands of those operators, but a chance is that if you open an issue somoene will pick it up and look closely.

[set92]

I was opening it as an Issue, but I noticed that said to open a discussion if I didn't have a reproducible case. It has happened a second time, but in another random task (which lasted 40 min, so not only in long tasks). Another coworker was looking through Bigquery logs and checking the code of the BigQueryInsertJobOperator, but he couldn't find anything, except that the last timestamp in the Google logs of getQueryResults matched the timestamp of the incorrect SUCCESSFUL. And the other thing that doesn't make sense is that we have 4 DAGs almost the same (1 per country), and so far, it has only happened in one of them.

But we are thinking of adding a Task at the start of all Taskgroups to check that their parents have the state = COMPLETED before running, and at least know when happens in PROD environment.

I'll open an issue with this, the other case we had, and what we saw in the google logs in 14h-15h.

[potiuk]

OK. Cool!

[r-richmond]

FWIW I'd wait / am waiting for #30067 to be merged. I've had several of these weird errors occur as well and until we can update to the latest SDKs I'm just chalking it up to the fact that we are stuck on old versions of the sdk.

[zdy963]

Do you have any solution for this problem?

I've encountered the very same issue. Also found that the timestamp when operator marking as success is the same as the last log of getQueryResults in BigQuery logs. The weird tasks is also a multi-statement sql with transaction. Thought it might be some bigquery python sdk bugs with detecting state of multi-statement job with transaction.