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 a763ae3ea2b Keep HttpEventTrigger asset watchers polling after a
failed request (#72376)
a763ae3ea2b is described below
commit a763ae3ea2b8c91f795759914ebfb711f9b4734f
Author: rjgoyln <[email protected]>
AuthorDate: Wed Sep 9 22:44:48 2026 +0800
Keep HttpEventTrigger asset watchers polling after a failed request (#72376)
* Keep HttpEventTrigger asset watchers polling after a failed request
An asset watcher that gives up on its first transient error is not
watching anything, and a failure reported as a bare str(e) leaves no way
to tell a 503 apart from a broken response_check callable. Because the
exception was swallowed rather than raised, the triggerer had nothing to
record either, so the traceback was lost at both layers.
Retrying forever is the opposite failure mode, so the retry gives up
after a bounded number of consecutive failures and lets the error reach
the triggerer.
* Log one HttpEventTrigger traceback per escalation, not per failed poll
A watcher riding out a flaky endpoint wrote a full traceback for every
failed poll, while the escalation itself carried none of its own. The
triggerer already routes its own record into the trigger's log, so the
volume and the emphasis were both backwards.
The effective failure tolerance is the cap times poll_interval rather
than a fixed duration, which the parameter documentation now says.
* Spell out the HttpEventTrigger failure cap in the parameter reference
The count alone does not tell a reader how long a watcher keeps trying,
and describing it as the number of failures "tolerated" read one poll
off from what the code does.
* Fix the HttpEventTrigger poll_interval docstring tag
``:parama`` renders as literal text rather than a parameter entry, so
poll_interval never appeared in the generated API reference. The dropped
test assertions checked for the absence of a log call that run() does not
make, so nothing could ever have failed them.
---
providers/http/docs/triggers.rst | 8 ++
.../src/airflow/providers/http/triggers/http.py | 57 +++++++++---
.../http/tests/unit/http/triggers/test_http.py | 102 +++++++++++++++++++--
3 files changed, 143 insertions(+), 24 deletions(-)
diff --git a/providers/http/docs/triggers.rst b/providers/http/docs/triggers.rst
index d394df6ee19..84294811287 100644
--- a/providers/http/docs/triggers.rst
+++ b/providers/http/docs/triggers.rst
@@ -34,6 +34,8 @@ How It Works
1. Sends requests to an API every ``poll_interval`` seconds (default 60).
2. Uses the callable at ``response_check_path`` to evaluate the API response.
3. If the callable returns ``True``, a ``TriggerEvent`` is emitted. This will
trigger DAGs using this ``AssetWatcher`` for scheduling.
+4. If the request fails or the callable raises, the error is logged and the
trigger polls again after ``poll_interval`` seconds.
+5. After ``max_consecutive_failures`` consecutive failed polls (default 10)
the trigger stops absorbing the error and raises, so the failure and its
traceback reach the trigger log. The triggerer then restarts the watcher.
.. note::
This trigger requires **Airflow >= 3.0** due to dependencies on
``AssetWatcher`` and event-driven scheduling infrastructure.
@@ -147,6 +149,12 @@ Parameters
``poll_interval``
How often, in seconds, the trigger should send a request to the API
+``max_consecutive_failures``
+ Maximum number of consecutive polling failures before the trigger raises.
+ The effective failure tolerance is approximately
+ ``max_consecutive_failures`` × ``poll_interval``.
+ Any poll that completes resets the count, including one where
``response_check`` returns ``False``.
+
Important Notes
---------------
diff --git a/providers/http/src/airflow/providers/http/triggers/http.py
b/providers/http/src/airflow/providers/http/triggers/http.py
index ec77d2634ee..d91559bd689 100644
--- a/providers/http/src/airflow/providers/http/triggers/http.py
+++ b/providers/http/src/airflow/providers/http/triggers/http.py
@@ -297,7 +297,9 @@ class HttpEventTrigger(HttpTrigger, BaseEventTrigger):
:param headers: Additional headers to be passed through as a dict.
:param data: Payload to be uploaded or request parameters.
:param extra_options: Additional kwargs to pass when creating a request.
- :parama poll_interval: How often, in seconds, the trigger should send a
request to the API.
+ :param poll_interval: How often, in seconds, the trigger should send a
request to the API.
+ :param max_consecutive_failures: Maximum number of consecutive polling
failures before the
+ trigger raises. The effective failure tolerance is roughly this times
``poll_interval``.
"""
def __init__(
@@ -311,10 +313,14 @@ class HttpEventTrigger(HttpTrigger, BaseEventTrigger):
data: dict[str, Any] | str | None = None,
extra_options: dict[str, Any] | None = None,
poll_interval: float = 60.0,
+ max_consecutive_failures: int = 10,
):
super().__init__(http_conn_id, auth_type, method, endpoint, headers,
data, extra_options)
self.response_check_path = response_check_path
self.poll_interval = poll_interval
+ if max_consecutive_failures < 1:
+ raise ValueError("max_consecutive_failures must be greater or
equal to 1")
+ self.max_consecutive_failures = max_consecutive_failures
def serialize(self) -> tuple[str, dict[str, Any]]:
"""Serialize HttpEventTrigger arguments and classpath."""
@@ -330,26 +336,49 @@ class HttpEventTrigger(HttpTrigger, BaseEventTrigger):
"extra_options": self.extra_options,
"response_check_path": self.response_check_path,
"poll_interval": self.poll_interval,
+ "max_consecutive_failures": self.max_consecutive_failures,
},
)
async def run(self) -> AsyncIterator[TriggerEvent]:
"""Make a series of asynchronous http calls via a http hook until the
response passes the response check."""
hook = super()._get_async_hook()
- try:
- while True:
+ failures = 0
+ while True:
+ try:
response = await super()._get_response(hook)
- if await self._run_response_check(response):
- break
- await asyncio.sleep(self.poll_interval)
- yield TriggerEvent(
- {
- "status": "success",
- "response": HttpResponseSerializer.serialize(response),
- }
- )
- except Exception as e:
- self.log.error("status: error, message: %s", str(e))
+ check_passed = await self._run_response_check(response)
+ event = (
+ TriggerEvent(
+ {
+ "status": "success",
+ "response":
HttpResponseSerializer.serialize(response),
+ }
+ )
+ if check_passed
+ else None
+ )
+ except Exception as exc:
+ failures += 1
+ if failures >= self.max_consecutive_failures:
+ # The triggerer logs the traceback into this trigger's log
and restarts the
+ # watcher; anything deferred on this trigger fails instead
of hanging.
+ raise
+ self.log.warning(
+ "Poll failed (%s/%s): %r, retrying in %s seconds",
+ failures,
+ self.max_consecutive_failures,
+ exc,
+ self.poll_interval,
+ )
+ else:
+ # "Not yet" clears the count too, or a healthy watcher would
escalate after
+ # max_consecutive_failures ordinary polls.
+ failures = 0
+ if event is not None:
+ yield event
+ return
+ await asyncio.sleep(self.poll_interval)
async def _import_from_response_check_path(self):
"""Import the response check callable from the path provided by the
user."""
diff --git a/providers/http/tests/unit/http/triggers/test_http.py
b/providers/http/tests/unit/http/triggers/test_http.py
index e3338674af4..a1542a41508 100644
--- a/providers/http/tests/unit/http/triggers/test_http.py
+++ b/providers/http/tests/unit/http/triggers/test_http.py
@@ -29,6 +29,7 @@ from requests.structures import CaseInsensitiveDict
from yarl import URL
from airflow.models import Connection
+from airflow.providers.http.exceptions import HttpErrorException
from airflow.providers.http.triggers.http import (
HttpEventTrigger,
HttpResponseSerializer,
@@ -235,6 +236,7 @@ class TestHttpEventTrigger:
"extra_options": TEST_EXTRA_OPTIONS,
"response_check_path": TEST_RESPONSE_CHECK_PATH,
"poll_interval": TEST_POLL_INTERVAL,
+ "max_consecutive_failures": 10,
}
@pytest.mark.asyncio
@@ -259,22 +261,95 @@ class TestHttpEventTrigger:
assert event_trigger._run_response_check.call_count == 2
@pytest.mark.asyncio
+ @pytest.mark.parametrize("failing_call", ["request", "response_check"])
+ @mock.patch(HTTP_PATH.format("asyncio.sleep"), autospec=True)
@mock.patch(HTTP_PATH.format("HttpAsyncHook"))
- async def test_trigger_on_exception_logs_error_and_never_yields(
- self, mock_hook, event_trigger, monkeypatch
+ async def test_trigger_keeps_polling_after_an_exception(
+ self, mock_hook, mock_sleep, failing_call, event_trigger,
client_response, monkeypatch
):
"""
- Tests the HttpEventTrigger logs the appropriate message and does not
yield a TriggerEvent when an exception is raised.
+ Tests the HttpEventTrigger reports the failure and fires on a later
poll.
"""
- mock_hook.return_value.run.side_effect = Exception("Test exception")
+ error = HttpErrorException("503:Service Unavailable")
+ if failing_call == "request":
+ mock_hook.return_value.run.side_effect = [error,
self._mock_run_result(client_response)]
+ event_trigger._run_response_check =
mock.AsyncMock(return_value=True)
+ else:
+ mock_hook.return_value.run.return_value =
self._mock_run_result(client_response)
+ event_trigger._run_response_check =
mock.AsyncMock(side_effect=[error, True])
mock_logger = mock.Mock()
monkeypatch.setattr(type(event_trigger), "log", mock_logger)
+ response = await HttpEventTrigger._convert_response(client_response)
generator = event_trigger.run()
- with pytest.raises(StopAsyncIteration):
- await generator.asend(None)
+ actual = await generator.asend(None)
- mock_logger.error.assert_called_once_with("status: error, message:
%s", "Test exception")
+ assert actual == TriggerEvent(
+ {
+ "status": "success",
+ "response": HttpResponseSerializer.serialize(response),
+ }
+ )
+ assert mock_hook.return_value.run.call_count == 2
+ mock_sleep.assert_awaited_once_with(TEST_POLL_INTERVAL)
+ assert mock_logger.warning.call_count == 1
+
+ @staticmethod
+ def build_trigger_with_failure_cap(max_consecutive_failures: int) ->
HttpEventTrigger:
+ return HttpEventTrigger(
+ endpoint=TEST_ENDPOINT,
+ response_check_path=TEST_RESPONSE_CHECK_PATH,
+ poll_interval=TEST_POLL_INTERVAL,
+ max_consecutive_failures=max_consecutive_failures,
+ )
+
+ @pytest.mark.asyncio
+ @mock.patch(HTTP_PATH.format("asyncio.sleep"), autospec=True)
+ @mock.patch(HTTP_PATH.format("HttpAsyncHook"))
+ async def test_trigger_gives_up_after_max_consecutive_failures(self,
mock_hook, mock_sleep, monkeypatch):
+ trigger = self.build_trigger_with_failure_cap(3)
+ mock_hook.return_value.run.side_effect =
HttpErrorException("503:Service Unavailable")
+ mock_logger = mock.Mock()
+ monkeypatch.setattr(type(trigger), "log", mock_logger)
+
+ with pytest.raises(HttpErrorException):
+ await trigger.run().asend(None)
+
+ assert mock_hook.return_value.run.call_count == 3
+ assert mock_sleep.await_args_list == [mock.call(TEST_POLL_INTERVAL)] *
2
+ assert mock_logger.warning.call_count == 2
+
+ @pytest.mark.asyncio
+ @mock.patch(HTTP_PATH.format("asyncio.sleep"), autospec=True)
+ @mock.patch(HTTP_PATH.format("HttpAsyncHook"))
+ async def test_trigger_resets_failure_count_after_a_completed_poll(
+ self, mock_hook, mock_sleep, client_response
+ ):
+ """
+ Two failures, a poll whose ``response_check`` returns False, then two
more failures:
+ the count must go back to zero rather than merely drop, or the cap of
3 would trip.
+ """
+ trigger = self.build_trigger_with_failure_cap(3)
+ error = HttpErrorException("503:Service Unavailable")
+ mock_hook.return_value.run.side_effect = [
+ error,
+ error,
+ self._mock_run_result(client_response),
+ error,
+ error,
+ self._mock_run_result(client_response),
+ ]
+ trigger._run_response_check = mock.AsyncMock(side_effect=[False, True])
+
+ event = await trigger.run().asend(None)
+
+ assert event.payload["status"] == "success"
+ assert mock_hook.return_value.run.call_count == 6
+ assert mock_sleep.await_args_list == [mock.call(TEST_POLL_INTERVAL)] *
5
+
+ def test_max_consecutive_failures_must_be_positive(self):
+ with pytest.raises(ValueError, match="max_consecutive_failures"):
+ self.build_trigger_with_failure_cap(0)
@pytest.mark.asyncio
async def test_convert_response(self, client_response):
@@ -293,14 +368,21 @@ class TestHttpEventTrigger:
@pytest.mark.db_test
@pytest.mark.asyncio
+ @mock.patch(HTTP_PATH.format("asyncio.sleep"), autospec=True)
@mock.patch("aiohttp.client.ClientSession.post")
- async def test_trigger_on_post_with_data(self, mock_http_post,
event_trigger):
+ async def test_trigger_on_post_with_data(
+ self, mock_http_post, mock_sleep, event_trigger, client_response
+ ):
"""
Test that HttpEventTrigger posts the correct payload when a request is
made.
"""
+ mock_http_post.return_value = self._mock_run_result(client_response)
+ event_trigger._run_response_check = mock.AsyncMock(return_value=True)
+
generator = event_trigger.run()
- with pytest.raises(StopAsyncIteration):
- await generator.asend(None)
+ await generator.asend(None)
+
+ mock_sleep.assert_not_awaited()
mock_http_post.assert_called_once()
_, kwargs = mock_http_post.call_args
assert kwargs["data"] == TEST_DATA
