fix: allow exceptions to be surfaced to the caller of BackgroundConsumer

Author: abbrowne126

google/api_core/bidi.py

@@ -624,12 +624,17 @@ def on_response(response):
             ``open()``ed yet.
         on_response (Callable[[protobuf.Message], None]): The callback to
             be called for every response on the stream.
+        reraise_exceptions (bool): Whether to reraise exceptions during
+            the lifetime of the consumer, generally those that are not
+            handled by `BidiRpc`'s `should_recover` or `should_terminate`.
+            Default `False`.
     """
 
-    def __init__(self, bidi_rpc, on_response):
+    def __init__(self, bidi_rpc, on_response, reraise_exceptions=False):
         self._bidi_rpc = bidi_rpc
         self._on_response = on_response
         self._paused = False
+        self._reraise_exceptions = reraise_exceptions
         self._wake = threading.Condition()
         self._thread = None
         self._operational_lock = threading.Lock()
@@ -676,15 +681,17 @@ def _thread_main(self, ready):
                 exc,
                 exc_info=True,
             )
-            self.stop()
+            if self._reraise_exceptions:
+                raise
 
         except Exception as exc:
             _LOGGER.exception(
                 "%s caught unexpected exception %s and will exit.",
                 _BIDIRECTIONAL_CONSUMER_NAME,
                 exc,
             )
-            self.stop()
+            if self._reraise_exceptions:
+                raise
 
         _LOGGER.info("%s exiting", _BIDIRECTIONAL_CONSUMER_NAME)
 
@@ -696,8 +703,8 @@ def start(self):
                 name=_BIDIRECTIONAL_CONSUMER_NAME,
                 target=self._thread_main,
                 args=(ready,),
+                daemon=True,
             )
-            thread.daemon = True
             thread.start()
             # Other parts of the code rely on `thread.is_alive` which
             # isn't sufficient to know if a thread is active, just that it may

tests/unit/test_bidi.py

@@ -922,22 +922,27 @@ def test_stop_error_logs(self, caplog):
     def test_fatal_exceptions_will_shutdown_consumer(self, caplog):
         """
         https://github.com/googleapis/python-api-core/issues/820
-        Exceptions thrown in the BackgroundConsumer that
-        lead to the consumer halting should also stop the thread and rpc.
+        Exceptions thrown in the BackgroundConsumer not caught by `should_recover` / `should_terminate`
+        on the RPC should be bubbled back to the caller if `reraise_exceptions` is `True`.
         """
         caplog.set_level(logging.DEBUG)
-        bidi_rpc = mock.create_autospec(bidi.ResumableBidiRpc, instance=True)
-        bidi_rpc.is_active = True
-        on_response = mock.Mock(spec=["__call__"])
 
-        bidi_rpc.open.side_effect = ValueError()
+        for fatal_exception in (
+            ValueError("some non-api error"),
+            exceptions.PermissionDenied("some api error"),
+        ):
+            bidi_rpc = mock.create_autospec(bidi.ResumableBidiRpc, instance=True)
+            bidi_rpc.is_active = True
+            on_response = mock.Mock(spec=["__call__"])
 
-        consumer = bidi.BackgroundConsumer(bidi_rpc, on_response)
+            bidi_rpc.open.side_effect = fatal_exception
 
-        consumer.start()
+            consumer = bidi.BackgroundConsumer(
+                bidi_rpc, on_response, reraise_exceptions=True
+            )
 
-        # let the background thread run for a while before exiting
-        time.sleep(0.1)
+            with pytest.raises(type(fatal_exception)):
+                consumer.start()
 
-        # We want to make sure that close is called, which will surface the error to the caller.
-        bidi_rpc.close.assert_called_once()
+                # let the background thread run for a while before exiting
+                time.sleep(0.1)