Memory leak in long running stream related to ConcurrentAsyncCallback's

[maxcom]

We have a long running stream that works fine in Akka 2.5.6. After update to 2.5.7, process runs out of memory after 5-10 minutes of processing. I see large number (~35000) of
GraphStageLogic.ConcurrentAsyncCallback instances in heap dump. Each of them holds a reference to some data objects that was processed in past.

I do not have a clean reproducer yet.

Here is screenshot from Eclipse Memory Analyzer:

and one more:

[patriknw]

@maxcon Thanks for reporting, it looks like a serious regression. The async callbacks were changed in 2.5.7. A reproducer would be great.

[maxcom]

Here is the reproducer: https://github.com/maxcom/akka-streams-oom.

Process runs fine for at least 10 minutes on Akka 2.5.6, and runs of of memory on first minute in 2.5.7.

[patriknw]

Thanks!

[2m]

With the new async callbacks we have started to store async callbacks in asyncCallbacksInProgress of the GraphStageLogic until graph stage is stopped.

Unfortunately streams of streams are implemented by using async callbacks, which never get released and thus OOM.

[johanandren]

To be more concrete, the issue here is akka.stream.stage.GraphStageLogic.SubSinkInlet which triggers the creation of one of those callbacks here:

akka/akka-stream/src/main/scala/akka/stream/stage/GraphStage.scala

Line 1250 in 3dc0adb

private val _sink = new SubSink[T](name, getAsyncCallback[ActorSubscriberMessage] { msg ⇒

For FlattenMerge one is created per sub stream that is materialized, but as there is no way to signal that the AsyncCallback should be released/not kept around anymore when the sub sink has completed/canceled it stays in memory.

Problem is probably there for the opposite as well (SubSourceOutlet)

[t3hnar]

Hi guys, any estimate on fix?

[patriknw]

We will take a deep look at this tomorrow.