# Akka Stream: initial backpressure - throttle on source has no effect

**URL:** <https://discuss.akka.io/t/akka-stream-initial-backpressure-throttle-on-source-has-no-effect/5865>\
**Category:** Akka Streams & Alpakka\
**Tags:** akka, scala, alpakka, streams\
**Created:** [February 5, 2020, 12:23pm UTC](https://discuss.akka.io/t/akka-stream-initial-backpressure-throttle-on-source-has-no-effect/5865 "2020-02-05T12:23:13Z")\
**Posts on this page:** 8\
**Page:** 1

<div class="post-metadata">

**Author:** ![mdibaiee](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/mdibaiee/32/1761_2.png) [@mdibaiee](https://discuss.akka.io/u/mdibaiee)\
**Post date:** [February 5, 2020, 12:23pm UTC](https://discuss.akka.io/t/akka-stream-initial-backpressure-throttle-on-source-has-no-effect/5865/1 "2020-02-05T12:23:14Z")

</div>

Hi,

Context:  
We use Alpakka Google PubSub GRPC to subscribe to messages from PubSub over GPRC. This is a Streaming subscription which receives messages in a stream.

The problem:  
When the service first starts up and connects to PubSub, there is no backpressure yet, so Alpakka/Source starts to pull a ton of messages from PubSub, which we can’t respond to in a timely manner, so they are left hanging in a buffer/queue that takes too long to be processed. Since this is on PubSub, it means that some of the messages which have not been discarded, but rather buffered, are going to be sent to the service again by PubSub ([that’s how retries work on PubSub: you get a “loan” on each message you receive, but if you don’t “acknowledge” the message in a specified period, the message is retried](https://cloud.google.com/pubsub/docs/subscriber)). Now since the service already has a queue full of messages to process, the retries will again be in the same state and the cycle continues on and our pubsub queue doesn’t drain. Horizontally scaling the service only makes this worse, each replica will fetch a ton of messages but now messages are sent between services and the effect is amplified.

Backpressure seems to work after that because after a short period of time passes after the startup, the number of requested messages from PubSub decreases significantly (and it matches our expectations of the throughput of our system). But, we have been looking for ways to avoid that initial spike in messages.

The spike is about a rate of ~3000-4000/s in the first 2-3 minutes, and then it goes down to 200/s. This is what it looks like when the service is restarted (or the connection to pubsub is lost and and a new connection is made):  
 ![image](https://us1.discourse-cdn.com/flex020/uploads/discuss_lightbend/original/2X/e/eb1f88f5b74792f036e3894b56bd623200fbd714.png)

We tried these things:

Using `.throttle` on the source, or even [a ticking `.zip` on the source](https://doc.akka.io/docs/akka/current/stream/stream-cookbook.html#triggering-the-flow-of-elements-programmatically) to slow down consumption: doesn’t work  
Using `.buffer(1, OverflowStrategy.backpressure)` on the source: doesn’t work  
Decreasing values of these configurations (basically reducing buffer capacity and output burst limit significantly):

```auto
     initial-input-buffer-size = 1
     max-input-buffer-size = 2
     output-burst-limit = 50

```

When I say “doesn’t work” I mean it does not reduce the initial spike, although the effect on the pipeline is visible for all these cases: a significant throttle of 1 per 1 second is visible in the processing pipeline and after the initial spike, no more than one item is received from pubsub per second, but the spike in the beginning is left untouched!

We would really appreciate it if someone could help us understand how can we apply a limit/throttling on the source or somehow overcome this issue.  
Thank you!

---

<div class="post-metadata">

**Author:** ![mdibaiee](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/mdibaiee/32/1761_2.png) [@mdibaiee](https://discuss.akka.io/u/mdibaiee)\
**Post date:** [February 5, 2020, 12:34pm UTC](https://discuss.akka.io/t/akka-stream-initial-backpressure-throttle-on-source-has-no-effect/5865/2 "2020-02-05T12:34:59Z")

</div>

Worth mentioning of the versions we use:

```auto
  "com.lightbend.akka" %% "akka-stream-alpakka-hbase" % "1.1.0",
  "com.lightbend.akka" %% "akka-stream-alpakka-google-cloud-pub-sub-grpc" % "1.1.0",

```

---

<div class="post-metadata">

**Author:** ![jrudolph](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/jrudolph/32/80_2.png) [@jrudolph](https://discuss.akka.io/u/jrudolph)\
**Post date:** [February 6, 2020, 10:33am UTC](https://discuss.akka.io/t/akka-stream-initial-backpressure-throttle-on-source-has-no-effect/5865/3 "2020-02-06T10:33:10Z")

</div>

Thanks for the question, @mdibaiee.

> [@mdibaiee](#):
>
> When the service first starts up and connects to PubSub, there is no backpressure yet, so Alpakka/Source starts to pull a ton of messages from PubSub

Why is there no backpressure yet? And where is the buffer/queue? Is that something in your code or part of the alpakka connector?

---

<div class="post-metadata">

**Author:** ![mdibaiee](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/mdibaiee/32/1761_2.png) [@mdibaiee](https://discuss.akka.io/u/mdibaiee)\
**Post date:** [February 6, 2020, 10:39am UTC](https://discuss.akka.io/t/akka-stream-initial-backpressure-throttle-on-source-has-no-effect/5865/4 "2020-02-06T10:39:27Z")

</div>

Hey @jrudolph, thanks for the response!

> Why is there no backpressure yet?

The server has just started, so I suppose there is no data yet about how much downstream can handle, hence no back pressure? So I’m not sure how the initial demand from downstream looks like. It’s something I suspect, not necessarily reflective of how Akka works under the hood because I don’t know how it works.

> And where is the buffer/queue? Is that something in your code or part of the alpakka connector?

This is also something we suspect Akka is handling on its own. We have not explicitly defined any buffer / queue.

I turned on `FINE` logging to be sure Akka is indeed “receiving” those messages, and yes, the logs show a spike in `ReceivedMessages` from PubSub in the beginning, even though downstream consumes those events slowly and never catches up. After a few minutes the amount of `ReceivedMessages` drops to a low rate, almost on-par with downstream, but there is already a ton of them waiting to be processed.

So if Akka is receiving all these messages, but they are being processed very slowly downstream, I suspect Akka has a kind of buffer/queue where it holds all those messages in.

---

<div class="post-metadata">

**Author:** ![javiramos1](https://avatars.discourse-cdn.com/v4/letter/j/dfb087/32.png) [@javiramos1](https://discuss.akka.io/u/javiramos1)\
**Post date:** [February 6, 2020, 10:52am UTC](https://discuss.akka.io/t/akka-stream-initial-backpressure-throttle-on-source-has-no-effect/5865/5 "2020-02-06T10:52:32Z")

</div>

Hi, I;m working with @mdibaiee

This is how we setup the source:

```auto
val source: Source[ReceivedMessage, Future[Cancellable]] = GooglePubSub.subscribe(request, 1.seconds)

```

And these are the settings:

```auto
 stream.materializer {
    # Initial size of buffers used in stream elements
    initial-input-buffer-size = 4
    # Maximum size of buffers used in stream elements
    max-input-buffer-size = 16
  }

```

---

<div class="post-metadata">

**Author:** ![jrudolph](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/jrudolph/32/80_2.png) [@jrudolph](https://discuss.akka.io/u/jrudolph)\
**Post date:** [February 6, 2020, 12:22pm UTC](https://discuss.akka.io/t/akka-stream-initial-backpressure-throttle-on-source-has-no-effect/5865/6 "2020-02-06T12:22:02Z")

</div>

Thanks for that information.

In general, akka-stream applies backpressure from start to end. But as you say, buffers might introduce unintended delays. So, it goes somewhat against the intention of akka-stream / alpakka to introduce extra buffers/queues so if the behavior should be caused by alpakka google pubsub, I’d say it’s a bug. At this point, it is hard to say whether the behavior is caused by alpakka or your stream setup. It would be best if you could provide a standalone example that would showcase what you are seeing.

Johannes

---

<div class="post-metadata">

**Author:** ![johanandren](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/johanandren/32/85_2.png) [@johanandren](https://discuss.akka.io/u/johanandren)\
**Post date:** [February 11, 2020, 7:56am UTC](https://discuss.akka.io/t/akka-stream-initial-backpressure-throttle-on-source-has-no-effect/5865/7 "2020-02-11T07:56:00Z")

</div>

One tricky thing with Akka gRPC (if that’s the one you are using) is that the underlying Netty client has quite large buffers internally, see [https://github.com/akka/akka-grpc/issues/501](https://github.com/akka/akka-grpc/issues/501)  
I wonder if that can explain what you are seeing perhaps?

---

<div class="post-metadata">

**Author:** ![jrudolph](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/jrudolph/32/80_2.png) [@jrudolph](https://discuss.akka.io/u/jrudolph)\
**Post date:** [February 11, 2020, 8:48am UTC](https://discuss.akka.io/t/akka-stream-initial-backpressure-throttle-on-source-has-no-effect/5865/8 "2020-02-11T08:48:05Z")

</div>

Great observation.

Google PubSub has even documented that fact here: [https://cloud.google.com/pubsub/docs/pull#streamingpull\_dealing\_with\_large\_backlogs\_of\_small\_messages](https://cloud.google.com/pubsub/docs/pull#streamingpull_dealing_with_large_backlogs_of_small_messages)

I wonder if there might be an alternative implementation that doesn’t rely on transport level buffers for backpressure. Created a ticket to discuss that here: [https://github.com/akka/alpakka/issues/2141](https://github.com/akka/alpakka/issues/2141)