ColtenOuO opened a new pull request, #72133: URL: https://github.com/apache/airflow/pull/72133
### Summary Deferrable operators that hand a long-lived request off to a remote server currently have no built-in way to resume once that server replies over a WebSocket connection. This closes #21139, which asked for exactly this: a trigger that opens a WebSocket connection, hands off the wait to the triggerer, and fires once the remote server replies. Without this, the only options today are a plain HTTP-based deferrable sensor/trigger that polls, or a non-deferrable operator that blocks synchronously, and both have real costs for a long-running, unknown-duration remote job. Polling means repeated API calls against the remote server for the entire duration of the job — on a `poke_interval` of a few seconds that adds up fast over a job that can run for hours, and it puts avoidable load on a server that has no way to tell Airflow "I'm still working" other than answering yet another request. Polling also caps how quickly Airflow finds out the job is done: the result only surfaces on the next poll, so the effective latency is bounded by `poke_interval`, not by when the server actually finished. A synchronous, non-deferrable wait avoids the polling cost but instead ties up a full worker slot for the whole duration, which does not scale when many such jobs run concurrently. A WebSocket trigger avoids all three: the remote server pu shes the reply the moment it is ready (no poll delay, no repeated calls), and the wait happens in the triggerer rather than on a worker. ### Changes - Added `WebSocketTrigger` in `airflow/providers/standard/triggers/websocket.py`. Opens a `ws://`/`wss://` connection with `websockets.asyncio.client`, optionally sends an initial message, and fires a `TriggerEvent` with the payload once a message is received. - Added `WebSocketSensor` in `airflow/providers/standard/sensors/websocket.py`, following the same `deferrable` / `poke()` / `execute()` / `execute_complete()` pattern as `FileSensor`. The synchronous `poke()` path uses `websockets.sync.client`. - Registered both in `provider.yaml` (`triggers` and `sensors` sections) and mirrored those additions by hand into `get_provider_info.py`, since that file is normally regenerated by the release tooling rather than by a local prek hook. - Added the `websocket` optional extra (`websockets>=14.0`) to `pyproject.toml`, and added `websockets` to the provider's dev dependency group so tests can run without needing the extra installed separately. `uv.lock` was updated to match via the `update-providers-dependencies` prek hook. - Added `docs/sensors/websocket.rst` and linked it from `provider.yaml`'s `how-to-guide` list, following the same structure as `docs/sensors/file.rst`. - Added a sync and a deferrable example to `example_dags/example_sensors.py` (`example_websocket_sensor` / `example_websocket_sensor_async` blocks), referenced by the new docs page via `exampleinclude`. - Added unit tests for both the trigger (serialization, message received, no-message-to-send path) and the sensor (poke success, poke timeout, defer wiring) in `tests/unit/standard/triggers/test_websocket.py` and `tests/unit/standard/sensors/test_websocket.py`. closes: #21139 --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Claude Code (Sonnet 5) Generated-by: Claude Code (Sonnet 5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
