Add ExecuteCallback support to AWS ECS Executor

[shivaam]

Implements executor callback support for the AWS ECS Executor.

Sibling to merged Lambda PR #63035; builds on #62645 (merged) and #65392 (merged).

related: #62887

Changes

• supports_callbacks = True gated on AIRFLOW_V_3_3_PLUS so base executor routes ExecuteCallback to ECS.
• Widens key types from TaskInstanceKey to WorkloadKey in ecs_executor.py, EcsQueuedTask, EcsTaskCollection.
• queue_workload() override branches on ExecuteTask vs ExecuteCallback (kept until min Airflow version reaches 3.3, after which BaseExecutor.queue_workload routes both natively; mirrors merged Lambda). No tracking issue yet — happy to open one for the override removal as a follow-up.
• _process_workloads() dispatches both workload types via execute_async().
• Renames methods/attrs to reflect workloads, with AirflowProviderDeprecationWarning shims for the public ones (matches merged Lambda):
  • public + shim: sync_running_tasks → sync_running_workloads, attempt_task_runs → attempt_workload_runs, pending_tasks → pending_workloads
  • private (no shim): __update_running_task → __update_running_workload, __handle_failed_task → __handle_failed_workload
• Relies on Widen BaseExecutor method signatures to accept WorkloadKey #65392 widening BaseExecutor signatures — no cast() needed, log_task_event() skips callback keys internally.

Tests

CI green: ruff, mypy, full test suite, plus compat suites on Airflow 2.11.1 / 3.0.6 / 3.1.8 / 3.2.0.

End-to-end re-verified 2026-05-02 against rebased main on real ECS Fargate. Each DAG runs a regular ExecuteTask (slow_task 120s) followed by a deadline ExecuteCallback:

DAG	Callback pattern	Task result	Callback result
ecs_callback_simple	Plain function from S3 bundle	SUCCESS ~129s	SUCCESS in 3.9s
ecs_callback_context	Function reading dag_run + deadline context	SUCCESS	SUCCESS in 3.8s
ecs_callback_failure	Function raises RuntimeError	SUCCESS	FAILED after 3 retries
ecs_callback_notifier_class	Callable class (notifier __call__ pattern)	SUCCESS	SUCCESS in 4.0s

Verified: PENDING → QUEUED → RUNNING → SUCCESS/FAILED for both workload types; __handle_failed_workload + pending_workloads.append(...) reschedule callbacks correctly; log_task_event and self.fail() with string callback keys handled correctly by #65392's base class. Callback UUIDs and ECS task ARNs from the run are available on request.

Backwards compatibility

Existing task path unchanged. Callback paths gated behind AIRFLOW_V_3_3_PLUS. Public renamed methods/attrs preserved as deprecation shims so subclassers continue to work.

---

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

• Yes — Claude Code (Opus 4.7) for code drafting, sibling-PR review comparison, and end-to-end test orchestration

Generated-by: Claude Code (Opus 4.7) following the guidelines

[shivaam]

@ferruzzi Draft PR

[ferruzzi]

Once #65392 merges (hopefully today) you'll need to rebase and make some small tweaks. That may also let you clean up some of those #ignore directives in _process_workloads, hopefully.

[shivaam]

@ferruzzi Thanks for the review! I have updated the Pr and addressed all your comments.

1. Dropped the cast() wrappers and isinstance(tuple) guards now that Widen BaseExecutor method signatures to accept WorkloadKey #65392 is merged — BaseExecutor handles WorkloadKey natively and skips callback keys in log_task_event internally. Removed the unused cast import and one test whose coverage moved to airflow-core.

2. Renamed the _process_workloads param to workload_items so the module import no longer needs as wl.

3. Renamed the task-named methods and pending_tasks to their workload equivalents. Straight rename, no compat shim

[ferruzzi]

hm. That's an odd CI failure....It looks like Sphinx is throwing a "duplicate object description" warning because oth the Lambda and ECS executors define _process_workloads with the same signature, and autoapi is indexing them as the same object.

I think the easiest fix is to add :sphinx-autoapi-skip: to the _process_workloads docstring. It's a private method that shouldn't be in the public API docs anyway:

def_process_workloads(self, workload_items: Sequence[workloads.All]) ->None:
""":sphinx-autoapi-skip:"""

devel-common/src/docs/utils/conf_constants.py has some code that skips methodfs with that instruction. The spellcheck failure should clear once the docs build passes.

Give that a try??

[shivaam]

Done — added :sphinx-autoapi-skip: on _process_workloads. Also had to split the inner union of CommandType (Sequence[ExecuteTask] | Sequence[ExecuteCallback] instead of Sequence[ExecuteTask | ExecuteCallback]) because Sphinx was flagging a second duplicate-object warning on the TypeAlias text itself shared with Lambda. The split form is also stricter — it disallows mixed task+callback lists, which our code never produces.

[ferruzzi]

That new failing static check is related, sorry about that. looks like you need to add a period since the first line of every docstring needs to be a "complete sentence" and end in a period. Try this: """:sphinx-autoapi-skip:.""" that's kind of awkward, but I think that's what it wants.

The provider one I don't think is related

[ferruzzi]

LGTM. That's an ugly docstring, and maybe it would look nicer as something like

 def _process_workloads(self, workload_items: Sequence[workloads.All]) -> None:
"""
Process queued workloads by dispatching to ECS.
:sphinx-autoapi-skip:
"""

but it was my suggestion, it works, and I'm not generally keen on docstrings that just rephrase the method name. Approving.

[shivaam]

LGTM. That's an ugly docstring, and maybe it would look nicer as something like

 def _process_workloads(self, workload_items: Sequence[workloads.All]) -> None:
"""
Process queued workloads by dispatching to ECS.
:sphinx-autoapi-skip:
"""

but it was my suggestion, it works, and I'm not generally keen on docstrings that just rephrase the method name. Approving.

Let me know if there are any other comments or if this ready to be merged.