Repository: googleapis/python-pubsub

Issue:
Publisher thread terminates, forever breaking publication when GCE metadata service blips
Thanks for stopping by to let us know something could be better!

**PLEASE READ**: If you have a support contract with Google, please create an issue in the [support console](https://cloud.google.com/support/) instead of filing on GitHub. This will ensure a timely response.

Please run down the following list and make sure you've tried the usual "quick fixes":

  - Search the issues already opened: https://github.com/googleapis/python-pubsub/issues
  - Search StackOverflow: https://stackoverflow.com/questions/tagged/google-cloud-platform+python

If you are still having issues, please be sure to include as much information as possible:

#### Environment details

  - OS type and version: Ubuntu 22.04
  - Python version: 3.10.9
  - pip version: `pip --version`
  - `google-cloud-pubsub` version: 2.21.1

#### Steps to reproduce

Run `google-cloud-pubsub` and suffer a metadata outage like https://status.cloud.google.com/incidents/u6rQ2nNVbhAFqGCcTm58.

Note that this can trigger even in an un-sustained GCE metadata outage as once this exception triggers even once, the commit thread is dead forever. In our case, there was a short outage on the metadata server, but the retries all happened so quickly that the exception was raised before the service recovered

```
2024-04-26T07:30:45.783 Compute Engine Metadata server unavailable on attempt 1 of 5. Reason: HTTPConnectionPool(host='metadata.google.internal', port=80): Max retries exceeded with url: /computeMetadata/v1/universe/universe_domain (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7c1ca813c1c0>: Failed to establish a new connection: [Errno 111] Connection refused'))
2024-04-26T07:30:45.788 Compute Engine Metadata server unavailable on attempt 2 of 5. Reason: HTTPConnectionPool(host='metadata.google.internal', port=80): Max retries exceeded with url: /computeMetadata/v1/universe/universe_domain (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7c1ca8290730>: Failed to establish a new connection: [Errno 111] Connection refused'))
2024-04-26T07:30:45.794 Compute Engine Metadata server unavailable on attempt 3 of 5. Reason: HTTPConnectionPool(host='metadata.google.internal', port=80): Max retries exceeded with url: /computeMetadata/v1/universe/universe_domain (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7c1ca82918a0>: Failed to establish a new connection: [Errno 111] Connection refused'))
2024-04-26T07:30:45.801 [...]
2024-04-26T07:30:45.806 [...]
```

#### Code example

```python
# example
```

#### Stack trace
```
Traceback (most recent call last):
  File "/app/device/trimark/proxy/proxy.runfiles/python3_10_x86_64-unknown-linux-gnu/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
    self.run()
  File "/app/device/trimark/proxy/proxy.runfiles/python3_10_x86_64-unknown-linux-gnu/lib/python3.10/threading.py", line 953, in run
    self._target(*self._args, **self._kwargs)
  File "/app/device/trimark/proxy/proxy.runfiles/common_deps_google_cloud_pubsub/site-packages/google/cloud/pubsub_v1/publisher/_batch/thread.py", line 274, in _commit
    response = self._client._gapic_publish(
  File "/app/device/trimark/proxy/proxy.runfiles/common_deps_google_cloud_pubsub/site-packages/google/cloud/pubsub_v1/publisher/client.py", line 267, in _gapic_publish
    return super().publish(*args, **kwargs)
  File "/app/device/trimark/proxy/proxy.runfiles/common_deps_google_cloud_pubsub/site-packages/google/pubsub_v1/services/publisher/client.py", line 1058, in publish
    self._validate_universe_domain()
  File "/app/device/trimark/proxy/proxy.runfiles/common_deps_google_cloud_pubsub/site-packages/google/pubsub_v1/services/publisher/client.py", line 554, in _validate_universe_domain
    or PublisherClient._compare_universes(
  File "/app/device/trimark/proxy/proxy.runfiles/common_deps_google_cloud_pubsub/site-packages/google/pubsub_v1/services/publisher/client.py", line 531, in _compare_universes
    credentials_universe = getattr(credentials, \"universe_domain\", default_universe)
  File "/app/device/trimark/proxy/proxy.runfiles/common_deps_google_auth/site-packages/google/auth/compute_engine/credentials.py", line 154, in universe_domain
    self._universe_domain = _metadata.get_universe_domain(
  File "/app/device/trimark/proxy/proxy.runfiles/common_deps_google_auth/site-packages/google/auth/compute_engine/_metadata.py", line 284, in get_universe_domain
    universe_domain = get(
  File "/app/device/trimark/proxy/proxy.runfiles/common_deps_google_auth/site-packages/google/auth/compute_engine/_metadata.py", line 217, in get
    raise exceptions.TransportError(
google.auth.exceptions.TransportError: Failed to retrieve http://metadata.google.internal/computeMetadata/v1/universe/universe_domain from the Google Compute Engine metadata service. Compute Engine Metadata server unavailable
```

#### Speculative analysis

It looks like the issue is that the `google-auth` library is raising a `TransportError` which is not caught by the batch commit thread in this library. Potential fixes include catching that in `Batch._commit` (e.g. [here](https://github.com/googleapis/python-pubsub/blob/main/google/cloud/pubsub_v1/publisher/_batch/thread.py#L272-L292)), or catching it further down in `google-cloud-pubsub` and wrapping it in a `GoogleAPIError`.

Discussion:
@mukund-ananthu: Hi [pgcamus](https://github.com/pgcamus), thank you for bringing this to our attention. Wanted to list some of my observations:

> the commit thread is dead forever. 

This is correct. But, one thing I observe here is that the publishing of subsequent messages aren't affected, because of this sequence of events:
1. TransportError thrown by gapic_publish: https://github.com/googleapis/python-pubsub/blob/cdaf6e9bcd0b918c2ab50fd4a8f987f60cbd610f/google/cloud/pubsub_v1/publisher/_batch/thread.py#L322
2. Not caught by the Except clause that only catches GoogleAPIError: https://github.com/googleapis/python-pubsub/blob/cdaf6e9bcd0b918c2ab50fd4a8f987f60cbd610f/google/cloud/pubsub_v1/publisher/_batch/thread.py#L345
3. Thread in which `_commit` runs dies 
4. Status of the batch instance still remains IN_PROGRESS: https://github.com/googleapis/python-pubsub/blob/cdaf6e9bcd0b918c2ab50fd4a8f987f60cbd610f/google/cloud/pubsub_v1/publisher/_batch/thread.py#L288 , i.e it is not set to `BatchStatus.ERROR` like it is in case the `GoogleAPIError` is thrown: https://github.com/googleapis/python-pubsub/blob/cdaf6e9bcd0b918c2ab50fd4a8f987f60cbd610f/google/cloud/pubsub_v1/publisher/_batch/thread.py#L348

(this causes a separate issue - more on this later)

5. For the next message that is attempted to be published in this batch, it batch.publish() would return a `None` future because of this condition check: https://github.com/googleapis/python-pubsub/blob/cdaf6e9bcd0b918c2ab50fd4a8f987f60cbd610f/google/cloud/pubsub_v1/publisher/_batch/thread.py#L449-L450

6. This would result in the creation of a new batch instance altogether in the sequencer that is trying to publish to the batch:
https://github.com/googleapis/python-pubsub/blob/cdaf6e9bcd0b918c2ab50fd4a8f987f60cbd610f/google/cloud/pubsub_v1/publisher/_sequencer/unordered_sequencer.py#L156-L162

7. This new batch instance would create a new commit thread altogether, whenever the criteria to flush the new batch of messages is met: https://github.com/googleapis/python-pubsub/blob/cdaf6e9bcd0b918c2ab50fd4a8f987f60cbd610f/google/cloud/pubsub_v1/publisher/_batch/thread.py#L230


Having mentioned the above, I do observe that there are a few related issues that need to be addressed:

1. Since the TransportError is never caught, and the code path that sets exceptions on the future objects corresponding to messages in the batch that failed never runs. Consequently, futures that are associated with the messages in the batch whose commit thread crashed would never get notified of the exception. They'd indefinitely hang unless they have a timeout set as part of `future.result(timeout=x)` (even then, they'd only raise a TimeoutError but not have the exception object set) : https://github.com/googleapis/python-pubsub/blob/cdaf6e9bcd0b918c2ab50fd4a8f987f60cbd610f/google/cloud/pubsub_v1/publisher/_batch/thread.py#L368-L369

>Potential fixes include catching that in Batch._commit

This does solve the problem of:
1. The commit thread crashing instead of catching the Exception and returning gracefully
2. The futures of the messages in the batch that failed having `future.set_exception` done on them

The side effect of the above change, however, would be that, given the state of the batch would now be set to `BatchStatus.ERROR`, it would cause an AssertionError to be raised for the publish of the subsequent message on the batch: https://github.com/googleapis/python-pubsub/blob/cdaf6e9bcd0b918c2ab50fd4a8f987f60cbd610f/google/cloud/pubsub_v1/publisher/_batch/thread.py#L445-L447

which would then be bubbled up until a BaseException is thrown and the client library crashes in the publisher client: https://github.com/googleapis/python-pubsub/blob/cdaf6e9bcd0b918c2ab50fd4a8f987f60cbd610f/google/cloud/pubsub_v1/publisher/client.py#L481-L495

I'm exploring a few options to solve the above issues, without causing unintended side effects and also factoring in the behavior of the Pub/Sub client libraries of other languages to keep the behavior consistent across the libraries. I'll keep this thread posted. Thanks!


Solution patch:
diff --git a/google/cloud/pubsub_v1/publisher/_batch/thread.py b/google/cloud/pubsub_v1/publisher/_batch/thread.py
index c4bf67c35..2afbe3761 100644
--- a/google/cloud/pubsub_v1/publisher/_batch/thread.py
+++ b/google/cloud/pubsub_v1/publisher/_batch/thread.py
@@ -24,6 +24,7 @@
 from opentelemetry import trace
 import google.api_core.exceptions
 from google.api_core import gapic_v1
+from google.auth import exceptions as auth_exceptions
 
 from google.cloud.pubsub_v1.publisher import exceptions
 from google.cloud.pubsub_v1.publisher import futures
@@ -342,7 +343,10 @@ def _commit(self) -> None:
                     )
                     span.set_attribute(key="messaging.message.id", value=message_id)
                     wrapper.end_create_span()
-        except google.api_core.exceptions.GoogleAPIError as exc:
+        except (
+            google.api_core.exceptions.GoogleAPIError,
+            auth_exceptions.TransportError,
+        ) as exc:
             # We failed to publish, even after retries, so set the exception on
             # all futures and exit.
             self._status = base.BatchStatus.ERROR


Test patch:
diff --git a/tests/unit/pubsub_v1/publisher/batch/test_thread.py b/tests/unit/pubsub_v1/publisher/batch/test_thread.py
index 32eaa3d98..ad8fa376b 100644
--- a/tests/unit/pubsub_v1/publisher/batch/test_thread.py
+++ b/tests/unit/pubsub_v1/publisher/batch/test_thread.py
@@ -31,6 +31,7 @@
 import google.api_core.exceptions
 from google.api_core import gapic_v1
 from google.auth import credentials
+from google.auth import exceptions as auth_exceptions
 from google.cloud.pubsub_v1 import publisher
 from google.cloud.pubsub_v1 import types
 from google.cloud.pubsub_v1.publisher import exceptions
@@ -329,7 +330,14 @@ def test_blocking__commit_wrong_messageid_length():
         assert isinstance(future.exception(), exceptions.PublishError)
 
 
-def test_block__commmit_api_error():
+@pytest.mark.parametrize(
+    "error",
+    [
+        (google.api_core.exceptions.InternalServerError("Internal server error"),),
+        (auth_exceptions.TransportError("some transport error"),),
+    ],
+)
+def test_block__commmit_api_error(error):
     batch = create_batch()
     futures = (
         batch.publish(
@@ -345,7 +353,6 @@ def test_block__commmit_api_error():
     )
 
     # Make the API throw an error when publishing.
-    error = google.api_core.exceptions.InternalServerError("uh oh")
     patch = mock.patch.object(type(batch.client), "_gapic_publish", side_effect=error)
 
     with patch:
@@ -353,7 +360,7 @@ def test_block__commmit_api_error():
 
     for future in futures:
         assert future.done()
-        assert future.exception() == error
+        assert future.exception() == error[0]
 
 
 def test_block__commmit_retry_error():