title: Is groupBy leaking in akka-stream?
body: <p>I want to write a flow on <code>akka-stream</code> for grouping events from infinite stream by session_uid and calculate sum of traffic for each session (details in my <a href="https://stackoverflow.com/questions/33840868/how-to-group-incoming-events-from-infinite-stream">previous question</a>).</p>
  
  <p>I am going to use <code>Source#groupBy</code> function for group events by session_uid but seems like this function accumulate all group keys inside and don't have a way to release them. This is caused <code>java.lang.OutOfMemoryError: Java heap space</code> exception. Here is code for reproduce it:</p>
  
  <pre><code>import akka.actor.ActorSystem
  import akka.stream.ActorMaterializer
  import akka.stream.scaladsl.{Flow, Sink, Source}
  
  import scala.util.Random
  
  object GroupByMemoryLeakApplication extends App {
    implicit val system = ActorSystem()
    import system.dispatcher
  
    implicit val materializer = ActorMaterializer()
  
    val bigString = Random.nextString(512 * 1024)
  
    // This is infinite stream of events (i.e. this is session ids)
    val eventsSource = Source(() =&gt; (1 to 1000000000).iterator)
      .map((i) =&gt; { (i, bigString + i) })
  
    // This is flow pass event through groupBy function
    val groupByFlow = Flow[(Int, String)]
      .groupBy(_._2)
      .map {
        case (sessionUid, sessionEvents) =&gt;
          sessionEvents
            .map(e =&gt; { println(e._1); e })
            .runWith(Sink.head)
      }
      .mapAsync(4)(identity)
  
    eventsSource
      .via(groupByFlow)
      .runWith(Sink.ignore)
      .onComplete(_ =&gt; system.shutdown())
  }
  </code></pre>
  
  <p>So, how to release grouping key (<code>sessionUid</code>) inside <code>groupBy</code> after complete processing of related stream of events (<code>sessionEvents</code>)?</p>
  
  <p>May be anybody known an other way for grouping events by session_uid base on <code>akka-stream</code>?</p>