Google Dataflow provider does not retry on service 503 errors

[pmcquighan-camus]

Apache Airflow Provider(s)

google

Versions of Apache Airflow Providers

apache-airflow-providers-google==18.0.0

Apache Airflow version

3.1.0

Operating System

Debian

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

What happened

The provider was polling for job status every 30 seconds for about 15-minutes as expected. However, the Dataflow API then returned a 503 - Service Unavailable error, and the task then failed. Retries of the task also failed since this was run in deferrable mode and so the task retries saw the trigger had completed with an exception, and then also failed.

The Google Dataflow API seems to return 503s with some regularity (weekly) with our current workload resulting in a failed DAG run every several days (that can be retried by clearing out the full task state, but is wasteful since the dataflow job likely already succeeded but any needed XCom values are not output)

Initial try:

[2025-10-26 22:39:17] ERROR - Exception occurred while checking for job completion. source=airflow.providers.google.cloud.triggers.dataflow.TemplateJobStartTrigger loc=dataflow.py:149
ServiceUnavailable: 503 The service is currently unavailable.
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/google/cloud/triggers/dataflow.py", line 113 in run
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/google/cloud/hooks/dataflow.py", line 1480 in get_job_status
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/google/cloud/hooks/dataflow.py", line 1457 in get_job
File "/home/airflow/.local/lib/python3.12/site-packages/google/cloud/dataflow_v1beta3/services/jobs_v1_beta3/async_client.py", line 478 in get_job
File "/home/airflow/.local/lib/python3.12/site-packages/google/api_core/grpc_helpers_async.py", line 88 in __await__
AioRpcError: <AioRpcError of RPC that terminated with:
	status = StatusCode.UNAVAILABLE
	details = "The service is currently unavailable."
	debug_error_string = "UNKNOWN:Error received from peer ipv4:74.125.132.95:443 {created_time:"2025-10-27T05:39:17.82102088+00:00", grpc_status:14, grpc_message:"The service is currently unavailable."}"
>
File "/home/airflow/.local/lib/python3.12/site-packages/google/api_core/grpc_helpers_async.py", line 85 in __await__
File "/home/airflow/.local/lib/python3.12/site-packages/grpc/aio/_interceptor.py", line 472 in __await__
File "/home/airflow/.local/lib/python3.12/site-packages/grpc/aio/_call.py", line 327 in __await__

Task retry a few minutes later:

[2025-10-26 22:42:11] DEBUG - Result from 'on_task_instance_running': [] source=airflow.listeners.listener loc=listener.py:42
[2025-10-26 22:42:11] INFO - status: error, msg: 503 The service is currently unavailable. source=airflow.task.operators.airflow.providers.google.cloud.operators.dataflow.DataflowStartFlexTemplateOperator loc=dataflow.py:648
[2025-10-26 22:42:11] ERROR - Task failed with exception source=task loc=task_runner.py:972
AirflowException: 503 The service is currently unavailable.
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 920 in run
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1307 in _execute_task
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 1632 in resume_execution
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/google/cloud/operators/dataflow.py", line 649 in execute_complete
[2025-10-26 22:42:11] ERROR - Top level error source=task loc=task_runner.py:1457
ValueError: dictionary update sequence element #0 has length 1; 2 is required
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1452 in main
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1390 in finalize
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/google/cloud/links/base.py", line 112 in get_link
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/google/cloud/links/base.py", line 92 in get_config
[2025-10-26 22:42:11] WARNING - Process exited abnormally exit_code=1 source=task

What you think should happen instead

Since the 503 error is a retryable error, the trigger should retry several times or perhaps indefinitely, before failing, just as if the job were still running.

How to reproduce

Run a dataflow flex template via the DataflowStartFlexTemplateOperator, if the google dataflow API returns a 503 while the trigger is polling for status the error will recur.

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

[jroachgolf84]

@pmcquighan-camus, just my two cents, but I think that we should stick to Airflow retries for something like this. When you add retries within the logic in .execute(), it can cause some general confusion/hinder understanding as to what's actually going on.

Let's say I'm a new user, and I only ever want my job to retry 3 times. I'd set my retries=3 at the Task-level. Now, unknown to me, there is logic in the operator that retries 10 times without the Task failing. This would be unintended behavior.

[pmcquighan-camus]

@pmcquighan-camus, just my two cents, but I think that we should stick to Airflow retries for something like this. When you add retries within the logic in .execute(), it can cause some general confusion/hinder understanding as to what's actually going on.

Let's say I'm a new user, and I only ever want my job to retry 3 times. I'd set my retries=3 at the Task-level. Now, unknown to me, there is logic in the operator that retries 10 times without the Task failing. This would be unintended behavior.

that makes sense to me in general. in this specific instance, the actual dataflow job was only ever launched/tried 1 time, and then the task failed while polling for completion status in the trigger. airflow did 2 retries, but it seemed like they both immediately failed as the trigger was marked as "failed" from the first attempt, and the 2nd/3rd attempts just started executing with on_complete and failed immediately here. my understanding of what happened here is the 3 airflow task attempts resulted in 1 dataflow job being executed, and all 3 task attempts failed from 1 service-level 503 when polling for status.

i feel that the 2nd/3rd tries should attempt to re-run the dataflow job (or at least retry querying for status of the job, since the dataflow job did in fact continue executing to completion and it would be surprising to have multiple dataflow jobs running for the airflow task)

[olegkachur-e]

Hi @pmcquighan-camus!

Thank you for reporting this issue.
I spotted in the logs two operators: TemplateJobStartTrigger and DataflowStartFlexTemplateOperator (as a retry?).

Can you please clarify your usage? A sample dag for reproduction is also highly appreciated.

[pmcquighan-camus]

This can happen from any dag using the DataflowStartFlexTemplateOperator (possibly only when in deferrable mode?). The log from TemplateJobStartTrigger is from the triggerer while waiting for the job (launched by DataflowStartFlexTemplateOperator) to complete. The triggerer marks the task as failed, and then on retries the DataflowStartFlexTemplateOperator tries to resume executing, sees that the job was marked as failed and dies again.

A sample task is defined like this, but it's not super useful without having a flex template defined in your GCP project:

DataflowStartFlexTemplateOperator(
        task_id="mytask",
        body={
            "launchParameter": {
                "containerSpecGcsPath": "gs://<some gcs bucket>/templates/<a flex template>",  # Need a dataflow flex template defined
                "environment": {}, # Any job-specific parameters needed here like workerRegion
                "jobName": "sample-job",
                "parameters": {},  # Any params here
            },
        },
        location="<location>",
        project_id="<project id>",
        deferrable=True,
        append_job_name=True,  # Add unique suffix to job names, so retries on a file will create unique names
    )

Since this failure case depends on GCP throwing 503's it's not very easy to replicate. The trigger catches the 503 exception and sets a TriggerEvent of an error here, which is where doing something like check the exception for a 503 from the service provider, and if that's the case, continue looping like it does when the job is still running

[olegkachur-e]

Hi, @pmcquighan-camus

I did some tests and took a closer look for the logs you provided, the retry error ValueError: dictionary update sequence element #0 has length 1; 2 is required is linked to the links construction problem, that was fixed a while ago in the google-provider >= 18.1.0. (#55821)

In case it is fixed, the task retry should work.

Can you please try with a newer version of google-provider and share the results?

[pmcquighan-camus]

Hi @olegkachur-e - Thanks, I will give that a try. Since this particular error occurs with GCP returns 503's it is hard to predict when it might occur again but will see if this makes things a little more stable.

[shahar1]

I'm closing this issue as non-reproducible. If you encounter this issue again after trying the newest version and manage to find how to reproduce it, please comment (without clear reproduction steps there's not too much that we can do).

[pmcquighan-camus]

Hi - we have been using apache-airflow-providers-google==19.3.0 for the past several weeks and the issue recurred:

[2026-02-02 04:42:52] ERROR - Exception occurred while checking for job completion. source=airflow.providers.google.cloud.triggers.dataflow.TemplateJobStartTrigger loc=dataflow.py:149
ServiceUnavailable: 503 Visibility check was unavailable. Please retry the request and contact support if the problem persists
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/google/cloud/triggers/dataflow.py", line 113 in run
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/google/cloud/hooks/dataflow.py", line 1480 in get_job_status
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/google/cloud/hooks/dataflow.py", line 1457 in get_job
File "/home/airflow/.local/lib/python3.12/site-packages/google/cloud/dataflow_v1beta3/services/jobs_v1_beta3/async_client.py", line 478 in get_job
File "/home/airflow/.local/lib/python3.12/site-packages/google/api_core/grpc_helpers_async.py", line 88 in __await__
AioRpcError: <AioRpcError of RPC that terminated with:
	status = StatusCode.UNAVAILABLE
	details = "Visibility check was unavailable. Please retry the request and contact support if the problem persists"
	debug_error_string = "UNKNOWN:Error received from peer ipv4:74.125.126.95:443 {created_time:"2026-02-02T12:42:52.166480427+00:00", grpc_status:14, grpc_message:"Visibility check was unavailable. Please retry the request and contact support if the problem persists"}"
>
File "/home/airflow/.local/lib/python3.12/site-packages/google/api_core/grpc_helpers_async.py", line 85 in __await__
File "/home/airflow/.local/lib/python3.12/site-packages/grpc/aio/_interceptor.py", line 472 in __await__
File "/home/airflow/.local/lib/python3.12/site-packages/grpc/aio/_call.py", line 327 in __await__

The task retry error from the initial bug report ValueError: dictionary update sequence element #0 has length 1; 2 is required no longer occurs but as expected the task is still just considered failed since the triggerer marked the task as failed, and the dataflow job is not actually retried (nor does the triggerer or any component attempt to get the status a second time)

[2026-02-02 04:45:10] ERROR - Task failed with exception source=task loc=task_runner.py:972
AirflowException: 503 Visibility check was unavailable. Please retry the request and contact support if the problem persists
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 920 in run
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1307 in _execute_task
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 1632 in resume_execution
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/google/cloud/operators/dataflow.py", line 650 in execute_complete

manage to find how to reproduce it, please comment (without clear reproduction steps there's not too much that we can do).

it is dependent on the GCP dataflow api returning a 503 - the only way to reliably reproduce it would be with a mock http server that the gcp client connects to and having it return a 503 (or by mocking out the JobsV1Beta3AsyncClient.get_job method to throw the AioRpcError in the stacktrace above). i don't have permissions to reopen this issue but it is definitely a still a bug with the retry logic in the provider