apache / apache/airflow

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

Open
#72,123 1 comment 0 reactions 0 assignees View on GitHub
area:providers kind:bug needs-triage provider:amazon
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 9h
Merged PRs (30d)
472

Description

### 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 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`):

```python
# 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):

```python
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):

```python
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:
```python
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:

```python
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

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Research direction

Start in airflow/providers/amazon/aws/operators/emr.py by reading EmrServerlessStopApplicationOperator.execute(), its force_stop path, and EmrServerlessDeleteApplicationOperator.execute_complete(). Reproduce the mocked deferrable flow from the issue, then add or update focused coverage for both deferrable and non-deferrable modes. Done means DeleteApplication is called after stopping and success is logged only after deletion completes.

Written by the indexing model from the issue text.

Assessment

Tech stack
aws, python
Domain
backend, cloud
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
75/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.