Pub/Sub streaming pull swallows Throwables that are not Exceptions

[elefeint]

This issue is a general case of the one in GoogleCloudPlatform/spring-cloud-gcp#1103. When user message processing results in a non-Exception Throwable (the two common usecases being assertion errors and system errors such as OutOfMemoryError), Pub/Sub client library effectively silently ignores the failure.
While the best practice is to avoid catching Throwable, in case of message processing it needs to be done -- the user application needs to have useful troubleshooting logs, and ideally the message ought to get nacked, as well. Of course, nacking with something like OutOfMemoryError won't reliably happen, but it's still worth an attempt.

Current behavior
Currently when user message processing code throws an Error, this MessageDispatcher code will pass the error on without catching, as the catch clause only applies to Exception. However, deliverMessageTask is a simple runnable that runs on a GAX thread. When the error bubbles up, the thread is returned to the thread pool.

Desired behavior
The catch block in MessageDispatcher should apply to any Throwable, so that ackReplySettableApiFuture.setException(e); can get set and processed as an error.

Environment details

1. Library: Pub/Sub
2. OS type and version: Linux
3. Java version: 17.0.2
4. version(s): 1.117.0

Steps to reproduce

A slightly modified code sample from docs demonstrates the issue:

public static void subscribeAsyncExample(String projectId, String subscriptionId) {
    ProjectSubscriptionName subscriptionName =
        ProjectSubscriptionName.of(projectId, subscriptionId);

    // Instantiate an asynchronous message receiver.
    MessageReceiver receiver =
        (PubsubMessage message, AckReplyConsumer consumer) -> {
          // Log incoming message, then throw a non-Exception Throwable
          System.out.println("Id: " + message.getMessageId());
          System.out.println("Data: " + message.getData().toStringUtf8());
          //consumer.ack();

          /****** This Error will disappear without a log; message won't be nacked and will time out *********/
          throw new Error("this Throwable will disappear");  
        };

    Subscriber subscriber = null;
    try {
      subscriber = Subscriber.newBuilder(subscriptionName, receiver).build();
      // Start the subscriber.
      subscriber.startAsync().awaitRunning();
      System.out.printf("Listening for messages on %s:\n", subscriptionName.toString());
      // Allow the subscriber to run for 30s unless an unrecoverable error occurs.
      subscriber.awaitTerminated(30, TimeUnit.MINUTES);
    } catch (TimeoutException timeoutException) {
      // Shut down the subscriber after 30s. Stop receiving messages.
      subscriber.stopAsync();
    }