Missing catch-all try-except in batch thread, and missing publish() argument validation

[lvt-mm]

Environment details

• OS type and version: N/A
• Python version: N/A
• pip version: N/A
• google-cloud-pubsub version: 2.21.1

Steps to reproduce

1. Create a google.cloud.pubsub_v1.PublisherClient instance and call the publish method with timeout="some_string" or retry="some_string" (causes error in batch thread) and then call the result() method on the returned future.
2. Expect pubsub to fail fast and hard with an appropriate error in the main thread, instead it proceeds without complaining, which causes an uncaught exception in the batch thread, which causes the main thread to hang.

Code example

Start a pubsub emulator locally, create the below topic and set the PUBSUB_EMULATOR_HOST variable to point to it. Then:

from google.cloud import pubsub_v1
from google.oauth2.service_account import Credentials

credentials = Credentials.from_service_account_file(
        '/path/to/some/dummy-service-account.json')
publisher = pubsub_v1.PublisherClient(credentials=credentials)
topic_path = publisher.topic_path("my-project", "my-topic")
data = "Hello, world!"
future = publisher.publish(topic_path, data.encode('utf-8'), retry="asdf")
print(f"Published message ID {future.result()}")

The script prints the below strack trace and hangs (indefinitely?), blocking in future.result().

While adding some sanity checks on the arguments to publish() would fix this particular instance of the problem, the fundamental problem here appears to be that there's no catch-all try-except in the batch thread ensuring that it will always terminate gracefully.

Stack trace

The uncaught error in the batch thread is caught and printed to stderr by code in the threading module:

Exception in thread Thread-CommitBatchPublisher:
Traceback (most recent call last):
  File "/path/to/pyenv/versions/3.9.16/lib/python3.9/threading.py", line 980, in _bootstrap_inner
    self.run()
  File "/path/to/pyenv/versions/3.9.16/lib/python3.9/threading.py", line 917, in run
    self._target(*self._args, **self._kwargs)
  File "/path/to/pyenv/versions/mm3.9.16/lib/python3.9/site-packages/google/cloud/pubsub_v1/publisher/_batch/thread.py", line 274, in _commit
    response = self._client._gapic_publish(
  File "/path/to/pyenv/versions/mm3.9.16/lib/python3.9/site-packages/google/cloud/pubsub_v1/publisher/client.py", line 267, in _gapic_publish
    return super().publish(*args, **kwargs)
  File "/path/to/pyenv/versions/mm3.9.16/lib/python3.9/site-packages/google/pubsub_v1/services/publisher/client.py", line 1061, in publish
    response = rpc(
  File "/path/to/pyenv/versions/mm3.9.16/lib/python3.9/site-packages/google/api_core/gapic_v1/method.py", line 116, in __call__
    wrapped_func = _apply_decorators(self._target, [retry, timeout])
  File "/path/to/pyenv/versions/mm3.9.16/lib/python3.9/site-packages/google/api_core/gapic_v1/method.py", line 58, in _apply_decorators
    func = decorator(func)
TypeError: 'str' object is not callable