FanoutPublisherSink leaks actors on upstream failure

[edwinchoi]

The implementation for FanoutPublisherSink doesn't terminate the actor FanoutProcessorImpl on upstream failure, leaving any elements buffered in FanoutOutputs strongly reachable from the StreamSupervisor instance.

I've tested this against the latest v2.5.16.

At the bottom of FanoutputProcessorImpl.fail, a comment indicates that the actor will be stopped after flushing... but if you take a look at FanoutOutput.error you can see that the failure flow never calls afterShutdown, which is how the actor is terminated by normal completion.

Example to reproduce:

// needs to be in akka package for access to StreamSupervisor
package akka.stream

import java.util.concurrent.CountDownLatch

import akka.NotUsed
import akka.actor._
import akka.stream.scaladsl._
import akka.stream.impl.StreamSupervisor

private case object Go
class MyActor(latch: CountDownLatch, numRepeat: Int, sourceF: () => Source[String, NotUsed]) extends Actor {
  implicit lazy val mat = ActorMaterializer()

  private var count = 0

  override def preStart(): Unit = self ! Go
  override def postStop(): Unit = latch.countDown()
  override def receive: Receive = {
    case Go if count < numRepeat =>
      count += 1
      val (_, source) = sourceF().preMaterialize()
      source.runForeach(_ => ()).onComplete(_ => self ! Go)(context.dispatcher)
    case Go =>
      mat.supervisor ! StreamSupervisor.GetChildren
    case StreamSupervisor.Children(xs) =>
      // this should report 0
      println("size=" + xs.size)
      context.stop(self)
  }
}

object ActorLeakTest extends App {
  val system = ActorSystem()
  val bigString = scala.util.Random.nextString(1000 * 1000) * 50

  locally {
    println("ABORT")
    val latch = new CountDownLatch(1)
    system.actorOf(Props(new MyActor(latch, 1000,
      () =>
        // if you uncomment this line, then there will be 1000x 50MB strings on the heap at the end of the test
        //Source.single((bigString + "0").dropRight(1)) ++
        Source.failed(new RuntimeException()))))
    latch.await()
  }

  locally {
    println("SHUTDOWN")
    val latch = new CountDownLatch(1)
    system.actorOf(Props(new MyActor(latch, 1000, () => Source.single("hello"))))
    latch.await()
  }

  system.terminate()
}

[patriknw]

Doesn't sound good. Would you be able to create a PR fixing the issue @edwinchoi ?