Merge branch 'main' into manually-upgrade-deps

Author: abbrowne126

.release-please-manifest.json

@@ -1,4 +1,4 @@
 {
-    ".": "2.30.0"
+    ".": "2.31.0"
 }
 

CHANGELOG.md

@@ -5,6 +5,18 @@
 [1]: https://pypi.org/project/google-cloud-pubsub/#history
 
 
+## [2.31.0](https://github.com/googleapis/python-pubsub/compare/v2.30.0...v2.31.0) (2025-06-26)
+
+
+### Features
+
+* Add MessageTransformationFailureReason to IngestionFailureEvent ([#1427](https://github.com/googleapis/python-pubsub/issues/1427)) ([8ab13e1](https://github.com/googleapis/python-pubsub/commit/8ab13e1b71c151f0146548e7224dd38c9d719a88))
+
+
+### Bug Fixes
+
+* Surface Fatal Stream Errors to Future; Adjust Retryable Error Codes ([#1422](https://github.com/googleapis/python-pubsub/issues/1422)) ([e081beb](https://github.com/googleapis/python-pubsub/commit/e081beb29056035304d365ec9c50fa7ffbac6886))
+
 ## [2.30.0](https://github.com/googleapis/python-pubsub/compare/v2.29.1...v2.30.0) (2025-06-07)
 
 

google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

@@ -16,6 +16,7 @@
 
 import collections
 import functools
+import inspect
 import itertools
 import logging
 import threading
@@ -62,14 +63,22 @@
 _REGULAR_SHUTDOWN_THREAD_NAME = "Thread-RegularStreamShutdown"
 _RPC_ERROR_THREAD_NAME = "Thread-OnRpcTerminated"
 _RETRYABLE_STREAM_ERRORS = (
+    exceptions.Aborted,
     exceptions.DeadlineExceeded,
-    exceptions.ServiceUnavailable,
+    exceptions.GatewayTimeout,
     exceptions.InternalServerError,
+    exceptions.ResourceExhausted,
+    exceptions.ServiceUnavailable,
     exceptions.Unknown,
-    exceptions.GatewayTimeout,
-    exceptions.Aborted,
 )
-_TERMINATING_STREAM_ERRORS = (exceptions.Cancelled,)
+_TERMINATING_STREAM_ERRORS = (
+    exceptions.Cancelled,
+    exceptions.InvalidArgument,
+    exceptions.NotFound,
+    exceptions.PermissionDenied,
+    exceptions.Unauthenticated,
+    exceptions.Unauthorized,
+)
 _MAX_LOAD = 1.0
 """The load threshold above which to pause the incoming message stream."""
 
@@ -98,6 +107,13 @@
     code_pb2.UNAVAILABLE,
 }
 
+# `on_fatal_exception` was added in `google-api-core v2.25.1``, which allows us to inform
+# callers on unrecoverable errors. We can only pass this arg if it's available in the
+# `BackgroundConsumer` spec.
+_SHOULD_USE_ON_FATAL_ERROR_CALLBACK = "on_fatal_exception" in inspect.getfullargspec(
+    bidi.BackgroundConsumer
+)
+
 
 def _wrap_as_exception(maybe_exception: Any) -> BaseException:
     """Wrap an object as a Python exception, if needed.
@@ -876,7 +892,18 @@ def open(
         assert self._scheduler is not None
         scheduler_queue = self._scheduler.queue
         self._dispatcher = dispatcher.Dispatcher(self, scheduler_queue)
-        self._consumer = bidi.BackgroundConsumer(self._rpc, self._on_response)
+
+        # `on_fatal_exception` is only available in more recent library versions.
+        # For backwards compatibility reasons, we only pass it when `google-api-core` supports it.
+        if _SHOULD_USE_ON_FATAL_ERROR_CALLBACK:
+            self._consumer = bidi.BackgroundConsumer(
+                self._rpc,
+                self._on_response,
+                on_fatal_exception=self._on_fatal_exception,
+            )
+        else:
+            self._consumer = bidi.BackgroundConsumer(self._rpc, self._on_response)
+
         self._leaser = leaser.Leaser(self)
         self._heartbeater = heartbeater.Heartbeater(self)
 
@@ -1247,6 +1274,17 @@ def _on_response(self, response: gapic_types.StreamingPullResponse) -> None:
 
         self.maybe_pause_consumer()
 
+    def _on_fatal_exception(self, exception: BaseException) -> None:
+        """
+        Called whenever `self.consumer` receives a non-retryable exception.
+        We close the manager on such non-retryable cases.
+        """
+        _LOGGER.exception(
+            "Streaming pull terminating after receiving non-recoverable error: %s",
+            exception,
+        )
+        self.close(exception)
+
     def _should_recover(self, exception: BaseException) -> bool:
         """Determine if an error on the RPC stream should be recovered.
 
@@ -1283,8 +1321,10 @@ def _should_terminate(self, exception: BaseException) -> bool:
             in a list of terminating exceptions.
         """
         exception = _wrap_as_exception(exception)
-        if isinstance(exception, _TERMINATING_STREAM_ERRORS):
-            _LOGGER.debug("Observed terminating stream error %s", exception)
+        is_api_error = isinstance(exception, exceptions.GoogleAPICallError)
+        # Terminate any non-API errors, or non-retryable errors (permission denied, unauthorized, etc.)
+        if not is_api_error or isinstance(exception, _TERMINATING_STREAM_ERRORS):
+            _LOGGER.error("Observed terminating stream error %s", exception)
             return True
         _LOGGER.debug("Observed non-terminating stream error %s", exception)
         return False

google/pubsub/gapic_version.py

@@ -13,4 +13,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-__version__ = "2.30.0"  # {x-release-please-version}
+__version__ = "2.31.0"  # {x-release-please-version}

google/pubsub_v1/gapic_version.py

@@ -13,4 +13,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-__version__ = "2.30.0"  # {x-release-please-version}
+__version__ = "2.31.0"  # {x-release-please-version}

google/pubsub_v1/types/pubsub.py

@@ -861,6 +861,12 @@ class SchemaViolationReason(proto.Message):
 
         """
 
+    class MessageTransformationFailureReason(proto.Message):
+        r"""Set when a Pub/Sub message fails to get published due to a
+        message transformation error.
+
+        """
+
     class CloudStorageFailure(proto.Message):
         r"""Failure when ingesting from a Cloud Storage source.
 
@@ -897,6 +903,11 @@ class CloudStorageFailure(proto.Message):
                 Optional. The Pub/Sub message failed schema
                 validation.
 
+                This field is a member of `oneof`_ ``reason``.
+            message_transformation_failure_reason (google.pubsub_v1.types.IngestionFailureEvent.MessageTransformationFailureReason):
+                Optional. Failure encountered when applying a
+                message transformation to the Pub/Sub message.
+
                 This field is a member of `oneof`_ ``reason``.
         """
 
@@ -932,6 +943,12 @@ class CloudStorageFailure(proto.Message):
                 message="IngestionFailureEvent.SchemaViolationReason",
             )
         )
+        message_transformation_failure_reason: "IngestionFailureEvent.MessageTransformationFailureReason" = proto.Field(
+            proto.MESSAGE,
+            number=8,
+            oneof="reason",
+            message="IngestionFailureEvent.MessageTransformationFailureReason",
+        )
 
     class AwsMskFailureReason(proto.Message):
         r"""Failure when ingesting from an Amazon MSK source.
@@ -965,6 +982,11 @@ class AwsMskFailureReason(proto.Message):
                 Optional. The Pub/Sub message failed schema
                 validation.
 
+                This field is a member of `oneof`_ ``reason``.
+            message_transformation_failure_reason (google.pubsub_v1.types.IngestionFailureEvent.MessageTransformationFailureReason):
+                Optional. Failure encountered when applying a
+                message transformation to the Pub/Sub message.
+
                 This field is a member of `oneof`_ ``reason``.
         """
 
@@ -998,6 +1020,12 @@ class AwsMskFailureReason(proto.Message):
                 message="IngestionFailureEvent.SchemaViolationReason",
             )
         )
+        message_transformation_failure_reason: "IngestionFailureEvent.MessageTransformationFailureReason" = proto.Field(
+            proto.MESSAGE,
+            number=7,
+            oneof="reason",
+            message="IngestionFailureEvent.MessageTransformationFailureReason",
+        )
 
     class AzureEventHubsFailureReason(proto.Message):
         r"""Failure when ingesting from an Azure Event Hubs source.
@@ -1031,6 +1059,11 @@ class AzureEventHubsFailureReason(proto.Message):
                 Optional. The Pub/Sub message failed schema
                 validation.
 
+                This field is a member of `oneof`_ ``reason``.
+            message_transformation_failure_reason (google.pubsub_v1.types.IngestionFailureEvent.MessageTransformationFailureReason):
+                Optional. Failure encountered when applying a
+                message transformation to the Pub/Sub message.
+
                 This field is a member of `oneof`_ ``reason``.
         """
 
@@ -1064,6 +1097,12 @@ class AzureEventHubsFailureReason(proto.Message):
                 message="IngestionFailureEvent.SchemaViolationReason",
             )
         )
+        message_transformation_failure_reason: "IngestionFailureEvent.MessageTransformationFailureReason" = proto.Field(
+            proto.MESSAGE,
+            number=7,
+            oneof="reason",
+            message="IngestionFailureEvent.MessageTransformationFailureReason",
+        )
 
     class ConfluentCloudFailureReason(proto.Message):
         r"""Failure when ingesting from a Confluent Cloud source.
@@ -1097,6 +1136,11 @@ class ConfluentCloudFailureReason(proto.Message):
                 Optional. The Pub/Sub message failed schema
                 validation.
 
+                This field is a member of `oneof`_ ``reason``.
+            message_transformation_failure_reason (google.pubsub_v1.types.IngestionFailureEvent.MessageTransformationFailureReason):
+                Optional. Failure encountered when applying a
+                message transformation to the Pub/Sub message.
+
                 This field is a member of `oneof`_ ``reason``.
         """
 
@@ -1130,10 +1174,21 @@ class ConfluentCloudFailureReason(proto.Message):
                 message="IngestionFailureEvent.SchemaViolationReason",
             )
         )
+        message_transformation_failure_reason: "IngestionFailureEvent.MessageTransformationFailureReason" = proto.Field(
+            proto.MESSAGE,
+            number=7,
+            oneof="reason",
+            message="IngestionFailureEvent.MessageTransformationFailureReason",
+        )
 
     class AwsKinesisFailureReason(proto.Message):
         r"""Failure when ingesting from an AWS Kinesis source.
 
+        This message has `oneof`_ fields (mutually exclusive fields).
+        For each oneof, at most one member field can be set at the same time.
+        Setting any member of the oneof automatically clears all other
+        members.
+
         .. _oneof: https://proto-plus-python.readthedocs.io/en/stable/fields.html#oneofs-mutually-exclusive-fields
 
         Attributes:
@@ -1150,6 +1205,11 @@ class AwsKinesisFailureReason(proto.Message):
                 Optional. The Pub/Sub message failed schema
                 validation.
 
+                This field is a member of `oneof`_ ``reason``.
+            message_transformation_failure_reason (google.pubsub_v1.types.IngestionFailureEvent.MessageTransformationFailureReason):
+                Optional. Failure encountered when applying a
+                message transformation to the Pub/Sub message.
+
                 This field is a member of `oneof`_ ``reason``.
         """
 
@@ -1173,6 +1233,12 @@ class AwsKinesisFailureReason(proto.Message):
                 message="IngestionFailureEvent.SchemaViolationReason",
             )
         )
+        message_transformation_failure_reason: "IngestionFailureEvent.MessageTransformationFailureReason" = proto.Field(
+            proto.MESSAGE,
+            number=5,
+            oneof="reason",
+            message="IngestionFailureEvent.MessageTransformationFailureReason",
+        )
 
     topic: str = proto.Field(
         proto.STRING,
@@ -3091,10 +3157,12 @@ class StreamingPullResponse(proto.Message):
             will not be empty.
         acknowledge_confirmation (google.pubsub_v1.types.StreamingPullResponse.AcknowledgeConfirmation):
             Optional. This field will only be set if
-            ``enable_exactly_once_delivery`` is set to ``true``.
+            ``enable_exactly_once_delivery`` is set to ``true`` and is
+            not guaranteed to be populated.
         modify_ack_deadline_confirmation (google.pubsub_v1.types.StreamingPullResponse.ModifyAckDeadlineConfirmation):
             Optional. This field will only be set if
-            ``enable_exactly_once_delivery`` is set to ``true``.
+            ``enable_exactly_once_delivery`` is set to ``true`` and is
+            not guaranteed to be populated.
         subscription_properties (google.pubsub_v1.types.StreamingPullResponse.SubscriptionProperties):
             Optional. Properties associated with this
             subscription.

samples/generated_samples/snippet_metadata_google.pubsub.v1.json

@@ -8,7 +8,7 @@
     ],
     "language": "PYTHON",
     "name": "google-cloud-pubsub",
-    "version": "2.30.0"
+    "version": "2.31.0"
   },
   "snippets": [
     {

tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py

@@ -1333,7 +1333,13 @@ def test_open(heartbeater, dispatcher, leaser, background_consumer, resumable_bi
     leaser.return_value.start.assert_called_once()
     assert manager.leaser == leaser.return_value
 
-    background_consumer.assert_called_once_with(manager._rpc, manager._on_response)
+    if streaming_pull_manager._SHOULD_USE_ON_FATAL_ERROR_CALLBACK:
+        background_consumer.assert_called_once_with(
+            manager._rpc, manager._on_response, manager._on_fatal_exception
+        )
+    else:
+        background_consumer.assert_called_once_with(manager._rpc, manager._on_response)
+
     background_consumer.return_value.start.assert_called_once()
     assert manager._consumer == background_consumer.return_value
 
@@ -1432,6 +1438,31 @@ def test_close():
     assert manager.is_active is False
 
 
+def test_closes_on_fatal_consumer_error():
+    (
+        manager,
+        consumer,
+        dispatcher,
+        leaser,
+        heartbeater,
+        scheduler,
+    ) = make_running_manager()
+
+    if streaming_pull_manager._SHOULD_USE_ON_FATAL_ERROR_CALLBACK:
+        error = ValueError("some fatal exception")
+        manager._on_fatal_exception(error)
+
+        await_manager_shutdown(manager, timeout=3)
+
+        consumer.stop.assert_called_once()
+        leaser.stop.assert_called_once()
+        dispatcher.stop.assert_called_once()
+        heartbeater.stop.assert_called_once()
+        scheduler.shutdown.assert_called_once()
+
+        assert manager.is_active is False
+
+
 def test_close_inactive_consumer():
     (
         manager,
@@ -2270,18 +2301,24 @@ def test__should_recover_false():
 def test__should_terminate_true():
     manager = make_manager()
 
-    details = "Cancelled. Go away, before I taunt you a second time."
-    exc = exceptions.Cancelled(details)
-
-    assert manager._should_terminate(exc) is True
+    for exc in [
+        exceptions.Cancelled(""),
+        exceptions.PermissionDenied(""),
+        TypeError(),
+        ValueError(),
+    ]:
+        assert manager._should_terminate(exc)
 
 
 def test__should_terminate_false():
     manager = make_manager()
 
-    exc = TypeError("wahhhhhh")
-
-    assert manager._should_terminate(exc) is False
+    for exc in [
+        exceptions.ResourceExhausted(""),
+        exceptions.ServiceUnavailable(""),
+        exceptions.DeadlineExceeded(""),
+    ]:
+        assert not manager._should_terminate(exc)
 
 
 @mock.patch("threading.Thread", autospec=True)