feat: add callback support to aws batch executor

[dondaum]

Implements executor callback support for the AWS BatchExecutor.

related: #62887

---

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

• Yes (please specify the tool below)

---

• 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 or {issue_number}.significant.rst, in airflow-core/newsfragments.

[o-nikolas]

I still see a lot of other methods with names, types, variables, comments all mentioning tasks where it should be workloads that need cleaning up.

@ferruzzi should have a look too of course

[o-nikolas]

If you don't mind can you narrow down the type for workload here? I think there is a new type that @ferruzzi made basically anywhere you need task | callback as a type.

[dondaum]

Thank you for the review.

Are you referring to types ExecuteTask and ExecuteCallback? https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/executors/workloads/__init__.py

I think we can only use it if we pin the provider's Airflow core dependency to when this change was implemented. This happened here #61153.

If we want to maintain backwards compatibility, I can't use the new types.. Currently the Amazon provider is apache-airflow>=2.11.0

[o-nikolas]

Even for just typing? @ferruzzi thoughts?

[ferruzzi]

@onikolas is likely thinking of SchedulerWorkload from workloads/types.py, but @dondaum may be right, that isn't going to be in 2.11 so we can't use that until we bunmp the min_ver.

I think we could do a conditional import though:

workload_type_hint = workloads.All

if airflow version  > 3.2:
    from airflow.executors.workloads.types import SchedulerWorkload 

    workload_type_hint = SchedulerWorkload

And that will force it to get cleaned up later when we pin the versions up??

[dondaum]

Let's say we make this conditional input work here. We are still limited by the BaseExecutor type having this

def queue_workload(self, workload: workloads.All, session: Session) -> None:
    ...

[dondaum]

Since #62645 is now merged, I can narrow down the types.

[o-nikolas]

What is the state of this thread? Is it still open or can we resolve this one?

[dondaum]

Thanks for another look at this.
In my opinion we can solve it.
The relevant comment can be found at https://github.com/apache/airflow/pull/63657/changes#r3113924410.

[potiuk]

@dondaum A few things need addressing before review — see our Pull Request quality criteria.

• ⚠️ Unresolved review comments: This PR has 3 unresolved review threads from maintainers. Please review and resolve all inline review comments before requesting another review. You can resolve a conversation by clicking 'Resolve conversation' on each thread after addressing the feedback. See pull request guidelines.

Note: Your branch is 278 commits behind main. Please rebase and push again to get up-to-date CI results.

No rush.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[dondaum]

I still see a lot of other methods with names, types, variables, comments all mentioning tasks where it should be workloads that need cleaning up.

@ferruzzi should have a look too of course

I changed the reference from task to workload where I could make a change. There is a method try_adopt_task_instances() which I have not changed since it is part of the base class. But technically it should work on both tasks and callbacks.

[shivaam]

Thanks. Left a comment.