From 241ea70994f11e7f01561f18a4ea3143911a3d48 Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Sat, 11 Jul 2026 06:46:27 +0000 Subject: [PATCH] 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 --- .../airflow/providers/http/operators/http.py | 63 +++++++++++++++++-- .../airflow/providers/http/triggers/http.py | 13 ++++ 2 files changed, 70 insertions(+), 6 deletions(-) diff --git a/providers/http/src/airflow/providers/http/operators/http.py b/providers/http/src/airflow/providers/http/operators/http.py index 1c5da7688e5e4..a79e0f804ea42 100644 --- a/providers/http/src/airflow/providers/http/operators/http.py +++ b/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,8 +212,32 @@ 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, @@ -218,7 +245,9 @@ def execute_async(self, context: Context) -> None: 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}, ) diff --git a/providers/http/src/airflow/providers/http/triggers/http.py b/providers/http/src/airflow/providers/http/triggers/http.py index ec77d2634ee32..e57ea41b47e2a 100644 --- a/providers/http/src/airflow/providers/http/triggers/http.py +++ b/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."""