EmrServerlessDeleteApplicationOperator never calls DeleteApplication when deferrable=True (silently leaks applications, task reports success)

[imonteroq]

Under which category would you file this issue?

Providers

Apache Airflow version

3.2.1

What happened and how to reproduce it?

Issue Description
EmrServerlessDeleteApplicationOperator never calls the DeleteApplication API when deferrable=True. The task succeeds and logs EMR serverless application <id> deleted successfully, but the application is only stopped, never deleted. This is silent: no error, no warning, and the log explicitly claims the deletion succeeded.

Root cause
EmrServerlessDeleteApplicationOperator subclasses EmrServerlessStopApplicationOperator and begins execute() by delegating to the parent (line numbers from 9.35.0, airflow/providers/amazon/aws/operators/emr.py):

# line 1819
def execute(self, context: Context) -> None:
    # super stops the app (or makes sure it's already stopped)
    super().execute(context)

    self.log.info("Now deleting application: %s", self.application_id)   # line 1821
    response = self.hook.conn.delete_application(applicationId=self.application_id)

With deferrable=True, the parent's execute() reaches this (line 1713):

self.hook.conn.stop_application(applicationId=self.application_id)
if self.deferrable:
    self.defer(
        trigger=EmrServerlessStopApplicationTrigger(...),
        method_name="execute_complete",
    )

self.defer() is typed NoReturn and works by raising TaskDeferred, so it unwinds the stack. super().execute(context) never returns, and line 1821 onwards never runs.

The task is then resumed at method_name="execute_complete". Because the subclass overrides that name, the stop trigger's completion is handled by the delete operator's completion handler (line 1859):

def execute_complete(self, context, event=None) -> None:
    validated_event = validate_execute_complete_event(event)
    if validated_event["status"] != "success":
        raise AirflowException(f"Error deleting EMR Serverless application: {validated_event}")
    self.log.info("EMR serverless application %s deleted successfully", self.application_id)

The event reports success because stopping genuinely succeeded, so the handler logs "deleted successfully" and the task turns green. DeleteApplication is never issued.

EmrServerlessStopApplicationOperator hardcodes method_name="execute_complete" in two places (lines 1713 and 1743), so the subclass has no way to insert the delete step between "stopped" and "done".

Steps to reproduce
Self-contained, no AWS account required:

from unittest import mock

from airflow.providers.amazon.aws.hooks.emr import EmrServerlessHook
from airflow.providers.amazon.aws.operators.emr import EmrServerlessDeleteApplicationOperator

with mock.patch.object(EmrServerlessHook, "conn") as conn:
    op = EmrServerlessDeleteApplicationOperator(
        task_id="delete_app", application_id="test-app-id", deferrable=True
    )
    try:
        op.execute(None)
    except BaseException as e:
        print("raised:", type(e).__name__)
        print("  trigger    :", type(getattr(e, "trigger", None)).__name__)
        print("  method_name:", getattr(e, "method_name", None))

    print("stop_application called  :", conn.stop_application.called)
    print("delete_application called:", conn.delete_application.called)

    # simulate the triggerer firing a successful STOP event
    op.execute_complete(None, {"status": "success"})
    print("after execute_complete -> delete_application called:", conn.delete_application.called)

Actual output:

Stopping application: test-app-id
raised: TaskDeferred
  trigger    : EmrServerlessStopApplicationTrigger
  method_name: execute_complete
stop_application called  : True
delete_application called: False
EMR serverless application test-app-id deleted successfully
after execute_complete -> delete_application called: False

Note the operator defers on the StopApplication trigger, and delete_application is never called on either the initial execution or the resume, yet the final log line reports a successful deletion.

Equivalent end-to-end reproduction against real AWS:

1. Create an EMR Serverless application.
2. Run EmrServerlessDeleteApplicationOperator(application_id=..., deferrable=True).
3. The task succeeds and logs "deleted successfully".
4. aws emr-serverless list-applications still shows the application in STOPPED state, and CloudTrail records StopApplication but no DeleteApplication.

With deferrable=False the operator behaves correctly: the parent does not defer, super().execute() returns normally, and delete_application() is reached.

What you think should happen instead?

DeleteApplication should be called in both deferrable and non-deferrable modes, and the
task should only log a successful deletion once the application has actually been deleted.

Currently the operator reports success while leaving the resource in place, which is worse
than failing, because there is no signal that anything is wrong.

The provider already contains the pattern needed to fix this. In the force_stop path,
EmrServerlessStopApplicationOperator.execute() defers with
method_name="stop_application" and that method performs the next API call before
deferring again. Chained deferral is therefore established style in this same class; the
delete operator simply does not use it.

A minimal fix would make the parent's post-stop resume target overridable, for example a
class attribute:

class EmrServerlessStopApplicationOperator(AwsBaseOperator[EmrServerlessHook]):
    #: Method to resume at once the application has stopped. Subclasses may override this
    #: to chain further steps instead of finishing the task.
    stop_complete_method_name: str = "execute_complete"

used in place of both hardcoded method_name="execute_complete" occurrences (lines 1713
and 1743). The delete operator then sets stop_complete_method_name = "stop_complete" and
implements stop_complete() to issue the DeleteApplication call and defer on
EmrServerlessDeleteApplicationTrigger, leaving execute_complete to handle only the
delete trigger's event.

This keeps the deferrable behaviour intact (no worker slot held during either wait) and is
backwards compatible, since execute_complete continues to exist for task instances that
were already deferred by an older version at upgrade time.

Operating System

Reproduced on macOS 15.6 (Darwin 25.6.0), Python 3.12. Also observed in production on Amazon Linux (Amazon MWAA). Not OS-dependent.

Deployment

Amazon (AWS) MWAA

Apache Airflow Provider(s)

amazon

Versions of Apache Airflow Providers

  apache-airflow-providers-amazon==9.25.0

Also verified present in apache-airflow-providers-amazon==9.35.0 (latest release at time of writing) and on the main branch. EmrServerlessStopApplicationOperator and EmrServerlessDeleteApplicationOperator are byte-identical between 9.25.0 and 9.35.0.

Official Helm Chart version

Not Applicable

Kubernetes Version

Not Applicable

Helm Chart configuration

Not Applicable

Docker Image customizations

Not Applicable

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.