fix: surface stream closures to calling future

Author: abbrowne126

black

@@ -884,7 +884,9 @@ 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)
+        self._consumer = bidi.BackgroundConsumer(
+            self._rpc, self._on_response, on_fatal_exception=self._on_fatal_exception
+        )
         self._leaser = leaser.Leaser(self)
         self._heartbeater = heartbeater.Heartbeater(self)
 
@@ -1255,6 +1257,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.
 

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

@@ -39,7 +39,7 @@
     "grpcio >= 1.51.3, < 2.0.0",  # https://github.com/googleapis/python-pubsub/issues/609
     # google-api-core >= 1.34.0 is allowed in order to support google-api-core 1.x
     "google-auth >= 2.14.1, <3.0.0",
-    "google-api-core[grpc] >= 1.34.0, <3.0.0,!=2.0.*,!=2.1.*,!=2.2.*,!=2.3.*,!=2.4.*,!=2.5.*,!=2.6.*,!=2.7.*,!=2.8.*,!=2.9.*,!=2.10.*",
+    "google-api-core[grpc] >= 2.25.1, <3.0.0,!=2.0.*,!=2.1.*,!=2.2.*,!=2.3.*,!=2.4.*,!=2.5.*,!=2.6.*,!=2.7.*,!=2.8.*,!=2.9.*,!=2.10.*",
     "proto-plus >= 1.22.0, <2.0.0",
     "proto-plus >= 1.22.2, <2.0.0; python_version>='3.11'",
     "proto-plus >= 1.25.0, < 2.0.0; python_version >= '3.13'",

setup.py

@@ -1333,7 +1333,9 @@ 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)
+    background_consumer.assert_called_once_with(
+        manager._rpc, manager._on_response, manager._on_fatal_exception
+    )
     background_consumer.return_value.start.assert_called_once()
     assert manager._consumer == background_consumer.return_value
 
@@ -1432,6 +1434,30 @@ 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()
+
+    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,