Make return_immediately configurable for all Pub/Sub modules and mark the setting as deprecated

providers/google/src/airflow/providers/google/cloud/hooks/pubsub.py

@@ -520,6 +520,12 @@ def pull(
             the base64-encoded message content. See
             https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.ReceivedMessage
         """
+        warnings.warn(
+            "The `return_immediately` parameter is deprecated and will be removed in a future release.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
+        )
+
         subscriber = self.subscriber_client
         # E501
         subscription_path = f"projects/{project_id}/subscriptions/{subscription}"
@@ -712,6 +718,12 @@ async def pull(
             the base64-encoded message content. See
             https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.ReceivedMessage
         """
+        warnings.warn(
+            "The `return_immediately` parameter is deprecated and will be removed in a future release.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
+        )
+
         subscriber = await self._get_subscriber_client()
         subscription_path = f"projects/{project_id}/subscriptions/{subscription}"
         self.log.info("Pulling max %d messages from subscription (path) %s", max_messages, subscription_path)

providers/google/src/airflow/providers/google/cloud/operators/pubsub.py

@@ -25,6 +25,7 @@
 
 from __future__ import annotations
 
+import warnings
 from collections.abc import Callable, Sequence
 from functools import cached_property
 from typing import TYPE_CHECKING, Any
@@ -42,6 +43,7 @@
     SchemaSettings,
 )
 
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.common.compat.sdk import AirflowException, conf
 from airflow.providers.google.cloud.hooks.pubsub import PubSubHook
 from airflow.providers.google.cloud.links.pubsub import PubSubSubscriptionLink, PubSubTopicLink
@@ -799,6 +801,13 @@ class PubSubPullOperator(GoogleCloudBaseOperator):
     :param deferrable: If True, run the task in the deferrable mode.
     :param poll_interval: Time (seconds) to wait between two consecutive calls to check the job.
         The default is 300 seconds.
+    :param return_immediately: If this field set to true, the system will
+        respond immediately even if it there are no messages available to
+        return in the ``Pull`` response. Otherwise, the system may wait
+        (for a bounded amount of time) until at least one message is available,
+        rather than returning no messages. Warning: setting this field to
+        ``true`` is discouraged because it adversely impacts the performance
+        of ``Pull`` operations. We recommend that users do not set this field.
     """
 
     template_fields: Sequence[str] = (
@@ -819,6 +828,7 @@ def __init__(
         impersonation_chain: str | Sequence[str] | None = None,
         deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
         poll_interval: int = 300,
+        return_immediately: bool = True,
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
@@ -831,6 +841,15 @@ def __init__(
         self.impersonation_chain = impersonation_chain
         self.deferrable = deferrable
         self.poll_interval = poll_interval
+        self.return_immediately = return_immediately
+
+        warnings.warn(
+            "The `return_immediately` parameter is deprecated and will be removed in a future release. "
+            "Its default value will be changed to `False` in the next major release. "
+            "Planned removal date: August 01, 2026.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
+        )
 
     def execute(self, context: Context) -> list:
         if self.deferrable:
@@ -843,6 +862,7 @@ def execute(self, context: Context) -> list:
                     gcp_conn_id=self.gcp_conn_id,
                     poke_interval=self.poll_interval,
                     impersonation_chain=self.impersonation_chain,
+                    return_immediately=self.return_immediately,
                 ),
                 method_name=GOOGLE_DEFAULT_DEFERRABLE_METHOD_NAME,
             )
@@ -855,7 +875,7 @@ def execute(self, context: Context) -> list:
             project_id=self.project_id,
             subscription=self.subscription,
             max_messages=self.max_messages,
-            return_immediately=True,
+            return_immediately=self.return_immediately,
         )
 
         handle_messages = self.messages_callback or self._default_message_callback

providers/google/src/airflow/providers/google/cloud/sensors/pubsub.py

@@ -22,10 +22,12 @@
 from collections.abc import Callable, Sequence
 from datetime import timedelta
 from typing import TYPE_CHECKING, Any
+import warnings
 
 from google.cloud import pubsub_v1
 from google.cloud.pubsub_v1.types import ReceivedMessage
 
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.common.compat.sdk import AirflowException, BaseSensorOperator, conf
 from airflow.providers.google.cloud.hooks.pubsub import PubSubHook
 from airflow.providers.google.cloud.triggers.pubsub import PubsubPullTrigger
@@ -135,6 +137,14 @@ def __init__(
         self.poke_interval = poke_interval
         self._return_value = None
 
+        warnings.warn(
+            "The `return_immediately` parameter is deprecated and will be removed in a future release. "
+            "Its default value will be changed to `False` in the next major release. "
+            "Planned removal date: August 01, 2026.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
+        )
+
     def poke(self, context: Context) -> bool:
         hook = PubSubHook(
             gcp_conn_id=self.gcp_conn_id,
@@ -176,6 +186,7 @@ def execute(self, context: Context) -> None:
                 poke_interval=self.poke_interval,
                 gcp_conn_id=self.gcp_conn_id,
                 impersonation_chain=self.impersonation_chain,
+                return_immediately=self.return_immediately,
             ),
             method_name="execute_complete",
         )

providers/google/src/airflow/providers/google/cloud/triggers/pubsub.py

@@ -19,12 +19,14 @@
 from __future__ import annotations
 
 import asyncio
+import warnings
 from collections.abc import AsyncIterator, Sequence
 from functools import cached_property
 from typing import Any
 
 from google.cloud.pubsub_v1.types import ReceivedMessage
 
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.google.cloud.hooks.pubsub import PubSubAsyncHook
 from airflow.providers.google.version_compat import AIRFLOW_V_3_0_PLUS
 from airflow.triggers.base import TriggerEvent
@@ -55,6 +57,13 @@ class PubsubPullTrigger(BaseEventTrigger):
         If set as a sequence, the identities from the list must grant
         Service Account Token Creator IAM role to the directly preceding identity, with first
         account from the list granting this role to the originating account (templated).
+    :param return_immediately: If this field set to true, the system will
+        respond immediately even if it there are no messages available to
+        return in the ``Pull`` response. Otherwise, the system may wait
+        (for a bounded amount of time) until at least one message is available,
+        rather than returning no messages. Warning: setting this field to
+        ``true`` is discouraged because it adversely impacts the performance
+        of ``Pull`` operations. We recommend that users do not set this field.
     """
 
     def __init__(
@@ -66,6 +75,7 @@ def __init__(
         gcp_conn_id: str,
         poke_interval: float = 10.0,
         impersonation_chain: str | Sequence[str] | None = None,
+        return_immediately: bool = True,
     ):
         super().__init__()
         self.project_id = project_id
@@ -75,6 +85,15 @@ def __init__(
         self.poke_interval = poke_interval
         self.gcp_conn_id = gcp_conn_id
         self.impersonation_chain = impersonation_chain
+        self.return_immediately = return_immediately
+
+        warnings.warn(
+            "The `return_immediately` parameter is deprecated and will be removed in a future release. "
+            "Its default value will be changed to `False` in the next major release. "
+            "Planned removal date: August 01, 2026.",
+            AirflowProviderDeprecationWarning,
+            stacklevel=2,
+        )
 
     def serialize(self) -> tuple[str, dict[str, Any]]:
         """Serialize PubsubPullTrigger arguments and classpath."""
@@ -88,6 +107,7 @@ def serialize(self) -> tuple[str, dict[str, Any]]:
                 "poke_interval": self.poke_interval,
                 "gcp_conn_id": self.gcp_conn_id,
                 "impersonation_chain": self.impersonation_chain,
+                "return_immediately": self.return_immediately,
             },
         )
 
@@ -97,7 +117,7 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
                 project_id=self.project_id,
                 subscription=self.subscription,
                 max_messages=self.max_messages,
-                return_immediately=True,
+                return_immediately=self.return_immediately,
             ):
                 if self.ack_messages:
                     await self.message_acknowledgement(pulled_messages)

providers/google/tests/unit/google/cloud/operators/test_pubsub.py

@@ -34,6 +34,7 @@
     PubSubPublishMessageOperator,
     PubSubPullOperator,
 )
+from airflow.providers.google.cloud.triggers.pubsub import PubsubPullTrigger
 
 TASK_ID = "test-task-id"
 TEST_PROJECT = "test-project"
@@ -513,6 +514,21 @@ def messages_callback(
 
         assert response == messages_callback_return_value
 
+    @mock.patch("airflow.providers.google.cloud.operators.pubsub.PubSubHook")
+    def test_execute_with_return_immediately_false(self, mock_hook):
+        operator = PubSubPullOperator(
+            task_id=TASK_ID,
+            project_id=TEST_PROJECT,
+            subscription=TEST_SUBSCRIPTION,
+            return_immediately=False,
+        )
+
+        mock_hook.return_value.pull.return_value = []
+        operator.execute({})
+        mock_hook.return_value.pull.assert_called_once_with(
+            project_id=TEST_PROJECT, subscription=TEST_SUBSCRIPTION, max_messages=5, return_immediately=False
+        )
+
     @pytest.mark.db_test
     @mock.patch("airflow.providers.google.cloud.operators.pubsub.PubSubHook")
     def test_execute_deferred(self, mock_hook):
@@ -526,9 +542,12 @@ def test_execute_deferred(self, mock_hook):
             subscription=TEST_SUBSCRIPTION,
             deferrable=True,
         )
-        with pytest.raises(TaskDeferred) as _:
+        with pytest.raises(TaskDeferred) as exc:
             task.execute(mock.MagicMock())
 
+        assert isinstance(exc.value.trigger, PubsubPullTrigger)
+        assert exc.value.trigger.return_immediately is True
+
     @mock.patch("airflow.providers.google.cloud.operators.pubsub.PubSubHook")
     def test_get_openlineage_facets(self, mock_hook):
         operator = PubSubPullOperator(

providers/google/tests/unit/google/cloud/sensors/test_pubsub.py

@@ -167,6 +167,25 @@ def test_pubsub_pull_sensor_async(self):
         with pytest.raises(TaskDeferred) as exc:
             task.execute(context={})
         assert isinstance(exc.value.trigger, PubsubPullTrigger), "Trigger is not a PubsubPullTrigger"
+        assert exc.value.trigger.return_immediately is True
+
+    def test_pubsub_pull_sensor_async_with_return_immediately_false(self):
+        """
+        Asserts that a task is deferred and a PubsubPullTrigger will be fired
+        with custom return_immediately value.
+        """
+        task = PubSubPullSensor(
+            task_id="test_task_id",
+            ack_messages=True,
+            project_id=TEST_PROJECT,
+            subscription=TEST_SUBSCRIPTION,
+            deferrable=True,
+            return_immediately=False,
+        )
+        with pytest.raises(TaskDeferred) as exc:
+            task.execute(context={})
+        assert isinstance(exc.value.trigger, PubsubPullTrigger), "Trigger is not a PubsubPullTrigger"
+        assert exc.value.trigger.return_immediately is False
 
     def test_pubsub_pull_sensor_async_execute_should_throw_exception(self):
         """Tests that an AirflowException is raised in case of error event"""

providers/google/tests/unit/google/cloud/triggers/test_pubsub.py

@@ -74,8 +74,34 @@ def test_async_pubsub_pull_trigger_serialization_should_execute_successfully(sel
             "poke_interval": TEST_POLL_INTERVAL,
             "gcp_conn_id": TEST_GCP_CONN_ID,
             "impersonation_chain": None,
+            "return_immediately": True,
         }
 
+    @pytest.mark.asyncio
+    @mock.patch("airflow.providers.google.cloud.hooks.pubsub.PubSubAsyncHook.pull")
+    async def test_async_pubsub_pull_trigger_passes_return_immediately_false(self, mock_pull):
+        """Test that return_immediately is passed to the hook."""
+        mock_pull.return_value = generate_messages(1)
+        trigger = PubsubPullTrigger(
+            project_id=PROJECT_ID,
+            subscription="subscription",
+            max_messages=MAX_MESSAGES,
+            ack_messages=False,
+            poke_interval=TEST_POLL_INTERVAL,
+            gcp_conn_id=TEST_GCP_CONN_ID,
+            impersonation_chain=None,
+            return_immediately=False,
+        )
+
+        await trigger.run().asend(None)
+
+        mock_pull.assert_called_once_with(
+            project_id=PROJECT_ID,
+            subscription="subscription",
+            max_messages=MAX_MESSAGES,
+            return_immediately=False,
+        )
+
     @pytest.mark.asyncio
     @mock.patch("airflow.providers.google.cloud.hooks.pubsub.PubSubAsyncHook.pull")
     async def test_async_pubsub_pull_trigger_return_event(self, mock_pull):