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 5dcee04992a Add warning for HttpOperator deferrable with
non-idempotent methods (#69748)
5dcee04992a is described below
commit 5dcee04992a075af7f8505801b5be0904e25ca9d
Author: deepinsight coder <[email protected]>
AuthorDate: Sat Aug 29 06:32:14 2026 -0700
Add warning for HttpOperator deferrable with non-idempotent methods (#69748)
* Add warning for HttpOperator deferrable with non-idempotent methods
HttpOperator(deferrable=True) executes the HTTP request inside
HttpTrigger.run() in the Triggerer. On triggerer restart, the trigger
is re-created from serialize() and run() re-executes, causing duplicate
POST/PATCH requests. This is silent data duplication.
Add UserWarning when deferrable=True is used with POST/PATCH etc,
guiding users to idempotent methods or sensor-based polling pattern
like AirbyteOperator (side-effect in worker, trigger polls with GET).
Long-term fix should execute non-idempotent requests in worker and
only poll in trigger.
Fixes: #67945
* Avoid duplicate HTTP deferrable warnings
The deferrable HTTP warning should stay visible for unsafe methods without
showing the operator and trigger versions together, so the shared wording now
has focused coverage for both entry points.
* Warn once for non-idempotent deferrable HTTP methods
Trigger reconstruction can happen repeatedly in the Triggerer, where
warnings do not reach Dag authors. Surface the guidance where the deferrable
choice is made without producing repeated trigger-side noise.
* Warn once from HttpOperator.execute for non-idempotent deferrable methods
Address review feedback on the delivery mechanism:
- Emit a single self.log.warning from HttpOperator.execute() when
deferrable=True and the method is not idempotent (RFC 9110 §9.2.2),
so Dag authors see the advisory in task logs once per attempt.
- Do not warn from __init__ (parse-time flood) or HttpTrigger (Triggerer
reconstructs from serialized kwargs on every resume/restart).
- No catch_warnings/filterwarnings suppression path.
- Cover method case, None, pagination cardinality, PUT/DELETE silence,
PATCH/custom methods, deferrable=False, and trigger deserialize.
Fixes: #67945
* Allow silencing HttpOperator deferrable non-idempotent warnings
Dag authors who accept Triggerer-restart retry risk need a per-task
opt-out, and task logs should point at provider docs rather than a
GitHub issue that operators cannot act on.
* Add a silence flag and docs for the HttpOperator deferrable warning
Dag authors who intentionally POST in deferrable mode had no documented
opt-out, and the advisory pointed at a GitHub issue instead of provider
docs.
---------
Co-authored-by: probe <[email protected]>
Co-authored-by: Cursor Agent <[email protected]>
---
providers/http/docs/deferrable.rst | 73 ++++++++
providers/http/docs/index.rst | 1 +
providers/http/docs/operators.rst | 7 +-
.../src/airflow/providers/http/operators/http.py | 32 ++++
.../http/tests/unit/http/operators/test_http.py | 184 ++++++++++++++++++++-
5 files changed, 294 insertions(+), 3 deletions(-)
diff --git a/providers/http/docs/deferrable.rst
b/providers/http/docs/deferrable.rst
new file mode 100644
index 00000000000..9c23d733d76
--- /dev/null
+++ b/providers/http/docs/deferrable.rst
@@ -0,0 +1,73 @@
+
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _howto/deferrable:HttpOperator:
+
+Deferrable HttpOperator
+=======================
+
+:class:`~airflow.providers.http.operators.http.HttpOperator` can run in
deferrable mode
+(``deferrable=True``, or via ``[operators] default_deferrable``). In that mode
the operator
+defers immediately and the HTTP request is executed in the Triggerer by
+:class:`~airflow.providers.http.triggers.http.HttpTrigger`.
+
+Triggerer restart can replay the request
+----------------------------------------
+
+The Triggerer persists trigger kwargs and reconstructs the trigger after a
restart.
+``HttpTrigger.run()`` then issues the HTTP request again. That is safe for
+idempotent methods and unsafe for methods that create a new side effect on
every
+call (the operator default is ``POST``).
+
+HttpOperator treats these methods as idempotent, matching RFC 9110 §9.2.2:
+
+* ``GET``
+* ``HEAD``
+* ``OPTIONS``
+* ``PUT``
+* ``DELETE``
+* ``TRACE``
+
+``POST``, ``PATCH``, and any other method emit one task-log warning per attempt
+when used with deferrable mode.
+
+Silencing the warning
+---------------------
+
+If a duplicate request is acceptable for your endpoint, set
+``warn_on_non_idempotent=False``:
+
+.. code-block:: python
+
+ HttpOperator(
+ task_id="create_resource",
+ method="POST",
+ endpoint="/resources",
+ deferrable=True,
+ warn_on_non_idempotent=False,
+ )
+
+Safer alternatives for polling or waiting on an HTTP condition are
+:class:`~airflow.providers.http.sensors.http.HttpSensor` or an event-driven
+trigger such as
:class:`~airflow.providers.http.triggers.http.HttpEventTrigger`.
+For a one-shot non-idempotent call, prefer ``deferrable=False`` so the worker
+issues the request once.
+
+This page is the target of the task-log warning. The warning does not prevent a
+duplicate request; it only tells Dag authors that a Triggerer restart can
replay
+it.
diff --git a/providers/http/docs/index.rst b/providers/http/docs/index.rst
index 2a6a1cf272b..99a7dbeefc9 100644
--- a/providers/http/docs/index.rst
+++ b/providers/http/docs/index.rst
@@ -36,6 +36,7 @@
Connection types <connections/http>
Operators <operators>
+ Deferrable HttpOperator <deferrable>
Triggers <triggers>
.. toctree::
diff --git a/providers/http/docs/operators.rst
b/providers/http/docs/operators.rst
index 006aaec7ead..1e75287742b 100644
--- a/providers/http/docs/operators.rst
+++ b/providers/http/docs/operators.rst
@@ -52,6 +52,11 @@ HttpOperator
Use the :class:`~airflow.providers.http.operators.http.HttpOperator` to call
HTTP requests and get
the response text back.
+Deferrable mode runs the request in the Triggerer. A Triggerer restart can
replay
+it, so non-idempotent methods (including the default ``POST``) log a warning.
+See :ref:`howto/deferrable:HttpOperator` for the idempotent method set and how
+to silence the warning with ``warn_on_non_idempotent=False``.
+
.. warning:: Configuring ``https`` via HttpOperator is counter-intuitive
For historical reasons, configuring ``HTTPS`` connectivity via HTTP
operator is, well, difficult and
@@ -127,8 +132,6 @@ Here we pass form data to a ``POST`` operation which is
equal to a usual form su
:start-after: [START howto_operator_http_task_post_op_formenc]
:end-before: [END howto_operator_http_task_post_op_formenc]
-
-
The :class:`~airflow.providers.http.operators.paginated.HttpOperator` also
allows to repeatedly call an API
endpoint, typically to loop over its pages. All API responses are stored in
memory by the Operator and returned
in one single result. Thus, it can be more memory and CPU intensive compared
to a non-paginated call.
diff --git a/providers/http/src/airflow/providers/http/operators/http.py
b/providers/http/src/airflow/providers/http/operators/http.py
index 1c5da7688e5..04afb068369 100644
--- a/providers/http/src/airflow/providers/http/operators/http.py
+++ b/providers/http/src/airflow/providers/http/operators/http.py
@@ -33,6 +33,11 @@ if TYPE_CHECKING:
from airflow.providers.http.hooks.http import HttpHook
from airflow.sdk import Context
+# Idempotent methods per RFC 9110 §9.2.2 (formerly RFC 7231).
+# PUT and DELETE are idempotent even though they are not safe; PATCH is not
idempotent.
+IDEMPOTENT_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "PUT", "DELETE",
"TRACE"})
+HTTP_DEFERRABLE_DOCS =
"https://airflow.apache.org/docs/apache-airflow-providers-http/stable/deferrable.html";
+
class HttpOperator(BaseOperator):
"""
@@ -42,6 +47,9 @@ class HttpOperator(BaseOperator):
For more information on how to use this operator, take a look at the
guide:
:ref:`howto/operator:HttpOperator`
+ For deferrable-mode idempotency caveats, see:
+ :ref:`howto/deferrable:HttpOperator`
+
:param http_conn_id: The :ref:`http connection<howto/connection:http>` to
run
the operator against
:param endpoint: The relative part of the full url. (templated)
@@ -85,6 +93,10 @@ class HttpOperator(BaseOperator):
:param tcp_keep_alive_interval: The TCP Keep Alive interval parameter
(corresponds to
``socket.TCP_KEEPINTVL``)
:param deferrable: Run operator in the deferrable mode
+ :param warn_on_non_idempotent: When True (default), log a warning if
deferrable
+ mode is used with a method outside the RFC 9110 §9.2.2 idempotent set.
+ Set False to silence the warning when a duplicate request is
acceptable.
+ See :ref:`howto/deferrable:HttpOperator`.
:param retry_args: Arguments which define the retry behaviour.
See Tenacity documentation at https://github.com/jd/tenacity
"""
@@ -119,6 +131,7 @@ class HttpOperator(BaseOperator):
tcp_keep_alive_count: int = 20,
tcp_keep_alive_interval: int = 30,
deferrable: bool = conf.getboolean("operators", "default_deferrable",
fallback=False),
+ warn_on_non_idempotent: bool = True,
retry_args: dict[str, Any] | None = None,
**kwargs: Any,
) -> None:
@@ -139,6 +152,7 @@ class HttpOperator(BaseOperator):
self.tcp_keep_alive_count = tcp_keep_alive_count
self.tcp_keep_alive_interval = tcp_keep_alive_interval
self.deferrable = deferrable
+ self.warn_on_non_idempotent = warn_on_non_idempotent
self.retry_args = retry_args
self.request_kwargs = request_kwargs or {}
@@ -163,10 +177,28 @@ class HttpOperator(BaseOperator):
def execute(self, context: Context) -> Any:
if self.deferrable:
+ self._warn_if_deferrable_non_idempotent_method()
self.execute_async(context=context)
else:
return self.execute_sync(context=context)
+ def _warn_if_deferrable_non_idempotent_method(self) -> None:
+ """Log once per attempt when a deferrable request may be re-sent on
Triggerer restart."""
+ if not self.warn_on_non_idempotent:
+ return
+ method = (self.method or "").upper()
+ if method in IDEMPOTENT_METHODS:
+ return
+ self.log.warning(
+ "HttpOperator with deferrable=True and method=%s may send
duplicate "
+ "requests if the Triggerer restarts. Deferrable mode executes the
request in "
+ "the Triggerer, which may be re-run on restart. Use only with
idempotent methods "
+ "or use HttpSensor/EventSensor for polling. Set
warn_on_non_idempotent=False to "
+ "silence this warning. See %s",
+ self.method,
+ HTTP_DEFERRABLE_DOCS,
+ )
+
def execute_sync(self, context: Context) -> Any:
self.log.info("Calling HTTP method")
if self.retry_args:
diff --git a/providers/http/tests/unit/http/operators/test_http.py
b/providers/http/tests/unit/http/operators/test_http.py
index 50b66e5f612..863df3ce01b 100644
--- a/providers/http/tests/unit/http/operators/test_http.py
+++ b/providers/http/tests/unit/http/operators/test_http.py
@@ -20,7 +20,9 @@ from __future__ import annotations
import base64
import contextlib
import json
+import logging
import pickle
+import warnings
from types import SimpleNamespace
from unittest import mock
from unittest.mock import call, patch
@@ -35,9 +37,15 @@ from airflow.hooks import base
from airflow.models import Connection
from airflow.providers.common.compat.sdk import AirflowException, TaskDeferred
from airflow.providers.http.hooks.http import HttpHook
-from airflow.providers.http.operators.http import HttpOperator
+from airflow.providers.http.operators.http import (
+ HTTP_DEFERRABLE_DOCS,
+ IDEMPOTENT_METHODS,
+ HttpOperator,
+)
from airflow.providers.http.triggers.http import HttpResponseSerializer,
HttpTrigger, serialize_auth_type
+_DEFER_WARN_FRAGMENT = "may send duplicate requests if the Triggerer restarts"
+
@mock.patch.dict("os.environ",
AIRFLOW_CONN_HTTP_EXAMPLE="http://www.example.com";)
class TestHttpOperator:
@@ -113,6 +121,180 @@ class TestHttpOperator:
operator.execute({})
assert isinstance(exc.value.trigger, HttpTrigger), "Trigger is not a
HttpTrigger"
+ def _defer_warning_records(self, caplog):
+ return [
+ r
+ for r in caplog.records
+ if r.levelno >= logging.WARNING and _DEFER_WARN_FRAGMENT in
r.getMessage()
+ ]
+
+ @pytest.mark.parametrize(
+ ("method", "deferrable", "expect_warning"),
+ [
+ # E1: lowercase non-idempotent
+ ("post", True, True),
+ # E6: PATCH is not idempotent (RFC 9110)
+ ("PATCH", True, True),
+ # E7: PUT / DELETE are idempotent — do not warn
+ ("PUT", True, False),
+ ("DELETE", True, False),
+ # E8: deferrable=False + POST — no warning
+ ("POST", False, False),
+ # E9: custom method outside the idempotent set
+ ("REPORT", True, True),
+ # Standard defaults / safe methods
+ ("POST", True, True),
+ ("GET", True, False),
+ ("HEAD", True, False),
+ ("OPTIONS", True, False),
+ ("TRACE", True, False),
+ ("get", True, False),
+ ],
+ )
+ def test_deferrable_non_idempotent_warning_on_execute(
+ self, monkeypatch, caplog, method, deferrable, expect_warning
+ ):
+ """Exactly one log.warning per task attempt when deferrable +
non-idempotent; else zero."""
+ captured = self._capture_defer(monkeypatch) if deferrable else None
+ operator = HttpOperator(task_id="test_HTTP_op", method=method,
deferrable=deferrable)
+
+ with caplog.at_level(logging.WARNING):
+ if deferrable:
+ operator.execute(context={})
+ else:
+ # Non-deferrable path needs a network mock; only assert no
advisory was logged.
+ with mock.patch.object(operator, "execute_sync",
return_value="ok") as sync:
+ result = operator.execute(context={})
+ assert result == "ok"
+ sync.assert_called_once()
+
+ records = self._defer_warning_records(caplog)
+ if expect_warning:
+ assert len(records) == 1
+ message = records[0].getMessage()
+ assert f"method={method}" in message
+ assert HTTP_DEFERRABLE_DOCS in message
+ assert "issues/67945" not in message
+ assert "warn_on_non_idempotent=False" in message
+ else:
+ assert len(records) == 0
+
+ if deferrable:
+ assert isinstance(captured["trigger"], HttpTrigger)
+
+ def test_silences_warning_when_warn_on_non_idempotent_false(self,
monkeypatch, caplog):
+ captured = self._capture_defer(monkeypatch)
+ operator = HttpOperator(
+ task_id="test_HTTP_op",
+ method="POST",
+ deferrable=True,
+ warn_on_non_idempotent=False,
+ )
+
+ with caplog.at_level(logging.WARNING):
+ operator.execute(context={})
+
+ assert self._defer_warning_records(caplog) == []
+ assert isinstance(captured["trigger"], HttpTrigger)
+
+ def test_does_not_warn_on_construction_for_deferrable_post(self, caplog):
+ """E10: construction / parse-time must not warn (would flood the Dag
processor)."""
+ with caplog.at_level(logging.WARNING),
warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ HttpOperator(task_id="test_HTTP_op", method="POST",
deferrable=True)
+
+ assert self._defer_warning_records(caplog) == []
+ assert not [w for w in caught if _DEFER_WARN_FRAGMENT in
str(w.message)]
+
+ def test_deferrable_method_none_does_not_raise_and_warns(self,
monkeypatch, caplog):
+ """E2: method is None — guard so .upper() is not called on None; treat
as non-idempotent."""
+ captured = self._capture_defer(monkeypatch)
+ operator = HttpOperator(task_id="test_HTTP_op", method="POST",
deferrable=True)
+ operator.method = None
+
+ with caplog.at_level(logging.WARNING):
+ operator.execute(context={})
+
+ assert len(self._defer_warning_records(caplog)) == 1
+ assert isinstance(captured["trigger"], HttpTrigger)
+
+ def test_paginated_deferrable_non_idempotent_warns_once(self, monkeypatch,
caplog):
+ """E3: paginated non-idempotent — exactly ONE warning for the task
attempt, not per page."""
+ captured_defers: list[dict] = []
+
+ def _fake_defer(self, *, trigger, method_name, **kwargs):
+ captured_defers.append({"trigger": trigger, "kwargs": kwargs})
+
+ monkeypatch.setattr(HttpOperator, "defer", _fake_defer)
+
+ def pagination_function(response: Response) -> dict | None:
+ if response.url.endswith("/page1"):
+ return {"endpoint": "/page2"}
+ return None
+
+ operator = HttpOperator(
+ task_id="test_HTTP_op",
+ method="POST",
+ deferrable=True,
+ pagination_function=pagination_function,
+ )
+
+ with caplog.at_level(logging.WARNING):
+ operator.execute(context={})
+ # Simulate first page completing and requesting another page
(second defer).
+ page1 = Response()
+ page1._content = b'{"page": 1}'
+ page1.url = "http://test:8080/page1";
+ page1.status_code = 200
+ page1.headers["Content-Type"] = "application/json"
+ operator.execute_complete(
+ context={},
+ event={
+ "status": "success",
+ "response": HttpResponseSerializer.serialize(page1),
+ },
+ )
+
+ assert len(self._defer_warning_records(caplog)) == 1
+ # Initial execute defers once; execute_complete with pagination defers
again — still one warn.
+ assert len(captured_defers) == 2
+
+ def test_http_trigger_deserialize_emits_no_warning(self, caplog):
+ """E4/E5: Trigger reconstruct from serialized kwargs (Triggerer
restart) must not warn."""
+ trigger = HttpTrigger(method="POST", endpoint="/",
http_conn_id="http_default")
+ classpath, kwargs = trigger.serialize()
+ assert classpath.endswith("HttpTrigger")
+
+ with caplog.at_level(logging.WARNING),
warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ # Triggerer path: trigger_class(**deserialised_kwargs)
+ reconstructed = HttpTrigger(**kwargs)
+
+ assert reconstructed.method == "POST"
+ assert self._defer_warning_records(caplog) == []
+ assert not [w for w in caught if _DEFER_WARN_FRAGMENT in
str(w.message)]
+ assert not any(issubclass(w.category, UserWarning) for w in caught)
+
+ def test_idempotent_methods_match_rfc9110(self):
+ # PUT and DELETE are idempotent per RFC 9110 §9.2.2 even though not
safe.
+ assert "PUT" in IDEMPOTENT_METHODS
+ assert "DELETE" in IDEMPOTENT_METHODS
+ assert "PATCH" not in IDEMPOTENT_METHODS
+ assert "POST" not in IDEMPOTENT_METHODS
+
+ def test_deferrable_warning_links_to_stable_docs(self, monkeypatch,
caplog):
+ self._capture_defer(monkeypatch)
+ operator = HttpOperator(task_id="test_HTTP_op", method="POST",
deferrable=True)
+
+ with caplog.at_level(logging.WARNING):
+ operator.execute(context={})
+
+ records = self._defer_warning_records(caplog)
+ assert len(records) == 1
+ message = records[0].getMessage()
+ assert HTTP_DEFERRABLE_DOCS in message
+ assert "issues/67945" not in message
+
def test_async_execute_successfully(self, requests_mock):
operator = HttpOperator(
task_id="test_HTTP_op",
