This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new c5d7f60fffe Fix EMR Serverless delete operator skipping deletion when
deferrable (#73323)
c5d7f60fffe is described below
commit c5d7f60fffe7052b9def28b4ededb8a1573506b6
Author: Shikhar Goel <[email protected]>
AuthorDate: Sun Sep 20 19:48:49 2026 +0530
Fix EMR Serverless delete operator skipping deletion when deferrable
(#73323)
* Fix EMR Serverless delete operator skipping deletion when deferrable
With deferrable=True the parent stop step defers, so the delete call placed
after super().execute() was never reached. The task then resumed in the
delete operator's execute_complete with the stop trigger's success event,
logged that the application was deleted and succeeded, silently leaving the
application in place.
closes: #72123
* Reword super() comment and autospec the cancel_running_jobs mock
Review nits: the comment in delete_stopped_application now records why
super().execute_complete is used (this class overrides execute_complete
for the delete trigger's event), and the cancel_running_jobs patch in the
force_stop test uses autospec=True.
Generated-by: Claude Opus 5
---------
Co-authored-by: sgoel2be24-cyber
<[email protected]>
Co-authored-by: Jarek Potiuk <[email protected]>
---
.../airflow/providers/amazon/aws/operators/emr.py | 19 +++++--
.../amazon/aws/operators/test_emr_serverless.py | 62 +++++++++++++++++++++-
2 files changed, 76 insertions(+), 5 deletions(-)
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py
b/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py
index aab67520149..20366b69487 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py
@@ -1646,6 +1646,9 @@ class
EmrServerlessStopApplicationOperator(AwsBaseOperator[EmrServerlessHook]):
template_fields: Sequence[str] = aws_template_fields(
"application_id",
)
+ # Method the task resumes at once the application has stopped in
deferrable mode.
+ # Subclasses can override it to run further steps after the stop.
+ stop_complete_method_name: str = "execute_complete"
def __init__(
self,
@@ -1710,7 +1713,7 @@ class
EmrServerlessStopApplicationOperator(AwsBaseOperator[EmrServerlessHook]):
waiter_max_attempts=self.waiter_max_attempts,
),
timeout=timedelta(seconds=self.waiter_max_attempts *
self.waiter_delay),
- method_name="execute_complete",
+ method_name=self.stop_complete_method_name,
)
if self.wait_for_completion:
waiter = self.hook.get_waiter("serverless_app_stopped")
@@ -1740,7 +1743,7 @@ class
EmrServerlessStopApplicationOperator(AwsBaseOperator[EmrServerlessHook]):
waiter_max_attempts=self.waiter_max_attempts,
),
timeout=timedelta(seconds=self.waiter_max_attempts *
self.waiter_delay),
- method_name="execute_complete",
+ method_name=self.stop_complete_method_name,
)
def execute_complete(self, context: Context, event: dict[str, Any] | None
= None) -> None:
@@ -1786,6 +1789,7 @@ class
EmrServerlessDeleteApplicationOperator(EmrServerlessStopApplicationOperato
template_fields: Sequence[str] = aws_template_fields(
"application_id",
)
+ stop_complete_method_name = "delete_stopped_application"
def __init__(
self,
@@ -1815,9 +1819,18 @@ class
EmrServerlessDeleteApplicationOperator(EmrServerlessStopApplicationOperato
self.wait_for_delete_completion = False if deferrable else
wait_for_completion
def execute(self, context: Context) -> None:
- # super stops the app (or makes sure it's already stopped)
+ # super stops the app (or makes sure it's already stopped). In
deferrable mode it defers
+ # instead of returning, and the task resumes in
``delete_stopped_application``.
super().execute(context)
+ self._delete_application()
+
+ def delete_stopped_application(self, context: Context, event: dict[str,
Any] | None = None) -> None:
+ # super(): this class overrides execute_complete to handle the delete
trigger's event,
+ # while the event here comes from the stop trigger.
+ super().execute_complete(context, event)
+ self._delete_application()
+ def _delete_application(self) -> None:
self.log.info("Now deleting application: %s", self.application_id)
response =
self.hook.conn.delete_application(applicationId=self.application_id)
diff --git
a/providers/amazon/tests/unit/amazon/aws/operators/test_emr_serverless.py
b/providers/amazon/tests/unit/amazon/aws/operators/test_emr_serverless.py
index 75fcf1bce3d..96203a53f7e 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_emr_serverless.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_emr_serverless.py
@@ -30,6 +30,10 @@ from airflow.providers.amazon.aws.operators.emr import (
EmrServerlessStartJobOperator,
EmrServerlessStopApplicationOperator,
)
+from airflow.providers.amazon.aws.triggers.emr import (
+ EmrServerlessDeleteApplicationTrigger,
+ EmrServerlessStopApplicationTrigger,
+)
from airflow.providers.amazon.version_compat import NOTSET
from airflow.providers.common.compat.sdk import AirflowException, TaskDeferred
@@ -1390,15 +1394,69 @@ class TestEmrServerlessDeleteOperator:
@mock.patch.object(EmrServerlessHook, "conn")
def test_delete_application_deferrable(self, mock_conn):
- mock_conn.delete_application.return_value = {"ResponseMetadata":
{"HTTPStatusCode": 200}}
+ operator = EmrServerlessDeleteApplicationOperator(
+ task_id=task_id,
+ application_id=application_id,
+ deferrable=True,
+ )
+ with pytest.raises(TaskDeferred) as defer:
+ operator.execute(None)
+
+ assert isinstance(defer.value.trigger,
EmrServerlessStopApplicationTrigger)
+ assert defer.value.method_name == "delete_stopped_application"
+
mock_conn.stop_application.assert_called_once_with(applicationId=application_id)
+ mock_conn.delete_application.assert_not_called()
+ @mock.patch.object(EmrServerlessHook, "cancel_running_jobs", autospec=True)
+ @mock.patch.object(EmrServerlessHook, "conn")
+ def test_delete_application_deferrable_with_force_stop(self, mock_conn,
mock_cancel_running_jobs):
+ mock_cancel_running_jobs.return_value = 1
operator = EmrServerlessDeleteApplicationOperator(
task_id=task_id,
application_id=application_id,
deferrable=True,
+ force_stop=True,
)
- with pytest.raises(TaskDeferred):
+ with pytest.raises(TaskDeferred) as defer:
operator.execute(None)
+ assert defer.value.method_name == "stop_application"
+
+ with pytest.raises(TaskDeferred) as defer:
+ operator.stop_application(None, {"status": "success",
"application_id": application_id})
+
+ assert isinstance(defer.value.trigger,
EmrServerlessStopApplicationTrigger)
+ assert defer.value.method_name == "delete_stopped_application"
+ mock_conn.delete_application.assert_not_called()
+
+ @mock.patch.object(EmrServerlessHook, "conn")
+ def test_delete_stopped_application_deferrable(self, mock_conn):
+ mock_conn.delete_application.return_value = {"ResponseMetadata":
{"HTTPStatusCode": 200}}
+ operator = EmrServerlessDeleteApplicationOperator(
+ task_id=task_id,
+ application_id=application_id,
+ deferrable=True,
+ )
+
+ with pytest.raises(TaskDeferred) as defer:
+ operator.delete_stopped_application(None, {"status": "success",
"application_id": application_id})
+
+
mock_conn.delete_application.assert_called_once_with(applicationId=application_id)
+ assert isinstance(defer.value.trigger,
EmrServerlessDeleteApplicationTrigger)
+ assert defer.value.method_name == "execute_complete"
+
+ @mock.patch.object(EmrServerlessHook, "conn")
+ def test_delete_stopped_application_stop_error(self, mock_conn):
+ operator = EmrServerlessDeleteApplicationOperator(
+ task_id=task_id,
+ application_id=application_id,
+ deferrable=True,
+ )
+ error_event = {"status": "error", "message": "Stop failed",
"application_id": application_id}
+
+ with pytest.raises(AirflowException, match="Error stopping EMR
Serverless application"):
+ operator.delete_stopped_application(None, error_event)
+
+ mock_conn.delete_application.assert_not_called()
def test_execute_complete_error(self):
operator = EmrServerlessDeleteApplicationOperator(
