Fix AirbyteJobSensor marking a cancelled job as success in deferrable mode

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

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
+        )