Fix AirbyteJobSensor marking a cancelled job as success in deferrable mode

[steveahnahn]

Problem

• AirbyteJobSensor in deferrable mode resumes via execute_complete, which only fails on an error event.
• A job cancelled while the sensor is deferred makes AirbyteSyncTrigger yield a cancelled event, which fell through to the success path: the task is marked SUCCESS and downstream tasks run against a sync that never completed.
• The same sensor fails correctly on a cancelled job in every other path: non-deferrable poke() raises, the deferrable execute() raises when it sees CANCELLED before deferring, and the sibling AirbyteTriggerSyncOperator.execute_complete already fails on the cancelled event (added in Implement execution_timeout semantics for AirbyteTriggerSyncOperator in deferrable mode #64051, which did not cover the sensor).

Change

• Fail on any non-success event in the sensor's execute_complete, matching poke(). The cancelled event now fails the task, and unrecognized statuses fail closed instead of being treated as success.

Live verification

Ran a real airflow standalone (scheduler, triggerer, API server) against a live local endpoint serving the Airbyte jobs API, with the connection host pointed at it. Both runs: sensor deferred while the job was running, then the job was flipped to cancelled.

• Control run on current main: trigger yielded the cancelled event, task log shows "completed successfully.", run marked Success.
• Same scenario with this fix: task fails with RuntimeError: Job run 1 has been cancelled., run marked Failed.

Tests

• test_execute_complete_fails_when_job_did_not_succeed (parametrized: cancelled, unrecognized status) fails on the code before this change; the failure output shows the bug verbatim ("completed successfully." on a cancelled event).
• test_execute_complete_succeeds_on_success_event guards the happy path.

---

Was generative AI tooling used to co-author this PR?

• Yes, Claude Code (Fable 5)

Generated-by: Claude Code (Fable 5) following the guidelines

[Vamsi-klu]

Confirmed the gap: AirbyteSyncTrigger.run() can yield cancelled events and the old execute_complete only raised on status == "error", so a cancelled job fell straight through to the success log. The new event["status"] != "success" guard closes that and matches poke(), which already raises on a cancelled job. One nit: the branch right above still raises AirflowException while the new one raises RuntimeError (that does line up with the sibling AirbyteTriggerSyncOperator.execute_complete), so this method now throws two different types depending on the path, worth a look if you care about consistency. Small thing on the comment too, execute() here defers without passing execution_deadline, so the trigger's timeout status can't actually fire for this sensor, though the catch-all still handles it fine. Tests cover the cancelled and fail-closed cases.

[jroachgolf84]

LGTM, tests look good.

[SameerMesiah97]

I think the state machine can be more explicit. Please see my comment below.

[steveahnahn]

Both good points and addressed in 02ac81f as it now makes the states explicit (success/error/cancelled, else fails closed) and both branches now raise RuntimeError.

[potiuk]

Thanks — this is a data-correctness bug, not a cosmetic one, and worth spelling out.

execute_complete raised only on status == "error". Every other status fell through to the success path — including "cancelled". So a deferred AirbyteJobSensor whose sync was cancelled logged "completed successfully", returned None, and the task went green. Downstream tasks then ran against incomplete data with nothing anywhere indicating a problem. Silent success on a failed job is about the worst failure mode a sensor can have, and it diverged from poke(), which handled cancellation correctly — so the same sensor behaved differently depending on deferrable.

The fix is right, and the fail-closed default is the part that matters most. Handling success/error/cancelled explicitly fixes today's bug; raising on any unmodelled status fixes tomorrow's, when Airbyte adds a state this sensor has never heard of. Inverting the default from "assume success" to "assume failure" is the correct posture for a sensor.

The tests match: parametrising over error, cancelled and unmapped_status covers all three failure routes, and asserting None on success confirms the happy path wasn't broken in the process. Decrementing the ratchet 6 -> 5 rather than removing the line is also correct, since the other five raises remain.

One small improvement, not blocking: the fail-closed branch raises RuntimeError(event["message"]), so for an unrecognised status the operator sees only whatever Airbyte put in message — the status itself appears just in a log.debug. Including it in the exception (f"Unexpected job status {status!r}: {event['message']}") would make the one case you can't anticipate the easiest to diagnose.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting