PubSub: Authentication errors leave workers in a broken state

Environment details

1. Specify the API at the beginning of the title. For example, "BigQuery: ...").
   General, Core, and Other are also allowed as types
2. OS type and version: Debian Bullseye
3. Java version: 11
4. version(s): 1.3.1

Steps to reproduce

Unable to manually reproduce, it requires some cryptic error being returned by GRPC authentication.

Stack trace

java.lang.RuntimeException: java.util.concurrent.ExecutionException: com.google.api.gax.rpc.UnauthenticatedException: io.grpc.StatusRuntimeException: UNAUTHENTICATED: Request had invalid euc header.
         at com.google.pubsub.kafka.sink.CloudPubSubSinkTask.flush(CloudPubSubSinkTask.java:352)
         at org.apache.kafka.connect.sink.SinkTask.preCommit(SinkTask.java:139)
         at org.apache.kafka.connect.runtime.WorkerSinkTask.commitOffsets(WorkerSinkTask.java:429)
         at io.grpc.stub.ClientCalls$GrpcFuture.setException(ClientCalls.java:568)
         at org.apache.kafka.connect.runtime.WorkerSinkTask.commitOffsets(WorkerSinkTask.java:399)
         at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:538)
         at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
         at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
         at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
         at com.google.api.gax.grpc.ChannelPool$ReleasingClientCall$1.onClose(ChannelPool.java:569)
         at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:564)
         at io.grpc.internal.ClientCallImpl.access$100(ClientCallImpl.java:72)
         at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:729)
         at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:710)
         at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
         at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
         at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:230)
         at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:215)
         at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:225)
         at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:280)
         at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:237)
         at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
         at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
         at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
         at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
         at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
         at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
         at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
         at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
         at java.base/java.lang.Thread.run(Thread.java:829)
 Caused by: io.grpc.StatusRuntimeException: UNAUTHENTICATED: Request had invalid euc header.
         at io.grpc.Status.asRuntimeException(Status.java:532)
         ... 17 more```

#### Any additional information below

We are getting these errors sporadically from pubsub. We aren't entirely sure what the root cause is, but the problem is that when we get these errors the connector goes into a zombie state where it stops processing messages from the source connector but doesn't crash the process. The process just keeps running, reporting the task status as RUNNING in the JMX status API, but does not actually do any work. We have to manually restart the process when it gets stuck in this zombie state. 

We're trying to figure out how to detect this state and restart the process automatically, or better yet if you could just retry the request to pubsub when it gets this exception it should work as it appears to be really intermittent (a few times a week).

I've pored through the source code and configuration options and don't see any easy way to handle this ourselves, but I'm happy to inject some custom code if you can point me in the right direction.

Start with CloudPubSubSinkTask.java at flush(), especially around line 352 in the stack trace, and inspect how the authentication exception affects task progress. Determine whether the connector can recover from this intermittent UNAUTHENTICATED error or detect the reported zombie state; done should include a verifiable recovery behavior, though the issue does not name a test or provide a reliable reproduction.