# Google Pub/Sub stops pulling randomly
**URL:** https://discuss.akka.io/t/google-pub-sub-stops-pulling-randomly/3759
**Category:** Akka Streams & Alpakka
**Created:** [March 26, 2019, 9:57am UTC](https://discuss.akka.io/t/google-pub-sub-stops-pulling-randomly/3759 "2019-03-26T09:57:21Z")
**Posts on this page:** 3
**Page:** 1
### Author:  [@DavidPuleri](https://discuss.akka.io/u/DavidPuleri)
#### Post date: [March 26, 2019, 9:57am UTC](https://discuss.akka.io/t/google-pub-sub-stops-pulling-randomly/3759/1 "2019-03-26T09:57:21Z")
Hello,
I follow the documentation to implement a system that subscribes to the http log flow on my kubernetes cluster and (will) send the json object to another system for analyses.
It works for a few items then it stops:
```auto
[DEBUG] [03/26/2019 10:44:48.682] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (Idle)] Dispatching request [POST /v1/projects/xxxxx-platform/subscriptions/yyyyyy:acknowledge Strict(360 bytes)]
[DEBUG] [03/26/2019 10:44:48.682] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (Idle)] Before event [onNewRequest] In state [Idle] for [0 ms]
[DEBUG] [03/26/2019 10:44:48.682] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (PushingRequestToConnection)] After event [onNewRequest] State change [Idle] -> [PushingRequestToConnection]
[DEBUG] [03/26/2019 10:44:48.682] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (PushingRequestToConnection)] Before event [onRequestDispatched] In state [PushingRequestToConnection] for [0 ms]
[DEBUG] [03/26/2019 10:44:48.682] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (WaitingForResponse)] After event [onRequestDispatched] State change [PushingRequestToConnection] -> [WaitingForResponse]
[DEBUG] [03/26/2019 10:44:48.715] [local-akka.actor.default-dispatcher-15] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (WaitingForResponse)] Received response
[DEBUG] [03/26/2019 10:44:48.715] [local-akka.actor.default-dispatcher-15] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (WaitingForResponse)] Before event [onResponseReceived] In state [WaitingForResponse] for [32 ms]
[DEBUG] [03/26/2019 10:44:48.715] [local-akka.actor.default-dispatcher-15] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (WaitingForResponse)] onResponseReceived in WaitingForResponse with false
[DEBUG] [03/26/2019 10:44:48.715] [local-akka.actor.default-dispatcher-15] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (WaitingForResponseDispatch)] After event [onResponseReceived] State change [WaitingForResponse] -> [WaitingForResponseDispatch]
[DEBUG] [03/26/2019 10:44:48.715] [local-akka.actor.default-dispatcher-15] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (WaitingForResponseDispatch)] Before event [onResponseDispatchable] In state [WaitingForResponseDispatch] for [0 ms]
[DEBUG] [03/26/2019 10:44:48.715] [local-akka.actor.default-dispatcher-15] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (WaitingForResponseEntitySubscription)] After event [onResponseDispatchable] State change [WaitingForResponseDispatch] -> [WaitingForResponseEntitySubscription]
[DEBUG] [03/26/2019 10:44:48.716] [local-akka.actor.default-dispatcher-5] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (WaitingForResponseEntitySubscription)] Before event [onResponseEntitySubscribed] In state [WaitingForResponseEntitySubscription] for [0 ms]
[DEBUG] [03/26/2019 10:44:48.716] [local-akka.actor.default-dispatcher-5] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (WaitingForEndOfResponseEntity)] After event [onResponseEntitySubscribed] State change [WaitingForResponseEntitySubscription] -> [WaitingForEndOfResponseEntity]
[DEBUG] [03/26/2019 10:44:48.716] [local-akka.actor.default-dispatcher-16] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (WaitingForEndOfResponseEntity)] Before event [onResponseEntityCompleted] In state [WaitingForEndOfResponseEntity] for [0 ms]
[DEBUG] [03/26/2019 10:44:48.716] [local-akka.actor.default-dispatcher-16] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (Idle)] After event [onResponseEntityCompleted] State change [WaitingForEndOfResponseEntity] -> [Idle]
[DEBUG] [03/26/2019 10:44:48.716] [local-akka.actor.default-dispatcher-16] [local/Pool(shared->https://pubsub.googleapis.com:443)] [0 (Idle)] Slot became idle... Trying to pull
[DEBUG] [03/26/2019 10:46:44.207] [local-akka.actor.default-dispatcher-16] [local/Pool(shared->https://www.googleapis.com:443)] Shutting down host connection pool immediately
[DEBUG] [03/26/2019 10:46:44.208] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://www.googleapis.com:443)] Pool upstream was completed
[DEBUG] [03/26/2019 10:46:44.208] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://www.googleapis.com:443)] Pool stopped
[DEBUG] [03/26/2019 10:46:46.745] [local-akka.actor.default-dispatcher-12] [akka://local/system/StreamSupervisor-1/flow-4-0-PoolFlow] Aborting tcp connection to pubsub.googleapis.com:443 because of upstream failure: akka.stream.scaladsl.TcpIdleTimeoutException: TCP idle-timeout encountered on connection to [pubsub.googleapis.com:443], no bytes passed in the last 2 minutes
[DEBUG] [03/26/2019 10:46:46.746] [local-akka.actor.default-dispatcher-16] [akka://local/system/StreamSupervisor-1/flow-7-1] closing output
[DEBUG] [03/26/2019 10:46:46.748] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://pubsub.googleapis.com:443)] [1 (Idle)] Connection cancelled
[DEBUG] [03/26/2019 10:46:46.748] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://pubsub.googleapis.com:443)] [1 (Idle)] Connection completed
[DEBUG] [03/26/2019 10:46:46.748] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://pubsub.googleapis.com:443)] [1 (Idle)] Before event [onConnectionCompleted] In state [Idle] for [120901 ms]
[DEBUG] [03/26/2019 10:46:46.748] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://pubsub.googleapis.com:443)] [1 (Unconnected)] After event [onConnectionCompleted] State change [Idle] -> [Unconnected]
[DEBUG] [03/26/2019 10:46:46.748] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://pubsub.googleapis.com:443)] [1 (Unconnected)] State change from [Idle] to [Unconnected]. Closing the existing connection.
[DEBUG] [03/26/2019 10:46:48.735] [local-akka.actor.default-dispatcher-12] [local/Pool(shared->https://pubsub.googleapis.com:443)] Shutting down host connection pool immediately
[DEBUG] [03/26/2019 10:46:48.735] [local-akka.actor.default-dispatcher-5] [local/Pool(shared->https://pubsub.googleapis.com:443)] Pool upstream was completed
[DEBUG] [03/26/2019 10:46:48.735] [local-akka.actor.default-dispatcher-5] [local/Pool(shared->https://pubsub.googleapis.com:443)] Pool stopped
```
My implementations is very simple, and by the book :
```auto
val subscriptionSource: Source[ReceivedMessage, NotUsed] = GooglePubSub.subscribe(subscription, config)
val ackSink: Sink[AcknowledgeRequest, Future[Done]] = GooglePubSub.acknowledge(subscription, config)
val processMessage: Sink[ReceivedMessage, NotUsed] = Flow[ReceivedMessage].map{ message =>
read[GkeLogData](new String(java.util.Base64.getDecoder.decode(message.message.data)))
}.to(Sink.foreach{ d =>
println(Parser.anonymiseUuid(d.httpRequest.requestUrl))
})
val batchAckSink: Sink[ReceivedMessage, NotUsed] = Flow[ReceivedMessage]
.map(_.ackId)
.groupedWithin(5, 10 seconds)
.map(AcknowledgeRequest.apply)
.to(ackSink)
val s: RunnableGraph[NotUsed] = subscriptionSource
.alsoTo(batchAckSink)
.to(processMessage).run()
```
I wonder if this is not something to do with non-discarded entities as the symptoms are the same, but I cannot be sure. Any ideas ?
Thank you for your help !
---
### Author:  [@DavidPuleri](https://discuss.akka.io/u/DavidPuleri)
#### Post date: [March 26, 2019, 10:35am UTC](https://discuss.akka.io/t/google-pub-sub-stops-pulling-randomly/3759/2 "2019-03-26T10:35:48Z")
Swallowed exception in the processMessage visible by adding .log(“error”).
---
### Author:  [@ennru](https://discuss.akka.io/u/ennru)
#### Post date: [April 26, 2019, 10:06am UTC](https://discuss.akka.io/t/google-pub-sub-stops-pulling-randomly/3759/3 "2019-04-26T10:06:46Z")
I assume that finding the exception helped you solve the problem.
Make it a habit to always have a stream completion future (eg. from `Sink.ignore`) and check it for proper error-free completion.
Cheers,
Enno.