You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@camel.apache.org by "Claus Ibsen (Jira)" <ji...@apache.org> on 2021/02/25 19:37:00 UTC
[jira] [Updated] (CAMEL-16263) camel-google-pubsub - Consumer does
not recover from 500 series error from Google
[ https://issues.apache.org/jira/browse/CAMEL-16263?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Claus Ibsen updated CAMEL-16263:
--------------------------------
Summary: camel-google-pubsub - Consumer does not recover from 500 series error from Google (was: Consumer does not recover from 500 series error from Google)
> camel-google-pubsub - Consumer does not recover from 500 series error from Google
> ---------------------------------------------------------------------------------
>
> Key: CAMEL-16263
> URL: https://issues.apache.org/jira/browse/CAMEL-16263
> Project: Camel
> Issue Type: Bug
> Components: camel-google-pubsub
> Affects Versions: 3.8.0
> Reporter: Donal Anglin
> Priority: Major
>
> When the {{GooglePubsubConsumer}} receives a 500 series error from Google I expect it to catch that exception, log it and continue trying to pull messages from PubSub.
> What is actually happening is when an error like that occurs, the {{SubscriberWrapper}} thread started by the {{GooglePubsubConsumer}} exits and never restarts. The application will pull no more messages from PubSub for the lifetime of the application.
> The error is intermittent but can be reproduced using a simple route and waiting:
> {code:java}
> public class PubSubRoutes extends RouteBuilder {
> @Override
> public void configure() throws Exception {
> from("google-pubsub://{{gcp.project.id}}:{{gcp.subscription.id}}?synchronousPull=true")
> .log(LoggingLevel.INFO, "MessageReceived! ${body}");
> }
> }
> {code}
> Sample project: [https://github.com/anglind/camel-pubsub-bug]
> The error log:
> {code:java}
> 16:58:36.394 [com.github.anglind.App.main()] INFO o.a.c.c.g.p.GooglePubsubConsumer - Starting Google PubSub consumer for myproject/camel-test-topic-sub
> 16:58:36.450 [com.github.anglind.App.main()] INFO o.a.c.i.engine.AbstractCamelContext - Routes startup summary (total:1 started:1)
> 16:58:36.451 [com.github.anglind.App.main()] INFO o.a.c.i.engine.AbstractCamelContext - Started route1 (google-pubsub://myproject:camel-test-topic-sub)
> 16:58:36.452 [com.github.anglind.App.main()] INFO o.a.c.i.engine.AbstractCamelContext - Apache Camel 3.8.0 (camel-1) started in 308ms (build:42ms init:204ms start:62ms)
> 16:59:04.044 [Camel (camel-1) thread #0 - GooglePubsubConsumer[camel-test-topic-sub]] INFO route1 - MessageReceived! This is a message I published
> 18:23:12.719 [Camel (camel-1) thread #0 - GooglePubsubConsumer[camel-test-topic-sub]] ERROR o.a.c.c.g.p.GooglePubsubConsumer - Failure getting messages from PubSub
> com.google.api.gax.rpc.DeadlineExceededException: io.grpc.StatusRuntimeException: DEADLINE_EXCEEDED: deadline exceeded after 23.965861200s. [buffered_nanos=7490862100, buffered_nanos=65424900, remote_addr=pubsub.googleapis.com/74.125.193.95:443]
> at com.google.api.gax.rpc.ApiExceptionFactory.createException(ApiExceptionFactory.java:51)
> at com.google.api.gax.grpc.GrpcApiExceptionFactory.create(GrpcApiExceptionFactory.java:72)
> at com.google.api.gax.grpc.GrpcApiExceptionFactory.create(GrpcApiExceptionFactory.java:60)
> at com.google.api.gax.grpc.GrpcExceptionCallable$ExceptionTransformingFuture.onFailure(GrpcExceptionCallable.java:97)
> at com.google.api.core.ApiFutures$1.onFailure(ApiFutures.java:68)
> at com.google.common.util.concurrent.Futures$CallbackListener.run(Futures.java:1041)
> at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
> at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1215)
> at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:983)
> at com.google.common.util.concurrent.AbstractFuture.setException(AbstractFuture.java:771)
> at io.grpc.stub.ClientCalls$GrpcFuture.setException(ClientCalls.java:563)
> at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:533)
> at io.grpc.internal.DelayedClientCall$DelayedListener$3.run(DelayedClientCall.java:464)
> at io.grpc.internal.DelayedClientCall$DelayedListener.delayOrExecute(DelayedClientCall.java:428)
> at io.grpc.internal.DelayedClientCall$DelayedListener.onClose(DelayedClientCall.java:461)
> at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:617)
> at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70)
> at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:803)
> at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:782)
> at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
> at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
> 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:1149)
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> Suppressed: com.google.api.gax.rpc.AsyncTaskException: Asynchronous task failed
> at com.google.api.gax.rpc.ApiExceptions.callAndTranslateApiException(ApiExceptions.java:57)
> at com.google.api.gax.rpc.UnaryCallable.call(UnaryCallable.java:112)
> at org.apache.camel.component.google.pubsub.GooglePubsubConsumer$SubscriberWrapper.synchronousPull(GooglePubsubConsumer.java:152)
> at org.apache.camel.component.google.pubsub.GooglePubsubConsumer$SubscriberWrapper.run(GooglePubsubConsumer.java:113)
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> ... 3 common frames omitted
> Caused by: io.grpc.StatusRuntimeException: DEADLINE_EXCEEDED: deadline exceeded after 23.965861200s. [buffered_nanos=7490862100, buffered_nanos=65424900, remote_addr=pubsub.googleapis.com/74.125.193.95:443]
> at io.grpc.Status.asRuntimeException(Status.java:533)
> ... 17 common frames omitted
> {code}
> I think the component should catch and continue pulling for all the 500 series error codes listed on this page: https://cloud.google.com/pubsub/docs/reference/error-codes
--
This message was sent by Atlassian Jira
(v8.3.4#803005)