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 4572e0b42f5 Fix AirbyteJobSensor marking a cancelled job as success in
deferrable mode (#69786)
4572e0b42f5 is described below
commit 4572e0b42f5cdf5aa024edb28df6eaf65fdb8e7a
Author: Steve Ahn <[email protected]>
AuthorDate: Fri Jul 31 12:27:55 2026 -0700
Fix AirbyteJobSensor marking a cancelled job as success in deferrable mode
(#69786)
---
generated/known_airflow_exceptions.txt | 2 +-
.../airflow/providers/airbyte/sensors/airbyte.py | 25 ++++++++++------
.../tests/unit/airbyte/sensors/test_airbyte.py | 34 ++++++++++++++++++++++
3 files changed, 51 insertions(+), 10 deletions(-)
diff --git a/generated/known_airflow_exceptions.txt
b/generated/known_airflow_exceptions.txt
index 53517fa9dcc..b99a61c7807 100644
--- a/generated/known_airflow_exceptions.txt
+++ b/generated/known_airflow_exceptions.txt
@@ -33,7 +33,7 @@
devel-common/src/tests_common/test_utils/sftp_system_helpers.py::1
devel-common/src/tests_common/test_utils/watcher.py::1
providers/airbyte/src/airflow/providers/airbyte/hooks/airbyte.py::8
providers/airbyte/src/airflow/providers/airbyte/operators/airbyte.py::3
-providers/airbyte/src/airflow/providers/airbyte/sensors/airbyte.py::6
+providers/airbyte/src/airflow/providers/airbyte/sensors/airbyte.py::5
providers/alibaba/src/airflow/providers/alibaba/cloud/hooks/analyticdb_spark.py::6
providers/alibaba/src/airflow/providers/alibaba/cloud/hooks/oss.py::11
providers/alibaba/src/airflow/providers/alibaba/cloud/operators/analyticdb_spark.py::1
diff --git a/providers/airbyte/src/airflow/providers/airbyte/sensors/airbyte.py
b/providers/airbyte/src/airflow/providers/airbyte/sensors/airbyte.py
index 5c6be4b9c33..5bd92b80e70 100644
--- a/providers/airbyte/src/airflow/providers/airbyte/sensors/airbyte.py
+++ b/providers/airbyte/src/airflow/providers/airbyte/sensors/airbyte.py
@@ -126,15 +126,22 @@ class AirbyteJobSensor(BaseSensorOperator):
)
def execute_complete(self, context: Context, event: Any = None) -> None:
- """
- Invoke this callback when the trigger fires; return immediately.
+ """Invoke this callback when the trigger fires and fail unless the job
succeeded."""
+ status = event["status"]
- Relies on trigger to throw an exception, otherwise it assumes
execution was
- successful.
- """
- if event["status"] == "error":
+ if status == "success":
+ self.log.info("%s completed successfully.", self.task_id)
+ return None
+
+ if status == "error":
self.log.debug("An error occurred with context: %s", context)
- raise AirflowException(event["message"])
+ raise RuntimeError(event["message"])
+
+ if status == "cancelled":
+ # A cancelled job must fail the sensor, matching the poke()
behavior.
+ self.log.debug("Job was cancelled, context: %s", context)
+ raise RuntimeError(event["message"])
- self.log.info("%s completed successfully.", self.task_id)
- return None
+ # Fail closed: a status this sensor does not model must not reach the
success path.
+ self.log.debug("Unexpected status %r, context: %s", status, context)
+ raise RuntimeError(event["message"])
diff --git a/providers/airbyte/tests/unit/airbyte/sensors/test_airbyte.py
b/providers/airbyte/tests/unit/airbyte/sensors/test_airbyte.py
index 69b5d0bffe0..706ff6e7bdb 100644
--- a/providers/airbyte/tests/unit/airbyte/sensors/test_airbyte.py
+++ b/providers/airbyte/tests/unit/airbyte/sensors/test_airbyte.py
@@ -117,3 +117,37 @@ class TestAirbyteJobSensor:
)
assert sensor.poke_interval == 10
assert sensor.timeout == 3600
+
+ @pytest.mark.parametrize(
+ ("status", "message"),
+ [
+ pytest.param("error", "Job run 1 has failed.", id="error"),
+ 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
+ )
