Propagate 'RetryError' in 'PublisherClient.publish'. (#7071)

pubsub/google/cloud/pubsub_v1/publisher/_batch/thread.py

@@ -199,7 +199,7 @@ def _commit(self):
 
             try:
                 response = self._client.api.publish(self._topic, self._messages)
-            except google.api_core.exceptions.GoogleAPICallError as exc:
+            except google.api_core.exceptions.GoogleAPIError as exc:
                 # We failed to publish, set the exception on all futures and
                 # exit.
                 self._status = base.BatchStatus.ERROR

pubsub/tests/unit/pubsub_v1/publisher/batch/test_thread.py

@@ -221,6 +221,25 @@ def test_block__commmit_api_error():
         assert future.exception() == error
 
 
+def test_block__commmit_retry_error():
+    batch = create_batch()
+    futures = (
+        batch.publish({"data": b"blah blah blah"}),
+        batch.publish({"data": b"blah blah blah blah"}),
+    )
+
+    # Make the API throw an error when publishing.
+    error = google.api_core.exceptions.RetryError("uh oh", None)
+    patch = mock.patch.object(type(batch.client.api), "publish", side_effect=error)
+
+    with patch:
+        batch._commit()
+
+    for future in futures:
+        assert future.done()
+        assert future.exception() == error
+
+
 def test_monitor():
     batch = create_batch(max_latency=5.0)
     with mock.patch.object(time, "sleep") as sleep: