Uh oh!
There was an error while loading. Please reload this page.
Feat/kubernetes executor callback support - #67449
Conversation
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
| key: TaskInstanceKey | ||
| command: Sequence[str] | ||
| key: WorkloadKey | ||
| command: Sequence[Any] |
There was a problem hiding this comment.
I liked the way the ECS Executor handled this; they did
if AIRFLOW_V_3_3_PLUS:
CommandType: TypeAlias = Sequence[str] | Sequence[ExecuteTask] | Sequence[ExecuteCallback]
else:
CommandType: TypeAlias = Sequence[str]
then defined command as a CommandType instead of Sequence[All]
There was a problem hiding this comment.
Created a WorkloadCommand to implement the same handle. The reason not name is as CommandType is because at L92, there is already a definition. Not sure if that would be safe to remove, but it might cause some shadowing when using the same name.
Uh oh!
There was an error while loading. Please reload this page.
Hi @ferruzzi , I am still working on resolving the compatibility issue and also looking into the trade-off for the feedback (#67449 (comment)) After creating the I am looking into a way if this typing can be handled a bit more better (without using Sorry, my progress is very slow since past months due to bandwidth limit. I will try to address the the review feedback by this week. |
ferruzzi
commented
Jun 2, 2026
I just did a very quick look, so don't hold me to this, but I think you can make |
9d88f0b to
66afca9Compare| from airflow.executors.workloads import ExecuteCallback | ||
| from airflow.models.callback import CallbackKey |
There was a problem hiding this comment.
The two imports here are under TYPE_CHECKING with is not guard by the AIRFLOW_V_3_3_PLUS. These are used for type hint in the following funciton.
def_run_next_callback(
self, key: CallbackKey, workload: ExecuteCallback, pod_template_file: str|None
) ->None:sjyangkevin
commented
Jun 7, 2026
Hi @ferruzzi , I've done a rework on the PR, this should be a cleaner implementation for the callback support. I am looking into the CI failures related to Serialization. Feel free to let me know if you have more feedback. Thanks!  |
sjyangkevin
commented
Jun 8, 2026
Opened the PR #68195 , attempt to address the issue in CI. |
Runs synchronous callbacks (e.g. Deadline Alerts) as supervised callback pods, mirroring LocalExecutor/CeleryExecutor. Adds callback dispatch in queue_workload/_process_workloads, a construct_callback_pod path, watcher key resolution via the callback_id annotation, WorkloadKey/WorkloadState type widening, and unit + k8s integration tests. Guarded by AIRFLOW_V_3_3_PLUS for provider backward compatibility.
66afca9 to
259646aComparepotiuk
commented
Jul 8, 2026
Note 🛠️ Maintainer triage note for @sjyangkevin · by This draft has been inactive for ~20 days, so I'm closing it to keep the review queue tidy — no judgment on the work itself. No rush — reopen it (or open a fresh PR) whenever you're ready to continue; nothing is lost. Automated triage — may be imperfect; a maintainer takes the next look. |
Implements
supports_callbackson KubernetesExecutor by running each ExecuteCallback workload as its own pod, alongside the existing task-pod pipeline. Gated onAIRFLOW_V_3_3_PLUS.Approach
Use the same way as how Task is dispatched to Kubernetes but with a different configuration for pod label/annotation.
Pod annotations: tasks vs callbacks
For task pods,
PodGenerator.construct_pod()writes these annotations so the watcher can reconstructTaskInstanceKey:For callback pods, there is no
task_id,try_number, ormap_index. The relevant identity is the callback UUID.Proposed annotations:
A new pod label is also needed to let the watcher distinguish callback pods from task pods without relying on annotation presence:
Pod construction for callbacks: no
PodGenerator.construct_pod()changesPodGenerator.construct_pod()is specifically built for task identity (takesdag_id,task_id,try_number, etc.). Callback pods should not go throughconstruct_pod(). They need their own construction path that:get_base_pod_from_template)workload_to_command_args(workload)forExecuteCallback, which serializes workload to--json-stringairflow-worker=<job_id>,kubernetes_executor=True,airflow-workload-type=callbackcallback_id,dag_id,run_id(extracted fromworkload.log_pathorworkload.callback.data)pod_idfromcreate_unique_id("callback", callback_id_short)to avoid name collisions with task podsThe command is identical because
execute_workloaddispatches on the workload type at runtime — the same subprocess entry point handles bothExecuteTaskandExecuteCallback.Differences from the task pod lifecycle:
queued_tasks[TaskInstanceKey]queued_callbacks[CallbackKey]TaskInstanceKey(dag_id, task_id, run_id, try_number, map_index)CallbackKey(id only)construct_pod()construct_callback_pod()task_id,try_numbercallback_idannotations_to_key()CallbackKey(id=annotations["callback_id"])TaskInstance.filter_for_tisCallbackState.SUCCESSdirectly (no TI row)TaskInstanceStateCallbackState.keyTaskInstanceKey(dag_id, task_id, run_id, try_number, map_index)CallbackKey(id=<uuid-str>)try_number)airflow-worker=<job_id>Both
TaskInstanceStateandCallbackStatefit intoWorkloadState(TaskInstanceState | CallbackState | ConnectionTestState), which is whyKubernetesResults.stateandKubernetesWatch.stateuse theWorkloadStatetype alias — the same queue structs carry either kind of state depending on what ran in the pod.sequenceDiagram participant Sched as Scheduler participant KE as KubernetesExecutor participant AKS as AirflowKubernetesScheduler participant KJW as "KubernetesJobWatcher (Process)" participant K8s as "Kubernetes API" Note over Sched,KE: Phase 1 — Queuing (CallbackKey has only .id — no dag_id) Sched->>KE: queue_workload(ExecuteCallback, session) Note right of KE: queued_callbacks[CallbackKey] = ExecuteCallback Note over Sched,KE: Phase 2 — Heartbeat dispatches callback Sched->>KE: heartbeat() → trigger_tasks() → _process_workloads() KE->>KE: key = workload.callback.key (CallbackKey) KE->>KE: execute_async(key=CallbackKey, command=[ExecuteCallback]) Note right of KE: task_queue.put(KubernetesJob)<br/>running.add(CallbackKey) Note over KE,K8s: Phase 3 — sync() builds and submits callback pod Sched->>KE: sync() KE->>AKS: run_next(KubernetesJob) AKS->>AKS: isinstance(command[0], ExecuteCallback) → True AKS->>AKS: _run_next_callback()<br/>workload_to_command_args(ExecuteCallback)<br/>→ ["python", "-m", "execute_workload", "--json-string", ...] AKS->>AKS: construct_callback_pod()<br/>annotations: {callback_id: uuid, dag_id, run_id}<br/>labels: {airflow-workload-type: callback, airflow-worker: ...} AKS->>K8s: create_namespaced_pod(callback pod spec) Note right of K8s: Pod: Pending → Running<br/>(Running MODIFIED: not re-queued,<br/>same behaviour as task pods) Note over KJW,K8s: Phase 4 — Watcher streams terminal pod event KJW->>K8s: watch.Watch().stream(label_selector=airflow-worker={job_id}) Note right of K8s: Pod: Running → Succeeded K8s-->>KJW: event{type:MODIFIED, phase:Succeeded} KJW->>AKS: watcher_queue.put(KubernetesWatch(<br/> state=None, ← Succeeded maps to None (same as tasks)<br/> annotations={callback_id: uuid, dag_id, run_id}<br/>)) Note over KE,AKS: Phase 5 — Promote to result: callback_id → CallbackKey Sched->>KE: sync() (next heartbeat) KE->>AKS: kube_scheduler.sync() AKS->>AKS: process_watcher_task()<br/>"callback_id" in annotations → CallbackKey(id=uuid)<br/>(falls back to annotations_to_key() for task pods) AKS->>KE: result_queue.put(KubernetesResults(<br/> key=CallbackKey, ← WorkloadKey union<br/> state=None ← WorkloadState | None<br/>)) Note over Sched,KE: Phase 6 — _change_state resolves None→SUCCESS without DB KE->>KE: _change_state(key=CallbackKey, state=None)<br/>running.remove(CallbackKey)<br/>state is None → AIRFLOW_V_3_3_PLUS and not hasattr(key,"dag_id")<br/> → state = CallbackState.SUCCESS (no TaskInstance row exists)<br/>event_buffer[CallbackKey] = (CallbackState.SUCCESS, None) KE->>AKS: delete_pod(callback pod) KE->>Sched: event_buffer consumed → callback result written to DBTest Cases
Unit Test
test_supports_callbacks_attributeexecutor.supports_callbacks is Truetest_queue_callback_workloadqueue_workload(callback_workload, session)stores workload inexecutor.queued_callbacks[callback_workload.callback.key]and does NOT touchqueued_taskstest_queue_task_and_callback_are_independentqueued_taskshas 1 entry,queued_callbackshas 1 entry, no cross-contaminationtest_process_workloads_callback_process_workloads([callback_workload]): key removed fromqueued_callbacks, key present inrunning,kube_scheduler.run_nextcalled oncetest_process_workloads_mixedExecuteTaskand oneExecuteCallbacktogether; both go torun_next, both inrunningtest_execute_async_callback_commandexecute_async(key=callback_key, command=[callback_workload])puts aKubernetesJobontask_queuewhose command is["python", "-m", "airflow.sdk.execution_time.execute_workload", "--json-string", ...]test_change_state_callback_success_explicit_stateKubernetesResults(key=CallbackKey, state=CallbackState.SUCCESS);event_buffer[callback_key]set, pod deleted, key removed fromrunningtest_change_state_callback_success_state_noneKubernetesResults(key=CallbackKey, state=None); implementation must NOT callTaskInstance.filter_for_tis(wrong key type) and must instead setstate=CallbackState.SUCCESSwithout querying DBtest_change_state_callback_failedstate=FAILED;event_buffer[callback_key]set to failed state; pod deleted (with failure-deletion config)test_change_state_callback_pod_not_deleted_if_keep_podsdelete_worker_pods=False;patch_pod_executor_donecalled instead ofdelete_podtest_sync_drains_callback_resultsKubernetesResults(key=CallbackKey)onresult_queue;sync()drains it and updatesevent_buffertest_run_next_callback_creates_podrun_next(KubernetesJob(key=CallbackKey, command=[callback_workload], ...))callskube_client.create_namespaced_podtest_callback_pod_has_callback_id_annotationcallback_id=<uuid>test_callback_pod_has_workload_type_labelairflow-workload-type=callbacktest_callback_pod_has_dag_id_and_run_id_annotationsdag_idandrun_idextracted fromworkload.log_pathtest_callback_pod_command_is_execute_workload["python", "-m", "airflow.sdk.execution_time.execute_workload", "--json-string", <json>];<json>round-trips back to the originalExecuteCallbacktest_callback_pod_does_not_have_task_annotationstask_id,try_number, ormap_indextest_callback_pod_name_does_not_collide_with_task_podtest_process_watcher_task_callback_succeededprocess_watcher_task(KubernetesWatch(..., annotations={"callback_id": "..."}))putsKubernetesResults(key=CallbackKey(...))onresult_queuetest_process_watcher_task_callback_failedstate=TaskInstanceState.FAILEDtest_process_watcher_task_task_still_uses_task_keytask_id/try_numberannotations (nocallback_id) still resolves toTaskInstanceKey— regression checktest_process_watcher_task_unknown_annotations_droppedcallback_idnortask_idannotations is silently dropped (no crash, no queue entry)test_process_status_callback_pod_succeededKubernetesJobWatcher.process_status()with phaseSucceededandcallback_idannotation puts correctKubernetesWatchonwatcher_queuetest_process_status_callback_pod_failedFailed;failure_detailsincluded in the queued watchtest_process_status_callback_pod_runningRunningputsKubernetesWatchwithstate=TaskInstanceState.RUNNINGtest_process_status_callback_pod_revoked_label_skippedairflow_pod_revoked=Truelabel is skipped even if it hascallback_idannotationtest_task_execution_unaffected_on_older_airflowAIRFLOW_V_3_3_PLUS=Falsepatched, processing anExecuteTaskworkload puts it ontask_queuewithout touching callback codetest_queue_non_callback_non_task_raisesRuntimeErrortest_supports_callbacks_does_not_break_old_airflowqueue_workload(ExecuteTask, session)succeeds and stores inqueued_tasks(regression guard for older Airflow)k8s Tests
These tests use
breeze k8sto deploy a local cluster and run against the Kubernetes API. They follow the polling pattern ofBaseK8STest: trigger a Dag run via the REST API, poll for the callback's state, and assert the terminal outcome.Test:
test_deadline_callback_executes_on_kubernetesTest:
test_deadline_callback_pod_failure_marks_callback_failedTest:
test_callback_pod_annotations_and_labelsTest:
test_callback_pod_survives_scheduler_restartTest:
test_callback_pod_is_cleaned_up_after_successSamples
Watch callback pods
Check a callback pod's logs
Check a callback pod's details
Was generative AI tooling used to co-author this PR?
Generated-by: [Claude Code (Sonnet 4.6)] following the guidelines
{pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.