diff --git a/.release-please-manifest.json b/.release-please-manifest.json index 8028c14b5..25b8f2e80 100644 --- a/.release-please-manifest.json +++ b/.release-please-manifest.json @@ -1,4 +1,4 @@ { - ".": "2.30.0" + ".": "2.31.0" } \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a9265758..c9daf8a3e 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py index 4c9e1c20e..486a728b4 100644 --- a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py +++ b/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 diff --git a/google/pubsub/gapic_version.py b/google/pubsub/gapic_version.py index 5ebb3bec4..8ab09c42e 100644 --- a/google/pubsub/gapic_version.py +++ b/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} diff --git a/google/pubsub_v1/gapic_version.py b/google/pubsub_v1/gapic_version.py index 5ebb3bec4..8ab09c42e 100644 --- a/google/pubsub_v1/gapic_version.py +++ b/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} diff --git a/google/pubsub_v1/types/pubsub.py b/google/pubsub_v1/types/pubsub.py index dcc25bb31..227da2208 100644 --- a/google/pubsub_v1/types/pubsub.py +++ b/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. diff --git a/samples/generated_samples/snippet_metadata_google.pubsub.v1.json b/samples/generated_samples/snippet_metadata_google.pubsub.v1.json index 28ac15bef..cb25ebf70 100644 --- a/samples/generated_samples/snippet_metadata_google.pubsub.v1.json +++ b/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": [ { diff --git a/samples/snippets/requirements.txt b/samples/snippets/requirements.txt index 83202d4f8..22d6f32d1 100644 --- a/samples/snippets/requirements.txt +++ b/samples/snippets/requirements.txt @@ -2,7 +2,7 @@ google-cloud-pubsub==2.29.0 avro==1.12.0 protobuf===4.24.4; python_version == '3.7' protobuf===5.29.4; python_version == '3.8' -protobuf==6.31.0; python_version >= '3.9' +protobuf==6.31.1; python_version >= '3.9' avro==1.12.0 opentelemetry-api===1.22.0; python_version == '3.7' opentelemetry-sdk===1.22.0; python_version == '3.7' diff --git a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py index f4ceedaf0..86d2461e7 100644 --- a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py +++ b/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)