fix: only pass on_fatal_exception if google-api-core client library version supports it

Author: abbrowne126

black

@@ -16,6 +16,7 @@
 
 import collections
 import functools
+import inspect
 import itertools
 import logging
 import threading
@@ -106,6 +107,13 @@
     code_pb2.UNAVAILABLE,
 }
 
+# `on_fatal_exception` 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.
@@ -884,9 +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=self._on_fatal_exception
-        )
+
+        # `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)
 

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] >= 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.*",
+    "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.*",
     "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

@@ -4,7 +4,7 @@
 # Pin the version to the lower bound.
 # e.g., if setup.py has "google-cloud-foo >= 1.14.0, < 2.0.0dev",
 # Then this file should have google-cloud-foo==1.14.0
-google-api-core==2.25.1
+google-api-core==1.34.0
 google-auth==2.14.1
 proto-plus==1.22.3
 protobuf==3.20.2

testing/constraints-3.7.txt

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 # This constraints file is required for unit tests.
 # List all library dependencies and extras in this file.
-google-api-core==2.25.1
+google-api-core==1.34.0
 proto-plus
 protobuf
 grpc-google-iam-v1

testing/constraints-3.8.txt

@@ -1333,9 +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, manager._on_fatal_exception
-    )
+    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
 
@@ -1444,18 +1448,19 @@ def test_closes_on_fatal_consumer_error():
         scheduler,
     ) = make_running_manager()
 
-    error = ValueError("some fatal exception")
-    manager._on_fatal_exception(error)
+    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)
+        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()
+        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
+        assert manager.is_active is False
 
 
 def test_close_inactive_consumer():