From f0fc76b9cbfc09ef35eb19a5af5963908e6a6346 Mon Sep 17 00:00:00 2001
From: Steve Ahn <38049807+steveahnahn@users.noreply.github.com>
Date: Sat, 11 Jul 2026 19:26:34 -0700
Subject: [PATCH] Fix AirbyteJobSensor marking a cancelled job as success in
 deferrable mode

When the deferred sensor resumes, execute_complete only failed on an "error"
event. A job cancelled while the sensor was deferred yields a "cancelled"
event, which fell through to the success path, so the task was marked SUCCESS
and downstream tasks ran against a sync that never completed. The same sensor
fails correctly on a cancelled job in non-deferrable mode, and the sibling
AirbyteTriggerSyncOperator already fails on the cancelled event, so deferrable
sensor runs silently diverged from both. Fail closed on any non-success status
so unrecognized statuses cannot be treated as success either.
---
 .../providers/airbyte/sensors/airbyte.py      | 13 ++++----
 .../unit/airbyte/sensors/test_airbyte.py      | 33 +++++++++++++++++++
 2 files changed, 40 insertions(+), 6 deletions(-)

diff --git a/providers/airbyte/src/airflow/providers/airbyte/sensors/airbyte.py b/providers/airbyte/src/airflow/providers/airbyte/sensors/airbyte.py
index 5c6be4b9c33e7..271d801d17c25 100644
--- a/providers/airbyte/src/airflow/providers/airbyte/sensors/airbyte.py
+++ b/providers/airbyte/src/airflow/providers/airbyte/sensors/airbyte.py
@@ -126,15 +126,16 @@ def execute(self, context: Context) -> Any:
                 )
 
     def execute_complete(self, context: Context, event: Any = None) -> None:
-        """
-        Invoke this callback when the trigger fires; return immediately.
-
-        Relies on trigger to throw an exception, otherwise it assumes execution was
-        successful.
-        """
+        """Invoke this callback when the trigger fires and fail unless the job succeeded."""
         if event["status"] == "error":
             self.log.debug("An error occurred with context: %s", context)
             raise AirflowException(event["message"])
 
+        if event["status"] != "success":
+            # A cancelled job must fail the sensor, matching the poke() behavior;
+            # fail closed on any status this sensor does not recognize.
+            self.log.debug("Job did not succeed, context: %s", context)
+            raise RuntimeError(event["message"])
+
         self.log.info("%s completed successfully.", self.task_id)
         return None
diff --git a/providers/airbyte/tests/unit/airbyte/sensors/test_airbyte.py b/providers/airbyte/tests/unit/airbyte/sensors/test_airbyte.py
index 69b5d0bffe093..65931085a0138 100644
--- a/providers/airbyte/tests/unit/airbyte/sensors/test_airbyte.py
+++ b/providers/airbyte/tests/unit/airbyte/sensors/test_airbyte.py
@@ -117,3 +117,36 @@ def test_airbyte_job_sensor_init(self):
         )
         assert sensor.poke_interval == 10
         assert sensor.timeout == 3600
+
+    @pytest.mark.parametrize(
+        ("status", "message"),
+        [
+            pytest.param("cancelled", "Job run 1 has been cancelled.", id="cancelled"),
+            pytest.param("unmapped_status", "Job run 1 did something unexpected.", id="fail-closed"),
+        ],
+    )
+    def test_execute_complete_fails_when_job_did_not_succeed(self, status, message):
+        """A deferred sensor must fail on a non-success event, matching the poke() behavior."""
+        sensor = AirbyteJobSensor(
+            task_id=self.task_id,
+            airbyte_job_id=self.job_id,
+            airbyte_conn_id=self.airbyte_conn_id,
+            deferrable=True,
+        )
+        with pytest.raises(RuntimeError, match=message):
+            sensor.execute_complete(context={}, event={"status": status, "message": message, "job_id": 1})
+
+    def test_execute_complete_succeeds_on_success_event(self):
+        sensor = AirbyteJobSensor(
+            task_id=self.task_id,
+            airbyte_job_id=self.job_id,
+            airbyte_conn_id=self.airbyte_conn_id,
+            deferrable=True,
+        )
+        assert (
+            sensor.execute_complete(
+                context={},
+                event={"status": "success", "message": "Job run 1 has completed successfully.", "job_id": 1},
+            )
+            is None
+        )