I use the java library to subscribe to a subscription from my code. Using sbt:
"com.google.cloud" % "google-cloud-pubsub" % "0.24.0-beta"
I followed this guide to write a subscriber: https://cloud.google.com/pubsub/docs/pull
val projectId = "test-topic"
val subscriptionId = "test-sub"
def main(args: Array[String]): Unit = {
val subscriptionName = SubscriptionName.create(projectId, subscriptionId)
val subscriber = Subscriber.defaultBuilder(subscriptionName, new PastEventMessageReceiver()).build()
subscriber.startAsync()
System.in.read()
}
class PastEventMessageReceiver extends MessageReceiver {
override def receiveMessage(message: PubsubMessage, consumer: AckReplyConsumer): Unit = {
println(message)
consumer.ack()
}
It works great, I can pull out the published messages, but I constantly see this error in my journal several times a minute.
com.google.cloud.pubsub.v1.StreamingSubscriberConnection$1 onFailure
WARNING: Terminated streaming with exception
io.grpc.StatusRuntimeException: UNAVAILABLE: The service was unable to fulfill your request. Please try again. [code=8a75]
at io.grpc.Status.asRuntimeException(Status.java:526)
at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:385)
at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:422)
at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:61)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.close(ClientCallImpl.java:504)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.access$600(ClientCallImpl.java:425)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:536)
at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:102)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
When I first launch the application, there is a slight delay (about 1-2 minutes), where I do not see this error, after the delay I see it several times a minute. My app seems to still be able to pull messages tightly.
source
share