PubSub publisher keeps getting RESOURCE_EXHAUSTED due to Pubsub quota

[mkuthan]

When Pubsub quota is exceeded the publisher reports the following exception without proper error handling:

java.lang.RuntimeException: java.util.concurrent.ExecutionException: com.google.api.gax.rpc.ResourceExhaustedException: io.grpc.StatusRuntimeException: RESOURCE_EXHAUSTED: Quota exceeded for quota metric 'Regional publisher throughput, kB' and limit 'Regional publisher throughput, kB per minute per region' of service 'pubsub.googleapis.com' for consumer 'project_number:foobar'.
	at com.google.pubsub.kafka.sink.CloudPubSubSinkTask.flush(CloudPubSubSinkTask.java:341)
	at org.apache.kafka.connect.sink.SinkTask.preCommit(SinkTask.java:125)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.commitOffsets(WorkerSinkTask.java:382)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:212)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:200)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:184)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.util.concurrent.ExecutionException: com.google.api.gax.rpc.ResourceExhaustedException: io.grpc.StatusRuntimeException: RESOURCE_EXHAUSTED: Quota exceeded for quota metric 'Regional publisher throughput, kB' and limit 'Regional publisher throughput, kB per minute per region' of service 'pubsub.googleapis.com' for consumer 'project_number: foobar'.
	at com.google.common.util.concurrent.AbstractFuture.getDoneValue(AbstractFuture.java:564)
	at com.google.common.util.concurrent.AbstractFuture.get(AbstractFuture.java:525)
	at com.google.common.util.concurrent.AbstractFuture$TrustedFuture.get(AbstractFuture.java:102)
	at com.google.common.util.concurrent.ForwardingFuture.get(ForwardingFuture.java:62)
	at com.google.pubsub.kafka.sink.CloudPubSubSinkTask.flush(CloudPubSubSinkTask.java:339)
	... 11 common frames omitted

The exception is thrown by the flush() method, and Kafka Connect / Kafka Consumer doesn't commit offset (to achieve at least once semantics). The undelivered messages are sent by Kafka Connect again and again at full rate. Effectively it kills the quota limits. I would expect back pressure / exponential backoff.