DataprocCreateBatchOperator: Retry on 5xx errors for deferrable

[vitstransky]

Hello,
we are using Airflow 2.11, apache-airflow-providers-google 15.1.0.

We run Dataproc Serverless batches via DataprocCreateBatchOperator with defferable = True. We are getting randomly error:
grpc.aio._call.AioRpcError: <AioRpcError of RPC that terminated with: status = StatusCode.UNAVAILABLE details = "Getting metadata from plugin failed with error: ("Error code {'code': 503, 'message': 'The service is currently unavailable.', 'status': 'UNAVAILABLE'}", '{\n "error": {\n "code": 503,\n "message": "The service is currently unavailable.",\n "status": "UNAVAILABLE"\n }\n}\n')" debug_error_string = "UNKNOWN:Error received from peer {grpc_message:"Getting metadata from plugin failed with error: (\"Error code {\'code\': 503, \'message\': \'The service is currently unavailable.\', \'status\': \'UNAVAILABLE\'}\", \'{\\n \"error\": {\\n \"code\": 503,\\n \"message\": \"The service is currently unavailable.\",\\n \"status\": \"UNAVAILABLE\"\\n }\\n}\\n\')", grpc_status:14, created_time:"2026-09-24T10:12:11.505328013+02:00"}"

When I look at params of call DataprocBatchTrigger, there are no retry options passed there.

According to GCP docs:
Any client interacting with Google Cloud APIs must implement retry logic with exponential backoff for 5xx errors (specifically 500, 503 and 409")"

Could somebody please help us how to solve this? Would update to latest version of providers help?

Thank you for help.

[qoxmfaktmxj]

You're right that the trigger doesn't retry. DataprocBatchTrigger.run() just loops on hook.get_batch(...) and has no retry or exception handling around it. That's still the case on the current main of the Google provider, so upgrading alone won't change this. Also note that the operator's retry= argument only applies to the synchronous create/get calls, it's not passed to the trigger.

The error itself (Getting metadata from plugin failed ... 503) happens while the gRPC client fetches credentials for the call, so it's a transient auth/metadata hiccup rather than a problem with your batch.

The practical fix is to let Airflow's task retry handle it, and make the retry re-attach to the same batch instead of creating a new one. The operator already supports this: if create_batch raises AlreadyExists, it logs "Attaching to the job ... if it is still running" and defers again on the existing batch. For that to work, you need a fixed, deterministic batch_id:

DataprocCreateBatchOperator(
    task_id="run_batch",
    batch_id="my-job-{{ ds_nodash }}",  # stable across retries, unique per run
    batch={...},
    region="europe-west1",
    deferrable=True,
    retries=3,
    retry_delay=timedelta(minutes=1),
    retry_exponential_backoff=True,
)

With this, a 503 in the triggerer fails the attempt, Airflow retries with exponential backoff, the new attempt hits AlreadyExists, and it simply resumes waiting for the running batch.

Two things to keep in mind:

• batch_id must be 4 to 63 characters of [a-z0-9-] and unique per DAG run (hence the templated date or a sanitized run id).
• If the batch itself really failed, a retry attaches to that same failed batch and fails again. In that case you need a new batch_id (or delete the old batch) to rerun it.

If you want the trigger itself to retry transient errors, that would be a feature request for the Google provider.

[vitstransky]

Hello,

thank you for your answer and suggestion. We currently generate unique batch_id for each task run, even for retries, so this would not work for us.

The same issue was already solved for Dataflow and Cloud Run.
#66293
#67219

Would it be possible to implement the same retry logic in DataprocCreateBatchOperator?

Best regards,
Vit