Backported BigQueryExecuteQueryOperator throws 409 Already Exists in job with multiple operator instances

[rbkn]

Airflow Version: v1.10.10

Kubernetes version (if you are using kubernetes) (use kubectl version):
Client Version: version.Info{Major:"1", Minor:"19", GitVersion:"v1.19.0", GitCommit:"e19964183377d0ec2052d1f1fa930c4d7575bd50", GitTreeState:"clean", BuildDate:"2020-08-26T14:30:33Z", GoVersion:"go1.15", Compiler:"gc", Platform:"linux/amd64"} Server Version: version.Info{Major:"1", Minor:"14+", GitVersion:"v1.14.10-gke.42", GitCommit:"42bef28c2031a74fc68840fce56834ff7ea08518", GitTreeState:"clean", BuildDate:"2020-06-02T16:07:00Z", GoVersion:"go1.12.12b4", Compiler:"gc", Platform:"l inux/amd64"}

Environment:

• Cloud provider or hardware configuration: GCP
• OS (e.g. from /etc/os-release): Debian GNU/Linux 10 (buster)
• Kernel (e.g. uname -a): Linux cs-695081671381-default-default-9zgvw 4.19.112+ Improving the search functionality in the graph view #1 SMP Wed Aug 19 07:58:58 PDT 2020 x86_64 GNU/Linux
• Install tools: From official docker image
• Others:

What happened:

The job failed with the following error:
[2020-09-06 17:35:17,013] {{taskinstance.py:1145}} ERROR - 409 POST https://bigquery.googleapis.com/bigquery/v2/projects/<project_id>/jobs: Already Exists: Job <project-id>:EU.airflow_1599413716

What you expected to happen:

What I think went wrong:
It looks like when there are multiple BigQueryExecuteQueryOperator tasks in a DAG and they run for more than a very short time and end up running in parallel, there is a conflict with the job Id between the jobs.

The issue can be resolved by waiting for ~1 minute and re-running.

How to reproduce it:

I have not reproduced in minikube/kind as this appears to be more of an operator problem that can be reproduced in a non-k8s environment.

In my local envrionment, this replicates it by running 3 simple queries in parallel.
You will need:

• google_cloud_default to be set up as a connection to a valid gcp instance
• 0_dummy to be a valid bigquery dataset in that instance

from airflow.utils.dates import days_ago
from airflow.providers.google.cloud.operators.bigquery import BigQueryExecuteQueryOperator
from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator

args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': days_ago(1),
    'retries': 1,
}

dag = DAG(
    dag_id='test_bq',
    default_args=args,
    max_active_runs=1,
    schedule_interval=None,
)

start = DummyOperator(task_id="run_queries_parallel",dag=dag)

first_query = BigQueryExecuteQueryOperator(
    task_id="first_query",
    sql="""
    SELECT CURRENT_DATE() as first,2,3
    """,
    gcp_conn_id="google_cloud_default",
    destination_dataset_table="0_dummy.first_query${{ ds_nodash }}",
    write_disposition="WRITE_TRUNCATE",
    allow_large_results=True,
    use_legacy_sql=False,
    time_partitioning={"field": "first", "require_partition_filter": False},
    dag=dag,
)

second_query = BigQueryExecuteQueryOperator(
    task_id="second_query",
    sql="SELECT CURRENT_DATE() as first,5,6",
    gcp_conn_id="google_cloud_default",
    destination_dataset_table="0_dummy.second_query${{ ds_nodash }}",
    write_disposition="WRITE_TRUNCATE",
    time_partitioning={"field": "first", "require_partition_filter": False},
    allow_large_results=True,
    use_legacy_sql=False,
    dag=dag,
)

third_query = BigQueryExecuteQueryOperator(
    task_id="third_query",
    sql="SELECT CURRENT_DATE() as first,5,6",
    gcp_conn_id="google_cloud_default",
    destination_dataset_table="0_dummy.third_query${{ ds_nodash }}",
    write_disposition="WRITE_TRUNCATE",
    time_partitioning={"field": "first", "require_partition_filter": False},
    allow_large_results=True,
    use_legacy_sql=False,
    dag=dag,
)

start >> [first_query, second_query, third_query] 

The outcome of this is:

Log output for DAG

> *** Reading local file: /opt/airflow/logs/test_bq/third_query/2020-09-07T11:12:48.939603+00:00/1.log
> [2020-09-07 11:13:01,903] {taskinstance.py:670} INFO - Dependencies all met for <TaskInstance: test_bq.third_query 2020-09-07T11:12:48.939603+00:00 [queued]>
> [2020-09-07 11:13:01,994] {taskinstance.py:670} INFO - Dependencies all met for <TaskInstance: test_bq.third_query 2020-09-07T11:12:48.939603+00:00 [queued]>
> [2020-09-07 11:13:01,995] {taskinstance.py:880} INFO - 
> --------------------------------------------------------------------------------
> [2020-09-07 11:13:01,996] {taskinstance.py:881} INFO - Starting attempt 1 of 2
> [2020-09-07 11:13:01,997] {taskinstance.py:882} INFO - 
> --------------------------------------------------------------------------------
> [2020-09-07 11:13:02,059] {taskinstance.py:901} INFO - Executing <Task(BigQueryExecuteQueryOperator): third_query> on 2020-09-07T11:12:48.939603+00:00
> [2020-09-07 11:13:02,093] {standard_task_runner.py:54} INFO - Started process 4889 to run task
> [2020-09-07 11:13:02,318] {standard_task_runner.py:77} INFO - Running: ['airflow', 'run', 'test_bq', 'third_query', '2020-09-07T11:12:48.939603+00:00', '--job_id', '35', '--pool', 'default_pool', '--raw', '-sd', 'DAGS_FOLDER/test.py', '--cfg_path', '/tmp/tmp0snb5oo2']
> [2020-09-07 11:13:02,325] {standard_task_runner.py:78} INFO - Job 35: Subtask third_query
> [2020-09-07 11:13:02,526] {logging_mixin.py:112} INFO - Running %s on host %s <TaskInstance: test_bq.third_query 2020-09-07T11:12:48.939603+00:00 [running]> e26af7fa4bfb
> [2020-09-07 11:13:02,647] {bigquery.py:597} INFO - Executing: SELECT CURRENT_DATE() as first,5,6
> [2020-09-07 11:13:02,662] {logging_mixin.py:112} WARNING - /home/airflow/.local/lib/python3.7/site-packages/airflow/providers/google/cloud/hooks/bigquery.py:2029: DeprecationWarning: This method is deprecated. Please use `BigQueryHook.insert_job` method.
>   DeprecationWarning
> [2020-09-07 11:13:03,320] {taskinstance.py:1150} ERROR - 409 POST https://bigquery.googleapis.com/bigquery/v2/projects/<project>/jobs: Already Exists: Job p*** Reading local file: /opt/airflow/logs/test_bq/third_query/2020-09-07T11:12:48.939603+00:00/1.log
[2020-09-07 11:13:01,903] {taskinstance.py:670} INFO - Dependencies all met for <TaskInstance: test_bq.third_query 2020-09-07T11:12:48.939603+00:00 [queued]>
[2020-09-07 11:13:01,994] {taskinstance.py:670} INFO - Dependencies all met for <TaskInstance: test_bq.third_query 2020-09-07T11:12:48.939603+00:00 [queued]>
[2020-09-07 11:13:01,995] {taskinstance.py:880} INFO - 
--------------------------------------------------------------------------------
[2020-09-07 11:13:01,996] {taskinstance.py:881} INFO - Starting attempt 1 of 2
[2020-09-07 11:13:01,997] {taskinstance.py:882} INFO - 
--------------------------------------------------------------------------------
[2020-09-07 11:13:02,059] {taskinstance.py:901} INFO - Executing <Task(BigQueryExecuteQueryOperator): third_query> on 2020-09-07T11:12:48.939603+00:00
[2020-09-07 11:13:02,093] {standard_task_runner.py:54} INFO - Started process 4889 to run task
[2020-09-07 11:13:02,318] {standard_task_runner.py:77} INFO - Running: ['airflow', 'run', 'test_bq', 'third_query', '2020-09-07T11:12:48.939603+00:00', '--job_id', '35', '--pool', 'default_pool', '--raw', '-sd', 'DAGS_FOLDER/test.py', '--cfg_path', '/tmp/tmp0snb5oo2']
[2020-09-07 11:13:02,325] {standard_task_runner.py:78} INFO - Job 35: Subtask third_query
[2020-09-07 11:13:02,526] {logging_mixin.py:112} INFO - Running %s on host %s <TaskInstance: test_bq.third_query 2020-09-07T11:12:48.939603+00:00 [running]> e26af7fa4bfb
[2020-09-07 11:13:02,647] {bigquery.py:597} INFO - Executing: SELECT CURRENT_DATE() as first,5,6
[2020-09-07 11:13:02,662] {logging_mixin.py:112} WARNING - /home/airflow/.local/lib/python3.7/site-packages/airflow/providers/google/cloud/hooks/bigquery.py:2029: DeprecationWarning: This method is deprecated. Please use `BigQueryHook.insert_job` method.
  DeprecationWarning
[2020-09-07 11:13:03,320] {taskinstance.py:1150} ERROR - 409 POST https://bigquery.googleapis.com/bigquery/v2/projects/<project>/jobs: Already Exists: Job <project>:EU.airflow_1599477182

(job ID: airflow_1599477182)

    -----Query Job SQL Follows-----    

    |    .    |    .    |    .    |
   1:SELECT CURRENT_DATE() as first,5,6
    |    .    |    .    |    .    |

(job ID: airflow_1599477182)

    -----Query Job SQL Follows-----    

    |    .    |    .    |    .    |
   1:SELECT CURRENT_DATE() as first,5,6
    |    .    |    .    |    .    |
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 984, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/google/cloud/operators/bigquery.py", line 622, in execute
    encryption_configuration=self.encryption_configuration
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/google/cloud/hooks/bigquery.py", line 2171, in run_query
    job.result()
  File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/job.py", line 3207, in result
    super(QueryJob, self).result(retry=retry, timeout=timeout)
  File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/job.py", line 810, in result
    self._begin(retry=retry, timeout=timeout)
  File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/job.py", line 3156, in _begin
    super(QueryJob, self)._begin(client=client, retry=retry, timeout=timeout)
  File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/job.py", line 638, in _begin
    retry, method="POST", path=path, data=self.to_api_repr(), timeout=timeout
  File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/client.py", line 574, in _call_api
    return call()
  File "/home/airflow/.local/lib/python3.7/site-packages/google/api_core/retry.py", line 286, in retry_wrapped_func
    on_error=on_error,
  File "/home/airflow/.local/lib/python3.7/site-packages/google/api_core/retry.py", line 184, in retry_target
    return target()
  File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/_http.py", line 423, in api_request
    raise exceptions.from_http_response(response)
google.api_core.exceptions.Conflict: 409 POST https://bigquery.googleapis.com/bigquery/v2/projects/<project>/jobs: Already Exists: Job <project>:EU.airflow_1599477182

(job ID: airflow_1599477182)

    -----Query Job SQL Follows-----    

    |    .    |    .    |    .    |
   1:SELECT CURRENT_DATE() as first,5,6
    |    .    |    .    |    .    |

(job ID: airflow_1599477182)

    -----Query Job SQL Follows-----    

    |    .    |    .    |    .    |
   1:SELECT CURRENT_DATE() as first,5,6
    |    .    |    .    |    .    |
[2020-09-07 11:13:03,351] {taskinstance.py:1194} INFO - Marking task as UP_FOR_RETRY. dag_id=test_bq, task_id=third_query, execution_date=20200907T111248, start_date=20200907T111301, end_date=20200907T111303
[2020-09-07 11:13:06,749] {local_task_job.py:102} INFO - Task exited with return code 1:EU.airflow_1599477182
> 
> (job ID: airflow_1599477182)
> 
>     -----Query Job SQL Follows-----    
> 
>     |    .    |    .    |    .    |
>    1:SELECT CURRENT_DATE() as first,5,6
>     |    .    |    .    |    .    |
> 
> (job ID: airflow_1599477182)
> 
>     -----Query Job SQL Follows-----    
> 
>     |    .    |    .    |    .    |
>    1:SELECT CURRENT_DATE() as first,5,6
>     |    .    |    .    |    .    |
> Traceback (most recent call last):
>   File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 984, in _run_raw_task
>     result = task_copy.execute(context=context)
>   File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/google/cloud/operators/bigquery.py", line 622, in execute
>     encryption_configuration=self.encryption_configuration
>   File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/google/cloud/hooks/bigquery.py", line 2171, in run_query
>     job.result()
>   File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/job.py", line 3207, in result
>     super(QueryJob, self).result(retry=retry, timeout=timeout)
>   File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/job.py", line 810, in result
>     self._begin(retry=retry, timeout=timeout)
>   File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/job.py", line 3156, in _begin
>     super(QueryJob, self)._begin(client=client, retry=retry, timeout=timeout)
>   File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/job.py", line 638, in _begin
>     retry, method="POST", path=path, data=self.to_api_repr(), timeout=timeout
>   File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/client.py", line 574, in _call_api
>     return call()
>   File "/home/airflow/.local/lib/python3.7/site-packages/google/api_core/retry.py", line 286, in retry_wrapped_func
>     on_error=on_error,
>   File "/home/airflow/.local/lib/python3.7/site-packages/google/api_core/retry.py", line 184, in retry_target
>     return target()
>   File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/_http.py", line 423, in api_request
>     raise exceptions.from_http_response(response)
> google.api_core.exceptions.Conflict: 409 POST https://bigquery.googleapis.com/bigquery/v2/projects/<project>/jobs: Already Exists: Job <project>:EU.airflow_1599477182
> 
> (job ID: airflow_1599477182)
> 
>     -----Query Job SQL Follows-----    
> 
>     |    .    |    .    |    .    |
>    1:SELECT CURRENT_DATE() as first,5,6
>     |    .    |    .    |    .    |
> 
> (job ID: airflow_1599477182)
> 
>     -----Query Job SQL Follows-----    
> 
>     |    .    |    .    |    .    |
>    1:SELECT CURRENT_DATE() as first,5,6
>     |    .    |    .    |    .    |
> [2020-09-07 11:13:03,351] {taskinstance.py:1194} INFO - Marking task as UP_FOR_RETRY. dag_id=test_bq, task_id=third_query, execution_date=20200907T111248, start_date=20200907T111301, end_date=20200907T111303
> [2020-09-07 11:13:06,749] {local_task_job.py:102} INFO - Task exited with return code 1

Anything else we need to know:

See reproduction - can be consistently reproduced.

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[turbaszek]

@rbkn have you tried using BigQueryInsertJobOperator from backport packages? It improves the idempotency as discussed #9590 and will replace BigQueryExecuteQueryOperator operator in the future.

[rbkn]

@turbaszek - No I haven't tried that as yet since I was looking for something that was roughly equivalent to the old BigQueryOperator and having not read the documentation properly (oops) I had assumed that the BigQueryInsertJobOperator would be non-blocking and not wait for completion. I will try that now, thanks.

[turbaszek]

BigQueryInsertJobOperator would be non-blocking and not wait for completion

That's something interesting. Currently, the operator waits for the job to complete but it may be reasonable to add an option to just submit job and then use a sensor to wait for it.

[rbkn]

Well, this has fixed my issue 👍 and is a lot more flexible!
I am closing this if BigQueryExecuteQueryOperator is to be deprecated.
Should I raise a feature request regarding the non-blocking option? Not something I need,but interesting as you say.