title: akka-stream pipeline backpressures despite inserted buffer
body: <p>I have an akka-stream pipeline that fans out events (via <code>BroadcastHub</code>) that are pushed into the stream via a <code>SourceQueueWithComplete</code>.</p>
  <p>Despite all downstream consumers having a <code>.buffer()</code> inserted (of which I'd expect that it ensures that upstream buffers of hub and queue remain drained), I still observe backpressure kicking in after the system has been running for a while.</p>
  <p>Here's a (simplified) snippet:</p>
  <pre><code>class NotificationHub[Event](
    implicit materializer: Materializer,
    ecForLogging: ExecutionContext
  ) {
  
    // a SourceQueue to enque events and a BroadcastHub to allow multiple subscribers
    private val (queue, broadCastSource) =
      Source.queue[Event](
        bufferSize = 64,
        // we expect the buffer to never run full and if it does, we want
        // to log that asap, so we use OverflowStrategy.backpressure
        OverflowStrategy.backpressure
      ).toMat(BroadcastHub.sink)(Keep.both).run()
  
    // This keeps the BroadCastHub drained while there are no subscribers
    // (see https://doc.akka.io/docs/akka/current/stream/stream-dynamic.html ):
    broadCastSource.to(Sink.ignore).run()
  
    def notificationSource(p: Event =&gt; Boolean): Source[Unit, NotUsed] = {
      broadCastSource
        .collect { case event if p(event) =&gt; () }
        // this buffer is intended to keep the upstream buffers of
        // queue and hub drained:
        .buffer(
          // if a downstream consumer ever becomes too slow to consume,
          // only the latest two notifications are relevant
          size = 2,
          // doesn't really matter whether we drop head or tail
          // as all elements are the same (), it's just important not
          // to backpressure in case of overflow:
          OverflowStrategy.dropHead
        )
    }
  
    def propagateEvent(
      event: Event
    ): Unit = {
      queue.offer(event).onComplete {
        case Failure(e) =&gt;
          // unexpected backpressure occurred!
          println(e.getMessage)
          e.printStackTrace()
        case _ =&gt;
          ()
      }
    }
  
  }
  </code></pre>
  <p>Since the doc for <code>buffer()</code> says that for <code>DropHead</code> it never backpressures, I would have expected that the upstream buffers remain drained. Yet still I end up with calls to <code>queue.offer()</code> failing because of backpressure.</p>
  <p>Reasons that I could think of:</p>
  <ol>
  <li>Evaluation of predicate <code>p</code> in <code>.collect</code> causes a lot of load and hence backpressure. This seems very unlikely because those are very simple non-blocking ops.</li>
  <li>Overall system is totally overloaded. Also rather unlikely.</li>
  </ol>
  <p>I have a felling I am missing something? Do I maybe need to add an async boundary via <code>.async</code> before or after <code>buffer()</code> to fully decouple the &quot;hub&quot; from possible heavy load that may occur somewhere further downstream?</p>