Akka Stream Stackoverflow Exception

Akka-stream-crash.scala

import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, SourceShape}
import akka.stream.scaladsl._

object CrashDemo extends App {

  implicit val system = ActorSystem("error-reproduction-demo")
  implicit val mat = ActorMaterializer()

  import system.dispatcher

  val numbers = Source(1 to 1000)
  val numberOfFlows = 5000

  Source.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[Unit] =>
    import GraphDSL.Implicits._

    val bcast = builder.add(Broadcast[Int](numberOfFlows))
    val merge = builder.add(Merge[Int](numberOfFlows))

    numbers ~> bcast

    for (x ← 1 to numberOfFlows) {
      bcast.out(x - 1) ~> Flow[Int].map(x ⇒ x * x) ~> merge.in(x - 1)
    }

    SourceShape(merge.out)
  }).runForeach(println).onComplete(_ ⇒  system.terminate())

}

output.txt

Exception in thread "main" java.lang.StackOverflowError
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	at akka.stream.impl.fusing.Fusing$.rewriteMat(Fusing.scala:415)
	....