Add warning for HttpOperator deferrable with non-idempotent methods

providers/http/src/airflow/providers/http/operators/http.py

@@ -17,6 +17,7 @@
 # under the License.
 from __future__ import annotations
 
+import warnings
 from collections.abc import Callable, Sequence
 from typing import TYPE_CHECKING, Any
 
@@ -27,6 +28,8 @@
 from airflow.providers.http.triggers.http import HttpResponseSerializer, HttpTrigger, serialize_auth_type
 from airflow.utils.helpers import merge_dicts
 
+IDEMPOTENT_METHODS = {"GET", "HEAD", "OPTIONS", "PUT", "DELETE", "TRACE"}
+
 if TYPE_CHECKING:
     from requests.auth import AuthBase
 
@@ -209,16 +212,42 @@ def paginate_sync(self, response: Response) -> Response | list[Response]:
         return all_responses
 
     def execute_async(self, context: Context) -> None:
-        self.defer(
-            trigger=HttpTrigger(
+        if self.method.upper() not in IDEMPOTENT_METHODS:
+            warnings.warn(
+                f"HttpOperator with deferrable=True and method={self.method} 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. "
+                "See https://github.com/apache/airflow/issues/67945",
+                UserWarning,
+                stacklevel=2,
+            )
+            # Suppress duplicate warning from HttpTrigger to avoid double warning for operator path
+            with warnings.catch_warnings():
+                warnings.filterwarnings(
+                    "ignore", category=UserWarning, message=".*may send duplicate requests.*"
+                )
+                trigger = HttpTrigger(
+                    http_conn_id=self.http_conn_id,
+                    auth_type=serialize_auth_type(self._resolve_auth_type()),
+                    method=self.method,
+                    endpoint=self.endpoint,
+                    headers=self.headers,
+                    data=self.data,
+                    extra_options=self.extra_options,
+                )
+        else:
+            trigger = HttpTrigger(
                 http_conn_id=self.http_conn_id,
                 auth_type=serialize_auth_type(self._resolve_auth_type()),
                 method=self.method,
                 endpoint=self.endpoint,
                 headers=self.headers,
                 data=self.data,
                 extra_options=self.extra_options,
-            ),
+            )
+        self.defer(
+            trigger=trigger,
             method_name="execute_complete",
         )
 
@@ -300,13 +329,35 @@ def paginate_async(
             next_page_params = self.pagination_function(response)
             if not next_page_params:
                 return self.process_response(context=context, response=all_responses)
-            self.defer(
-                trigger=HttpTrigger(
+            if self.method.upper() not in IDEMPOTENT_METHODS:
+                warnings.warn(
+                    f"HttpOperator with deferrable=True and method={self.method} 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. "
+                    "See https://github.com/apache/airflow/issues/67945",
+                    UserWarning,
+                    stacklevel=2,
+                )
+                with warnings.catch_warnings():
+                    warnings.filterwarnings(
+                        "ignore", category=UserWarning, message=".*may send duplicate requests.*"
+                    )
+                    trigger = HttpTrigger(
+                        http_conn_id=self.http_conn_id,
+                        auth_type=serialize_auth_type(self._resolve_auth_type()),
+                        method=self.method,
+                        **self._merge_next_page_parameters(next_page_params),
+                    )
+            else:
+                trigger = HttpTrigger(
                     http_conn_id=self.http_conn_id,
                     auth_type=serialize_auth_type(self._resolve_auth_type()),
                     method=self.method,
                     **self._merge_next_page_parameters(next_page_params),
-                ),
+                )
+            self.defer(
+                trigger=trigger,
                 method_name="execute_complete",
                 kwargs={"paginated_responses": all_responses},
             )

providers/http/src/airflow/providers/http/triggers/http.py

@@ -21,6 +21,7 @@
 import importlib
 import inspect
 import sys
+import warnings
 from collections.abc import AsyncIterator
 from importlib import import_module
 from typing import TYPE_CHECKING, Any
@@ -36,6 +37,8 @@
 from airflow.providers.http.hooks.http import HttpAsyncHook
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 
+IDEMPOTENT_METHODS = {"GET", "HEAD", "OPTIONS", "PUT", "DELETE", "TRACE"}
+
 if AIRFLOW_V_3_0_PLUS:
     from airflow.triggers.base import BaseEventTrigger
 else:
@@ -141,6 +144,16 @@ def __init__(
         self.headers = headers
         self.data = data
         self.extra_options = extra_options
+        if self.method.upper() not in IDEMPOTENT_METHODS:
+            warnings.warn(
+                f"HttpTrigger with method={self.method} may send duplicate requests if the "
+                "Triggerer restarts. The trigger executes the request in the Triggerer, which "
+                "may be re-run on restart. Use only with idempotent methods or use "
+                "HttpSensorTrigger/HttpEventTrigger for polling. "
+                "See https://github.com/apache/airflow/issues/67945",
+                UserWarning,
+                stacklevel=2,
+            )
 
     def serialize(self) -> tuple[str, dict[str, Any]]:
         """Serialize HttpTrigger arguments and classpath."""