DatabricksRunNowOperator: Repair run doesnt pass job parameters and a retry on a job with an idempotency token results always results in succes.

[roydobbe]

Under which category would you file this issue?

Providers

Apache Airflow version

2.11.1

What happened and how to reproduce it?

Repair run:

Whenever a repair run is triggered, the job parameters need to be passed to the api endpoint for it to register the params for the tasks again.

I tried replicating the operator logic locally, but only when i explicitly passed the job parameters to the repair endpoint, I saw the parameters for the task it was repairing resolved.

Currently the operator only updates the job parameters as an intermediate step before triggering the repair endpoint, but doesn't pass it along in the repair_json object.

Idempotency token

This may be expected behavior but, when we retry an airflow task with an idempotency token, it results in a successful state, even if the task its retrying has failed initially.

We would expect it to trigger a repair run if repair run is true, else we would expect a failed state.

What you think should happen instead?

No response

Operating System

No response

Deployment

Astronomer

Apache Airflow Provider(s)

databricks

Versions of Apache Airflow Providers

databricks - 7.12

Official Helm Chart version

Not Applicable

Kubernetes Version

No response

Helm Chart configuration

No response

Docker Image customizations

No response

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

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[jroachgolf84]

Can you share a snip of the code that you are running?

[roydobbe]

I just created a job that always fails with some params (python wheel task). Leaving out the job params in the second part means they are not applied.

initial run:

import json
import os
import requests
from azure.identity import ClientSecretCredential

JOB_ID = 12345
TENANT_ID = os.environ.get("AZURE_TENANT_ID", "")
CLIENT_ID = os.environ.get("AZURE_CLIENT_ID", "")
CLIENT_SECRET = os.environ.get("AZURE_CLIENT_SECRET", "")
DBX_HOST = os.environ.get("DBX_HOST", "")
DBX_RESOURCE = "2ff814a6-3304-4ab8-85cb-cd0e6f879c1d" # the same for every azure tenant

credential = ClientSecretCredential(
    tenant_id=TENANT_ID,
    client_id=CLIENT_ID,
    client_secret=CLIENT_SECRET,
)

job_params = {
    "execution_date": "20260101",
    "environment": "dev",
    "log_level": "INFO",
}

def get_headers() -> dict:
    token = credential.get_token(f"{DBX_RESOURCE}/.default")
    return {
        "Authorization": f"Bearer {token.token}",
        "Content-Type": "application/json",
    }

run_now_payload = {
    "job_id": JOB_ID,
    "job_parameters": job_params,
}

resp = requests.post(
    f"{DBX_HOST}/api/2.1/jobs/run-now",
    headers=get_headers(),
    json=run_now_payload,
)
resp.raise_for_status()
run_now_result = resp.json()
print("run-now result:", run_now_result)

RUN_ID = run_now_result["run_id"]
print(f"Triggered run_id: {RUN_ID}")

repair run:

repair_payload = {
    "run_id": RUN_ID,
    "rerun_all_failed_tasks": True, 
    "job_parameters": job_params # leave out and the job params will not be applied to the repaired run.
}

resp = requests.post(
    f"{DBX_HOST}/api/2.1/jobs/runs/repair",
    headers=get_headers(),
    json=repair_payload,
)
resp.raise_for_status()
repair_result = resp.json()
print("repair-run result:", repair_result)

REPAIR_ID = repair_result["repair_id"]
print(f"Repair started. repair_id: {REPAIR_ID}  (same run_id: {RUN_ID})")