Databricks workflow repair airflow3

[Beat-Nick]

Add Databricks-native retry settings to task operators

Summary

Adds first-class Databricks task retry settings to
DatabricksNotebookOperator and DatabricksTaskOperator:
max_retries, min_retry_interval_millis, and retry_on_timeout.

These are Databricks task-level retries, not Airflow task retries. Databricks
reruns the failed task attempt inside the same job run; Airflow retries rerun
the operator.

The payload shape change is gated on explicit retry configuration, so existing
standalone tasks keep their current runs/submit payload unless users opt in by
setting a Databricks retry field.

This follows the recovery-model discussion in
apache/airflow#68358: native
task retries handle transient task failures first, while workflow repair remains
separate follow-up work for run-level recovery.

Details

The retry fields live on Databricks Jobs API tasks, so the implementation sits
in DatabricksTaskBaseOperator and applies to both standalone submits and tasks
inside DatabricksWorkflowTaskGroup.

For standalone DatabricksNotebookOperator and DatabricksTaskOperator,
_get_run_json() switches to the tasks[] submit form only when a retry field
is configured through operator arguments or, for DatabricksTaskOperator,
task_config. This is required because Databricks ignores these fields at the
top level of runs/submit; they must be placed on a SubmitTask.

Monitoring becomes retry-aware only when the effective Databricks max_retries
permits another native attempt (-1 or a positive integer). In that mode:

• Standalone operators wait on the submit run, whose terminal state includes all
  Databricks retry attempts.
• Workflow task operators re-resolve the latest attempt for the same task_key
  and treat a failed attempt as final only after the parent workflow run is
  terminal.
• Deferrable workflow monitoring passes workflow_run_id and
  databricks_task_key to DatabricksExecutionTrigger, so on_kill can cancel
  the latest retry attempt instead of a stale attempt id.

Explicit settings that do not enable retries, such as max_retries=0,
retry_on_timeout=False, or min_retry_interval_millis alone, still land in
the task payload but keep existing single-attempt monitoring behavior.

Changes

• Adds retry settings to DatabricksNotebookOperator and
  DatabricksTaskOperator.
• Preserves DatabricksTaskOperator precedence: direct operator arguments
  override matching task_config fields, and the operator-managed task_key
  cannot be shadowed by task_config.
• Updates sync and deferrable monitoring to wait for the final Databricks retry
  outcome.
• Accepts WAITING_FOR_RETRY and BLOCKED as non-terminal RunState life
  cycle states.
• Adds tests for payload generation, argument precedence, sync and deferrable
  monitoring, trigger serialization, and waiting through WAITING_FOR_RETRY.

DatabricksSubmitRunOperator and DatabricksCreateJobsOperator remain raw
payload pass-through operators; users can already set per-task retry fields in
their task payloads.

Was generative AI tooling used to co-author this PR?

• Yes - Codex (GPT-5)

Generated-by: Codex (GPT-5) following the guidelines