Support max retries in Azure Batch operator - #73071

Thanks for picking this up, it has the right shape. A few things before a maintainer looks at it:

1. retry_total=None must not reach the client. With the new default the operator builds the hook with retry_total=None, and the hook passes that straight to BatchClient(...). azure-core takes an explicit None over its own default: the retry policy ends up with total_retries = None, and its increment() runs settings["total"] -= 1 on the first retry, which raises TypeError (checked against azure-core 1.41.0). So every transient error would fail immediately instead of being retried, for everyone who leaves the parameter unset. Pass the kwarg only when it is set, for example:

   client_kwargs = {} if self.retry_total is None else {"retry_total": self.retry_total}
   batch_client = BatchClient(endpoint=batch_account_url, credential=credential, **client_kwargs)

   and add a test asserting that BatchClient is called without retry_total when the hook default is used.

2. closes: #72982 should be related: #72982. That issue tracks several parameters, and merging this as is would auto-close it.

3. Docstrings. The operator docstring still says "Default is 3" for batch_max_retries (operators/batch.py, line 70). With None the Azure SDK default applies, which is 10 retries, and it is worth stating that this keeps today's behaviour for anyone who has not set the parameter. AzureBatchHook gained retry_total, but the class docstring (hooks/batch.py, line 45) does not list it.

4. The Static checks failure is the commit message. Airflow rejects Conventional Commits prefixes, so feat: support max retries in Azure Batch operator has to become plain prose, for example Honor batch_max_retries in AzureBatchOperator. git commit --amend, then force-push the branch.

5. Optional: a test on the operator side that batch_max_retries reaches the hook (op.hook.retry_total == 5, or patch AzureBatchHook in the operator module and assert its kwargs), so the wiring is covered end to end.

​

Thanks, re-wiring batch_max_retries after the 15.x SDK migration is the right idea, and forwarding retry_total only when it's set keeps 0 working. Two things before this can go in:

1. Deferrable mode still ignores it. AzureBatchTrigger.run() builds its own AzureBatchHook(azure_batch_conn_id=...) without retry_total, and the trigger neither takes nor serializes a retry setting. So once the task defers, all the task-completion polling in the triggerer, which is most of the job's runtime, uses the SDK default. With deferrable=True, batch_max_retries=0, only the pool/job/task creation in execute() honours the 0. Could you add a retry argument to the trigger's __init__ and serialize(), pass it from the operator's defer() call, and cover it with a test?
2. The default quietly changes from 3 to the SDK default. Before #71071, execute() applied batch_max_retries (documented "Default is 3") to the client's retry policy. With None, a Dag that never set it now gets the SDK default of 10 retries. Either keep 3 as the default, or keep None and update the docstring plus providers/microsoft/azure/docs/changelog.rst so users see the change.

Drafted-by: Claude Code (Opus 5.5); reviewed by @potiuk before posting