diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py
index 4c9e1c20e..4bb729a46 100644
--- a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py
+++ b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py
@@ -62,14 +62,22 @@
 _REGULAR_SHUTDOWN_THREAD_NAME = "Thread-RegularStreamShutdown"
 _RPC_ERROR_THREAD_NAME = "Thread-OnRpcTerminated"
 _RETRYABLE_STREAM_ERRORS = (
+    exceptions.Aborted,
     exceptions.DeadlineExceeded,
-    exceptions.ServiceUnavailable,
+    exceptions.GatewayTimeout,
     exceptions.InternalServerError,
+    exceptions.ResourceExhausted,
+    exceptions.ServiceUnavailable,
     exceptions.Unknown,
-    exceptions.GatewayTimeout,
-    exceptions.Aborted,
 )
-_TERMINATING_STREAM_ERRORS = (exceptions.Cancelled,)
+_TERMINATING_STREAM_ERRORS = (
+    exceptions.Cancelled,
+    exceptions.InvalidArgument,
+    exceptions.NotFound,
+    exceptions.PermissionDenied,
+    exceptions.Unauthenticated,
+    exceptions.Unauthorized,
+)
 _MAX_LOAD = 1.0
 """The load threshold above which to pause the incoming message stream."""
 
@@ -1283,8 +1291,10 @@ def _should_terminate(self, exception: BaseException) -> bool:
             in a list of terminating exceptions.
         """
         exception = _wrap_as_exception(exception)
-        if isinstance(exception, _TERMINATING_STREAM_ERRORS):
-            _LOGGER.debug("Observed terminating stream error %s", exception)
+        is_api_error = isinstance(exception, exceptions.GoogleAPICallError)
+        # Terminate any non-API errors, or non-retryable errors (permission denied, unauthorized, etc.)
+        if not is_api_error or isinstance(exception, _TERMINATING_STREAM_ERRORS):
+            _LOGGER.error("Observed terminating stream error %s", exception)
             return True
         _LOGGER.debug("Observed non-terminating stream error %s", exception)
         return False
diff --git a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py
index f4ceedaf0..331d067db 100644
--- a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py
+++ b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py
@@ -2270,18 +2270,24 @@ def test__should_recover_false():
 def test__should_terminate_true():
     manager = make_manager()
 
-    details = "Cancelled. Go away, before I taunt you a second time."
-    exc = exceptions.Cancelled(details)
-
-    assert manager._should_terminate(exc) is True
+    for exc in [
+        exceptions.Cancelled(""),
+        exceptions.PermissionDenied(""),
+        TypeError(),
+        ValueError(),
+    ]:
+        assert manager._should_terminate(exc)
 
 
 def test__should_terminate_false():
     manager = make_manager()
 
-    exc = TypeError("wahhhhhh")
-
-    assert manager._should_terminate(exc) is False
+    for exc in [
+        exceptions.ResourceExhausted(""),
+        exceptions.ServiceUnavailable(""),
+        exceptions.DeadlineExceeded(""),
+    ]:
+        assert not manager._should_terminate(exc)
 
 
 @mock.patch("threading.Thread", autospec=True)