This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 90354f166e1 Fix EmrContainerSensor reporting success for an unknown
job state (#72500)
90354f166e1 is described below
commit 90354f166e1d511483c7084d11d0678b34f11cc4
Author: Priyadharshini Jaffar Ali <[email protected]>
AuthorDate: Fri Sep 4 12:44:50 2026 -0400
Fix EmrContainerSensor reporting success for an unknown job state (#72500)
EmrContainerSensor.poke() returned True for any state that was neither a
failure state nor an intermediate state, so an unknown state was treated
as a successful job run. SUCCESS_STATES was declared but never used.
EmrContainerHook.check_query_status returns None when it swallows a
generic ClientError, and poll_query_status passes that None through once
max_polling_attempts is reached. A throttled or otherwise transiently
failing DescribeJobRun therefore marked the task successful while the job
was still running.
Check state against SUCCESS_STATES instead, and add tests for the None
and unrecognised-state cases.
---
.../src/airflow/providers/amazon/aws/sensors/emr.py | 2 +-
.../tests/unit/amazon/aws/sensors/test_emr_containers.py | 15 +++++++++++++++
2 files changed, 16 insertions(+), 1 deletion(-)
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/sensors/emr.py
b/providers/amazon/src/airflow/providers/amazon/aws/sensors/emr.py
index f9a63289388..cfb8575a752 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/sensors/emr.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/sensors/emr.py
@@ -309,7 +309,7 @@ class EmrContainerSensor(AwsBaseSensor[EmrContainerHook]):
if state in self.INTERMEDIATE_STATES:
return False
- return True
+ return state in self.SUCCESS_STATES
def execute(self, context: Context):
if not self.deferrable:
diff --git
a/providers/amazon/tests/unit/amazon/aws/sensors/test_emr_containers.py
b/providers/amazon/tests/unit/amazon/aws/sensors/test_emr_containers.py
index 79cf485d7af..e1c7a820155 100644
--- a/providers/amazon/tests/unit/amazon/aws/sensors/test_emr_containers.py
+++ b/providers/amazon/tests/unit/amazon/aws/sensors/test_emr_containers.py
@@ -78,6 +78,21 @@ class TestEmrContainerSensor:
assert "EMR Containers sensor failed" in str(ctx.value)
assert "CANCEL_PENDING" in str(ctx.value)
+ @mock.patch.object(EmrContainerHook, "check_query_status",
side_effect=(None,))
+ def test_poke_unknown_state(self, mock_check_query_status):
+ """An unknown state must not be reported as a successful job run.
+
+ ``EmrContainerHook.check_query_status`` swallows a generic
``ClientError``
+ and returns ``None``, which ``poll_query_status`` passes straight
through
+ once ``max_polling_attempts`` is reached.
+ """
+ assert not self.sensor.poke(None)
+
+ @mock.patch.object(EmrContainerHook, "check_query_status",
side_effect=("SOME_FUTURE_STATE",))
+ def test_poke_unrecognised_state(self, mock_check_query_status):
+ """A state added by AWS later must not be reported as a successful job
run."""
+ assert not self.sensor.poke(None)
+
@mock.patch("airflow.providers.amazon.aws.sensors.emr.EmrContainerSensor.poke")
def test_sensor_defer(self, mock_poke):
self.sensor.deferrable = True
