Add callback support on kubernetes executors

[JH-A-Kim]

---

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.

[JH-A-Kim]

@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]

Shouldn't this be wrapped in a state check if Callback.state is still QUEUED? Assuming SUCCESS feels dangerous.

[JH-A-Kim]

From how I understand it the callback state is never updated during the running so it will always be in the QUEUED state so adding the check wouldnt really change the behaviour of the code as far as i am aware

[JH-A-Kim]

And there is no state row in callback states for the db because there is a lot more cases that tasks can be than callbacks with callbacks only having 6 potential states and tasks having 12ish potential states.

[JH-A-Kim]

But if we do add the check it could help to catch anything in the future?

[ferruzzi]

Alright, I've been thinking about this. It looks like this is fine how it is for now, but it may need some more thinking later. Something feels off with the callback state flow here, but I can't place it right now.

Go ahead and resolve this when you read it. If you do go on to do the callback-retry work, maybe spend some time thinking this through... it seems like there's a step missing here with the state updates compared to the task state updates.

[ferruzzi]

@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]

@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]

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.