diff --git a/packages/google-cloud-pubsub/google/cloud/pubsub_v1/publisher/_batch/thread.py b/packages/google-cloud-pubsub/google/cloud/pubsub_v1/publisher/_batch/thread.py
index c874fc49941a..5a17d4aaad1d 100644
--- a/packages/google-cloud-pubsub/google/cloud/pubsub_v1/publisher/_batch/thread.py
+++ b/packages/google-cloud-pubsub/google/cloud/pubsub_v1/publisher/_batch/thread.py
@@ -21,9 +21,7 @@
 from datetime import datetime
 from typing import Any, Callable, List, Optional, Sequence
 
-import google.api_core.exceptions
 from google.api_core import gapic_v1
-from google.auth import exceptions as auth_exceptions
 from opentelemetry import trace
 
 from google.cloud.pubsub_v1.open_telemetry.publish_message_wrapper import (
@@ -341,12 +339,9 @@ def _commit(self) -> None:
                     )
                     span.set_attribute(key="messaging.message.id", value=message_id)
                     wrapper.end_create_span()
-        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.
+        except Exception as exc:
+            # Any publishing error must reach the futures. Otherwise, an
+            # unexpected exception leaves callers waiting indefinitely.
             self._status = base.BatchStatus.ERROR
 
             if self._client.open_telemetry_enabled:
diff --git a/packages/google-cloud-pubsub/tests/unit/pubsub_v1/publisher/batch/test_thread.py b/packages/google-cloud-pubsub/tests/unit/pubsub_v1/publisher/batch/test_thread.py
index 72b30a9ad139..3cc06ef38bd9 100644
--- a/packages/google-cloud-pubsub/tests/unit/pubsub_v1/publisher/batch/test_thread.py
+++ b/packages/google-cloud-pubsub/tests/unit/pubsub_v1/publisher/batch/test_thread.py
@@ -329,10 +329,14 @@ def test_blocking__commit_wrong_messageid_length():
     [
         (google.api_core.exceptions.InternalServerError("Internal server error"),),
         (auth_exceptions.TransportError("some transport error"),),
+        (auth_exceptions.RefreshError("credentials could not be refreshed"),),
+        (TypeError("invalid retry"),),
+        (ValueError("invalid timeout"),),
     ],
 )
 def test_block__commmit_api_error(error):
-    batch = create_batch()
+    batch_done_callback = mock.Mock()
+    batch = create_batch(batch_done_callback=batch_done_callback)
     futures = (
         batch.publish(
             wrapper=PublishMessageWrapper(
@@ -352,9 +356,29 @@ def test_block__commmit_api_error(error):
     with patch:
         batch._commit()
 
+    assert batch.status == BatchStatus.ERROR
+    batch_done_callback.assert_called_once_with(False)
     for future in futures:
         assert future.done()
-        assert future.exception() == error[0]
+        assert future.exception() is error[0]
+
+
+@pytest.mark.parametrize("option", ["retry", "timeout"])
+def test_commit_invalid_rpc_options(option):
+    batch = create_batch(**{f"commit_{option}": "invalid"})
+    future = batch.publish(
+        wrapper=PublishMessageWrapper(
+            message=gapic_types.PubsubMessage(data=b"message")
+        )
+    )
+
+    # Exercise the real GAPIC wrapper in the commit thread. Invalid RPC options
+    # fail before sending a request and must reach the caller through its future.
+    batch.commit()
+
+    with pytest.raises(TypeError, match="not callable"):
+        future.result(timeout=3)
+    assert batch.status == BatchStatus.ERROR
 
 
 def test_block__commmit_retry_error():