Add callback support on kubernetes executors by JH-A-Kim · Pull Request #70475 · apache/airflow

JH-A-Kim · 2026-07-26T18:00:29Z

Description

Related PR's: #67449
#63454
Related bug fix PR: #70148

To reiterate the main points this pr aims to implement support for callbacks on the Kubernetes executor because it currently only knows how to handle task instances. This PR makes it so it can also handle ExecuteCallback workloads. These are keyed by CallbackKey so that callbacks can execute as their own pod instead of requiring another mechanism.

The core changes are the following:

New Generic types in kubernetes_executor_types.py where WorkloadKey/WorkloadState/WorkloadCommand type aliases help to widen the executors internal types from always being a task to now being able to be a task or a callback and is gated behind AIRFLOW_V_3_3_PLUS so that behaviour older versions of Airflow can remain unchanged
The routing in queue_workload and process_workloads now branch on the the workload is either a ExecuteTask or ExecuteCallback
Pods now have another build path for ExecuteCallback called _run_next_callback which builds and submits the pod itself and returns None.
Now with this there is a new pod constructor called PodGenerator.construct_callback_pod which builds a minimal pod for a callback and tags it with a callback_id id annotation and a airflow-workload-type: callback label so that the watcher can tell it apart from task pods
For the watcher when the pod events annotation contains a callback_id the watcher reconstructs a CallbackKey instead of a TaskInstanceKey and state resolution treats a callback with no error as CallbackState.SUCCESS rather than querying the (nonexistent) TaskInstance row.
For pod adoption after a scheduler restart, it just re-labels it with a new schedulers worker ID, since callback pods dont participate in the normal orphan recovery/retry logic that task pods do.

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

Yes (please specify the tool below)

Generated-by: Sonnet 5 following the guidelines

Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
When adding dependency, check compliance with the ASF 3rd Party License Policy.
For significant user-facing changes create newsfragment: {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.

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.

local import to follow pre-existing file conventions

JH-A-Kim · 2026-08-03T21:35:59Z

@ferruzzi Currently one of the big bug fixes was in _change_state() where there would be a timeout because callbacks would enter into the task retry logic for the specific case of before execution a failure would occur and would then check for dag_id which it does not have and continuously add it back to the retry queue. Which would cause tests to timeout. But do you think its worth adding logic to restart the callback pod if that were to occur? Currently my check just makes sure entrance into that particular bit of logic is only for tasks. But if the pod were to fail before running I think it might be necessary for it to try and restart in that case automatically.

ferruzzi · 2026-08-06T22:14:24Z

@JH-A-Kim It seems worth doing, but maybe in a different PR, this one is already pretty chonky and I don't believe any execujtors currently retry callbacks. I think the right answer there is to convert _get_task_instance_state to _get_workload_state which checks Callback.state directly; Callback.state already has ACTIVE_STATES and `TERMINAL_STATES`` defined which you can check against.

JH-A-Kim · 2026-08-07T18:51:11Z

@JH-A-Kim It seems worth doing, but maybe in a different PR, this one is already pretty chonky and I don't believe any execujtors currently retry callbacks. I think the right answer there is to convert _get_task_instance_state to _get_workload_state which checks Callback.state directly; Callback.state already has ACTIVE_STATES and `TERMINAL_STATES`` defined which you can check against.

yeah that makes sense ill create an issue when I get the time to ask about potentially adding retry logic to callbacks, or should I create a discussion on the devlist?

ferruzzi · 2026-08-10T21:38:11Z

yeah that makes sense ill create an issue when I get the time to ask about potentially adding retry logic to callbacks, or should I create a discussion on the devlist?

An Issue is fine for now, but mark is as dependent on this PR and on #63491 both being merged, that way we can reduce merge conflicts all around.

SameerMesiah97

I have left some comments.

SameerMesiah97 · 2026-08-20T21:11:29Z

+            """Placeholder task for the failing-callback DAG."""
+            print("dummy_task_failing executed")
+
+        dummy_task_failing()


Given that this DAG covers multiple scenarios rather than 'standard usage', I wonder whether this belongs in example dags at all. Could we keep the example DAG focused on demonstrating Kubernetes executor callback support and move the scheduler-restart/failure fixtures into the relevant integration test setup?

SameerMesiah97 · 2026-08-20T21:18:45Z

+                )
+            except ApiException as e:
+                self.log.info("Failed to adopt pod %s. Reason: %s", pod.metadata.name, e)
+            return


This seems inconsitent with what has been done for the other executors (for e.g. LambdaExecutor) wrt adding callback support. . Do we actually want callback workloads to participate in scheduler adoption here, or should the lifecycle semantics be consistent across executors?

SameerMesiah97 · 2026-08-20T21:20:34Z

+
+                ti_id = CallbackKey(id=pod.metadata.annotations[CALLBACK_POD_ANNOTATION_KEY])
+            else:
+                ti_id = annotations_to_key(pod.metadata.annotations)


The above comment on adopting callbacks applies here too.

SameerMesiah97 · 2026-08-20T21:24:44Z

@@ -45,8 +69,8 @@ class FailureDetails(TypedDict, total=False):
 class KubernetesResults(NamedTuple):
    """Results from Kubernetes task execution."""


Suggested change

"""Results from Kubernetes task execution."""

"""Results from Kubernetes workload execution."""

SameerMesiah97 · 2026-08-20T21:36:20Z

+            workload = command[0]
+            if isinstance(workload, ExecuteCallback):
+                self._run_next_callback(workload.key, workload, pod_template_file)
+                return None


Why does _build_pod_request now also submit the pod for callback workloads? This previously had a clear side-effect-free contract where it only constructed and returned the pod, with submission handled by the caller. Could _run_next_callback just construct and return the callback pod so we retain that separation and callbacks can follow the same pod creation path as tasks?

SameerMesiah97 · 2026-08-20T21:42:36Z

+        pod_id = add_unique_suffix(name=f"callback-{short_id}", rand_len=8, max_len=POD_NAME_MAX_LENGTH)
+
+        try:
+            image = base_worker_pod.spec.containers[0].image


Is the precedence here intentional? kube_image is effectively only a fallback since we copy the image from base_worker_pod into the dynamic pod whenever it exists. This also seems different from the normal task pod construction where kube_image is passed into the dynamic pod and the normal reconciliation determines the final result. Could we follow the same approach here?

SameerMesiah97 · 2026-08-20T21:43:28Z

+            image = base_worker_pod.spec.containers[0].image
+            if not image:
+                image = kube_image
+        except Exception:


Exception here is too broad. Not necessarily unjustified but I wish to know why a narrower exception was not used/

SameerMesiah97 · 2026-08-20T21:45:16Z

+            if AIRFLOW_V_3_3_PLUS and CALLBACK_POD_ANNOTATION_KEY in annotations:
+                # Callback pod: forward only the annotations that process_watcher_task needs.
+                # Callback pods carry callback_id instead of task_id/try_number.
+                task_instance_related_annotations = {


nit: could be workload_related_annotations to be more accurate.

SameerMesiah97 · 2026-08-20T21:50:05Z

+        phase = self._wait_for_pod_phase(pod_name, ["Succeeded", "Failed", "Deleted"], timeout=180)
+        assert phase in ("Succeeded", "Deleted"), (
+            f"Callback pod {pod_name!r} reached {phase!r} after scheduler restart (expected Succeeded/Deleted)"
+        )


Not sure if this tests actually covers what it is supposed to. Consdier a situation where the new scheduler never adopts the callback pod. The pod would continue running independently and eventually reach Succeeded anyway, so this test would still pass. Should we explicitly verify that the callback pod is adopted by the new scheduler and that its terminal state is processed?

That's if we decide that callbacks should be adopted.

SameerMesiah97 · 2026-08-20T21:51:26Z

+        try:
+            # Failed pods are NOT auto-deleted (delete_worker_pods_on_failure=False default).
+            phase = self._wait_for_pod_phase(pod_name, ["Failed"], timeout=120)
+            assert phase == "Failed", f"Expected Failed phase, got {phase!r}"


The test description says we're verifying that the watcher emits FAILED, but asserting the Kubernetes pod phase only verifies that the callback container failed. Is there a way to assert that the failed result was actually processed by the executor?

aaron-y-chen

Hi, I think there's a potential bug that might needs your help to fix. 🙂

On KubernetesExecutor (Airflow 3.4.0, k8s v1.30.13), if the scheduler restarts while an ExecuteCallback pod from example_deadline_callback.py is still running, the scheduler afterwards crashes on every start and never recovers on its own.

How to reproduce

Deploy Airflow on K8s with KubernetesExecutor.
Raise the sleep in slow_deadline_callback from 30 to 300 (or more longer) so there is time to act.
Trigger example_deadline_callback_slow.
Wait until the callback pod is Running
(kubectl get pods -l airflow-workload-type=callback).
While it is still running, restart the scheduler
(kubectl delete pod -l component=scheduler).
Wait for the callback pod to reach Succeeded, then allow up to one
orphaned_tasks_check_interval (default 300s), the crash does not start immediately.

From then on the scheduler crash-loops: every restart dies within ~20s and it never recovers on its own. The only way out was deleting the leftover callback pod by hand, and nothing in the logs or the UI points at which pod to delete.

The scheduler log shows:

File ".../executors/kubernetes_executor.py", line 1271, in _adopt_completed_pods
    ti_id = annotations_to_key(pod.metadata.annotations)
File ".../kubernetes_helper_functions.py", line 171, in annotations_to_key
    try_number = int(annotations["try_number"])
TypeError: int() argument must be a string, a bytes-like object or a real number, not 'NoneType'

The callback also never reaches a terminal state. Its pod logs "event":"Callback executed successfully", but callback.state stays queued indefinitely:

Note the DAG run and its task instance are both success, so this is not visible from the UI, only the callback result is lost.

sjyangkevin and others added 6 commits June 18, 2026 21:20

fix mypy error

b56f8ca

move AIRFLOW_V_3_3_PLUS out of TYPE_CHECKING to fix mypy error

ee13931

add TODO comment to queue_workload

989ac29

reolsve implicit typing

259646a

Merge branch 'main' into kube-callbacks

acfd28c

boring-cyborg Bot added area:kubernetes-tests area:providers provider:cncf-kubernetes Kubernetes (k8s) provider related issues labels Jul 26, 2026

JH-A-Kim added 17 commits July 27, 2026 15:25

test fix for kubernetes wait for callback pods

d72bd13

Merge branch 'apache:main' into kube-callbacks

b80c00b

revert back before changes

e8eb1e1

test fixs for kubernetes_executor

cb42ea8

test fix with trigger_dag_run

ecc5a41

testing crash logs

d206963

test callback annotation guards

0bee71f

test new fix

5303775

fix key name

acde97d

Merge branch 'apache:main' into kube-callbacks

e1d2596

Merge branch 'apache:main' into kube-callbacks

8032d56

Merge branch 'apache:main' into kube-callbacks

e2e3d50

fix breaking change from main

f7bab73

resolve type assignment

932074c

fix annotation issue

7286dff

add block to id add to retry queue

ff1614e

local import to taskinstance key to fix CI error

fe04826

local import to follow pre-existing file conventions

ferruzzi reviewed Aug 6, 2026

View reviewed changes

Comment thread ...iders/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py

Merge branch 'apache:main' into kube-callbacks

6370418

JH-A-Kim marked this pull request as ready for review August 19, 2026 18:32

JH-A-Kim requested review from ashb, gopidesupavan, hussein-awala, jason810496, jedcunningham, jscheffl and potiuk as code owners August 19, 2026 18:32

SameerMesiah97 reviewed Aug 20, 2026

View reviewed changes

aaron-y-chen reviewed Aug 21, 2026

View reviewed changes

JH-A-Kim marked this pull request as draft August 25, 2026 22:01

		@@ -45,8 +69,8 @@ class FailureDetails(TypedDict, total=False):
		class KubernetesResults(NamedTuple):
		"""Results from Kubernetes task execution."""

	"""Results from Kubernetes task execution."""
	"""Results from Kubernetes workload execution."""

Uh oh!

Conversation

JH-A-Kim commented Jul 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

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

Uh oh!

JH-A-Kim commented Aug 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

ferruzzi commented Aug 6, 2026

Uh oh!

JH-A-Kim commented Aug 7, 2026

Uh oh!

ferruzzi commented Aug 10, 2026

Uh oh!

SameerMesiah97 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aaron-y-chen left a comment

Choose a reason for hiding this comment

How to reproduce

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

JH-A-Kim commented Jul 26, 2026 •

edited

Loading

JH-A-Kim commented Aug 3, 2026 •

edited

Loading