fix: allow BackgroundConsumer caller to pass on_fatal_exception to be informed of fatal processing errors

Author: abbrowne126

google/api_core/bidi.py

@@ -624,17 +624,15 @@ 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`.
+        on_fatal_exception (Callable[[Exception], None]): The callback to
+            be called on fatal errors during consumption. Default None.
     """
 
-    def __init__(self, bidi_rpc, on_response, reraise_exceptions=False):
+    def __init__(self, bidi_rpc, on_response, on_fatal_exception=None):
         self._bidi_rpc = bidi_rpc
         self._on_response = on_response
         self._paused = False
-        self._reraise_exceptions = reraise_exceptions
+        self._on_fatal_exception = on_fatal_exception
         self._wake = threading.Condition()
         self._thread = None
         self._operational_lock = threading.Lock()
@@ -681,17 +679,17 @@ def _thread_main(self, ready):
                 exc,
                 exc_info=True,
             )
-            if self._reraise_exceptions:
-                raise
+            if self._on_fatal_exception is not None:
+                self._on_fatal_exception(exc)
 
         except Exception as exc:
             _LOGGER.exception(
                 "%s caught unexpected exception %s and will exit.",
                 _BIDIRECTIONAL_CONSUMER_NAME,
                 exc,
             )
-            if self._reraise_exceptions:
-                raise
+            if self._on_fatal_exception is not None:
+                self._on_fatal_exception(exc)
 
         _LOGGER.info("%s exiting", _BIDIRECTIONAL_CONSUMER_NAME)
 
@@ -734,6 +732,7 @@ def stop(self):
 
             self._thread = None
             self._on_response = None
+            self._on_fatal_exception = None
 
     @property
     def is_active(self):

tests/unit/test_bidi.py

@@ -919,11 +919,11 @@ def test_stop_error_logs(self, caplog):
         assert not error_logs, f"Found unexpected ERROR logs: {error_logs}"
         bidi_rpc.is_active = False
 
-    def test_fatal_exceptions_will_shutdown_consumer(self, caplog):
+    def test_fatal_exceptions_can_inform_consumer(self, caplog):
         """
         https://github.com/googleapis/python-api-core/issues/820
         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`.
+        on the RPC should be bubbled back to the caller through `on_fatal_exception` if passed.
         """
         caplog.set_level(logging.DEBUG)
 
@@ -935,14 +935,16 @@ def test_fatal_exceptions_will_shutdown_consumer(self, caplog):
             bidi_rpc.is_active = True
             on_response = mock.Mock(spec=["__call__"])
 
+            on_fatal_exception = mock.Mock(spec=["__call__"])
+
             bidi_rpc.open.side_effect = fatal_exception
 
             consumer = bidi.BackgroundConsumer(
-                bidi_rpc, on_response, reraise_exceptions=True
+                bidi_rpc, on_response, on_fatal_exception
             )
 
-            with pytest.raises(type(fatal_exception)):
-                consumer.start()
+            consumer.start()
+            # let the background thread run for a while before exiting
+            time.sleep(0.1)
 
-                # let the background thread run for a while before exiting
-                time.sleep(0.1)
+            on_fatal_exception.assert_called_once_with(fatal_exception)