GoogleCloudPlatform / GoogleCloudPlatform/pubsub
PubSub publisher keeps getting RESOURCE_EXHAUSTED due to Pubsub quota
- Dominant language
- Java
- Stars
- 270
- Forks
- 156
- PR merge metrics
- No merged PRs in 30d
Description
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.
Contributor guide
Assessment
This issue has not been assessed yet.