# MergeHub is not cleaning up in the heap area

**URL:** <https://discuss.akka.io/t/mergehub-is-not-cleaning-up-in-the-heap-area/9740>\
**Category:** Actors\
**Created:** [May 2, 2022, 9:25am UTC](https://discuss.akka.io/t/mergehub-is-not-cleaning-up-in-the-heap-area/9740 "2022-05-02T09:25:27Z")\
**Posts on this page:** 7\
**Page:** 1

<div class="post-metadata">

**Author:** ![salitha94](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/salitha94/32/2672_2.png) [@salitha94](https://discuss.akka.io/u/salitha94)\
**Post date:** [May 2, 2022, 9:25am UTC](https://discuss.akka.io/t/mergehub-is-not-cleaning-up-in-the-heap-area/9740/1 "2022-05-02T09:25:27Z")

</div>

We are experiencing something weird on Akka streams implementation. When so many incoming messages are there(after creating a set of actors simultaneously), we get full GC issues. We have analyzed backend using MAT and VisualVM. It complains that we have lot of **immutable scala lists** (Top dominators seen in MAT) connected to **BroadcasHub** is remaining in the heap, which is taking much space. When actors are killed, still **MergeHub** , **BroadcastHub** and **AbstractNodeQueue** and not cleaning in the heap area. They are keeping growing more and more when we instantiate more actors. But Flow is getting cleared though under this situation also.

**Play version: 2.7.3**  
**Akka version: 2.6.18**

 ![heap](https://us1.discourse-cdn.com/flex020/uploads/discuss_lightbend/original/2X/4/426ba57c9e7f6a57c82d0dbe500b935e46b17c1a.jpeg)

Code for creating hub sink.

```auto
        Pair<Sink<JsonNode, NotUsed>, Source<JsonNode, NotUsed>> sinkSourcePair =
        MergeHub.of(JsonNode.class, 16).toMat(BroadcastHub.of(JsonNode.class, 256), Keep.both()).run(getContext().getSystem());

        this.hubSink = sinkSourcePair.first();
        Source<JsonNode, NotUsed> hubSource = sinkSourcePair.second();

        adminResponseQueue = hubSink.runWith(Source.queue(1, OverflowStrategy.dropHead()), getContext().getSystem());

        Sink<JsonNode, CompletionStage<Done>> jsonSink = Sink.foreach(this::messageReceivedFromClient); // this method accepts json messages

        this.clientFlow = Flow.fromSinkAndSourceCoupled(jsonSink, hubSource)
            .watchTermination((n, stage) -> {
                // When the flow shuts down, make sure this actor also stops.
                stage.whenComplete((__, error) -> {
                    if(error != null) {
                        logger.error("Sink completed with exception", error);
                    }
                    context().stop(self());
                });
                return NotUsed.getInstance();
            }
        );

```

We expect a possible kind answer here.  
Can you give me any possible solution? @jrudolph

---

<div class="post-metadata">

**Author:** ![johanandren](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/johanandren/32/85_2.png) [@johanandren](https://discuss.akka.io/u/johanandren)\
**Post date:** [May 4, 2022, 6:26am UTC](https://discuss.akka.io/t/mergehub-is-not-cleaning-up-in-the-heap-area/9740/2 "2022-05-04T06:26:25Z")

</div>

Not sure if that explains your leak, maybe: You must not call `context.stop` from inside a future callback, it’s actor internal state so should not be touched from arbitrary threads. Instead pipe the future completion back to the actor as a message using `pipeTo` and make the actor stop when receiving that message.

---

<div class="post-metadata">

**Author:** ![salitha94](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/salitha94/32/2672_2.png) [@salitha94](https://discuss.akka.io/u/salitha94)\
**Post date:** [May 4, 2022, 7:03am UTC](https://discuss.akka.io/t/mergehub-is-not-cleaning-up-in-the-heap-area/9740/3 "2022-05-04T07:03:38Z")

</div>

@johanandren Context stop call was there in Akka example repos also. Please refer here: [play-java-websocket-example/UserActor.java at 2.7.x · playframework/play-java-websocket-example · GitHub](https://github.com/playframework/play-java-websocket-example/blob/2.7.x/app/actors/UserActor.java)

We use **whenComplete()** instead of **thenAccept()**. That is the only change.

And we have noticed same kind of issue has been raised here: [[akka-actor\_2.12 2.5.21] AbstractNodeQueue is not getting garbage collected even after multiple full GC · Issue #27255 · akka/akka · GitHub](https://github.com/akka/akka/issues/27255#:~:text=likely%20reason%20is%20that%20the%20sink%20is%20too%20slow%20to%20service%20all%20the%20incoming%20elements.%20Since%20attaching%20new%20producers%20to%20a%20hub%20is%20not%20backpressured%20or%20limited%20you%20can%20run%20into%20OOM%20if%20you%20continuously%20attach%20an%20unlimited%20number%20of%20new%20producers%20to%20the%20hub), but the scenario is bit different. Eventually they are also having the memory leak in heap.

Due to this issue, we have to restart server time to time to clear the GC, which is hectic. In our service, there are lot of actors running in real time around 500 per 1 service instance.

Is there any other way to create Sink and Source dynamically, without using **MergeHub** or **BroadcastHub**? Could you please let me know? I could not find it in the docs.

---

<div class="post-metadata">

**Author:** ![johanandren](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/johanandren/32/85_2.png) [@johanandren](https://discuss.akka.io/u/johanandren)\
**Post date:** [May 4, 2022, 7:17am UTC](https://discuss.akka.io/t/mergehub-is-not-cleaning-up-in-the-heap-area/9740/4 "2022-05-04T07:17:11Z")

</div>

That is a bug in that sample, I’d recommend that you start with changing it into a stop message and see if the leak is still there after that.

That kind of dynamic fan-in/out is exactly what the merge/broadcast-hub are for, the alternative I can think of would be to make a lower level solution using `Source.actorRef` and `Sink.actorRef` and make the dynamic routing with regular actor messaging.

---

<div class="post-metadata">

**Author:** ![salitha94](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/salitha94/32/2672_2.png) [@salitha94](https://discuss.akka.io/u/salitha94)\
**Post date:** [May 5, 2022, 5:16am UTC](https://discuss.akka.io/t/mergehub-is-not-cleaning-up-in-the-heap-area/9740/5 "2022-05-05T05:16:55Z")

</div>

Stop message in the sense? Could you please elaborate more @johanandren ?

---

<div class="post-metadata">

**Author:** ![wsargent](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/wsargent/32/118_2.png) [@wsargent](https://discuss.akka.io/u/wsargent)\
**Post date:** [May 8, 2022, 3:04am UTC](https://discuss.akka.io/t/mergehub-is-not-cleaning-up-in-the-heap-area/9740/6 "2022-05-08T03:04:30Z")

</div>

I’ve added a PR to fix this in the 2.7.x branch.

> <https://github.com/playframework/play-samples/pull/164/files>
>
> Fix for the context.stop bug described in https://discuss.akka.io/t/mergeh…ub-is-not-cleaning-up-in-the-heap-area/9740

---

<div class="post-metadata">

**Author:** ![salitha94](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/salitha94/32/2672_2.png) [@salitha94](https://discuss.akka.io/u/salitha94)\
**Post date:** [May 29, 2022, 8:38am UTC](https://discuss.akka.io/t/mergehub-is-not-cleaning-up-in-the-heap-area/9740/7 "2022-05-29T08:38:40Z")

</div>

@wsargent @johanandren We have tried a different way. We have **implemented the hub sink with kill switch attached to sink**. Then while closing WS, we cleared the hub using that kill switch…

```auto
Pair<Pair<Sink<JsonNode, NotUsed>, UniqueKillSwitch>, Source<JsonNode, NotUsed>> sinkSourcePair =
                MergeHub.of(JsonNode.class, 16)
                        .viaMat(KillSwitches.single(), Keep.both())
                        .toMat(BroadcastHub.of(JsonNode.class, 256), Keep.both())
                        .run(getContext().getSystem());

        this.hubSink = sinkSourcePair.first().first();
        this.hubKillSwitch = sinkSourcePair.first().second();
        Source<JsonNode, NotUsed> hubSource = sinkSourcePair.second();
        Sink<JsonNode, CompletionStage<Done>> jsonSink = Sink.foreach(this::messageReceivedFromClient);
        this.clientFlow = Flow.fromSinkAndSourceCoupled(jsonSink, hubSource)
            .watchTermination((n, stage) -> {
                stage.whenComplete((__, error) -> {
                    if(error != null) {
                        logger.error("Sink completed with exception", error);
                    }
                    cleanUpHub(error);
                    context().stop(self());
                });
                return NotUsed.getInstance();
            }
        );
    }

    // cleanup method for hub
    protected void cleanUpHub(Throwable t) {
        try {
            if(t == null)
                this.hubKillSwitch.shutdown();
            else
                this.hubKillSwitch.abort(t);
        } catch(Exception ex) {
            logger.error("Error cleaning up", ex);
        }
    }

```

When we locally checked this using **VisualVm** heap dumps, we saw that heap is not getting filled as previous.

And this was deployed also. Currently it’s under monitoring. So far, we did not get any Full GC issues. We have checked production heap dumps also using MAT. The heap size itself has been drastically reduced! **Previously it was more than 2GB** (since service Xmx = 2GB, heap grows until that). **But now the max heap size we saw is around 600MB**! And now these heap dumps does not contain **LocalActorRef / BroadcastHub / Scala.Immutable Lists** objects which was causing the memory leak…

I will update after a monitoring of 1 week…