OutOfMemoryError

[ProChirathF]

I am using SockJS with Play2.7 & I am seeing OutOfMemeory crashes intermittently. We never experienced this issue with Play2.5.
SockJS router is handled with an actor(WebSocketActor). WebSocketActor receives a very high volume of messages, hence large buffer is used(1048576) is defined for ActorFlow.actorRef

SockJs router

@Override
public SockJSSettings settings() {
	return new SockJSSettings().withWebsocket(true)
			.withSessionTimeout(FiniteDuration.create(5000, TimeUnit.MILLISECONDS))
			.withHeartbeat(FiniteDuration.create(20000, TimeUnit.MILLISECONDS))
			.withSessionBufferSize(512 * 1024).withStreamingQuota(4 * 1024);
}

// SockJS endpoint handler
@Override
public SockJS sockjs() {
	return SockJS.Text.acceptOrResult(request -> {
		if (isValid(request)) {
			return CompletableFuture
					.completedFuture(F.Either.<Result, Flow<String, String, ?>>Right(ActorFlow.actorRef(	
							actorRef -> createActorRef(() -> websocketActorFactory.create(actorRef, params)),
							1048576, // support high volume of messages
							OverflowStrategy.fail(),
							actorSystem,
							materializer)));
			
		} else {
			logger.error("Invalid request');
		}
		return CompletableFuture.completedFuture(F.Either.Left(forbidden()));
	});
}

private Props createActorRef(Supplier<Actor> create) {
     return Props.create(Actor.class, create::get);
}

Uncaught error from thread [application-akka.actor.default-dispatcher-65743]: Java heap space, shutting down JVM since 'akka.jvm-exit-on-fatal-error' is enabled for ActorSystem[application]
java.lang.OutOfMemoryError: Java heap space
	at akka.stream.impl.FixedSizeBuffer$FixedSizeBuffer.<init>(Buffers.scala:71)
	at akka.stream.impl.FixedSizeBuffer$PowerOfTwoFixedSizeBuffer.<init>(Buffers.scala:141)
	at akka.stream.impl.FixedSizeBuffer$.apply(Buffers.scala:65)
	at akka.stream.impl.Buffer$.apply(Buffers.scala:45)
	at akka.stream.impl.Buffer$.apply(Buffers.scala:40)
	at akka.stream.impl.ActorRefSource$$anon$1.<init>(ActorRefSource.scala:45)
	at akka.stream.impl.ActorRefSource.createLogicAndMaterializedValue(ActorRefSource.scala:39)
	at akka.stream.impl.GraphStageIsland.materializeAtomic(PhasedFusingActorMaterializer.scala:674)
	at akka.stream.impl.PhasedFusingActorMaterializer.materialize(PhasedFusingActorMaterializer.scala:491)
	at akka.stream.impl.PhasedFusingActorMaterializer.materialize(PhasedFusingActorMaterializer.scala:450)
	at akka.stream.impl.PhasedFusingActorMaterializer.materialize(PhasedFusingActorMaterializer.scala:443)
	at akka.stream.scaladsl.RunnableGraph.run(Flow.scala:629)
	at play.api.libs.streams.ActorFlow$.actorRef(ActorFlow.scala:46)
	at play.api.libs.streams.ActorFlow.actorRef(ActorFlow.scala)
	at play.libs.streams.ActorFlow.actorRef(ActorFlow.java:54)
	at mycontroller.controllers.SockJSRouterController.lambda$sockjs$2(SockJSRouterController.java:81)
	at mycontroller.controllers.SockJSRouterController$$Lambda$1771/0x0000000800e62c40.apply(Unknown Source)
	at play.sockjs.SockJS$1.apply(SockJS.java:134)
	at play.sockjs.core.j.JavaSockJS.$anonfun$withComponents$1(JavaSockJS.scala:23)
	at play.sockjs.core.j.JavaSockJS$$Lambda$1758/0x0000000800e5c040.apply(Unknown Source)
	at play.sockjs.api.SockJS$$anon$1.apply(SockJS.scala:34)
	at play.sockjs.core.transports.WebSocket.$anonfun$framed$2(WebSocket.scala:27)
	at play.sockjs.core.transports.WebSocket$$Lambda$1762/0x0000000800e5f040.apply(Unknown Source)
	at play.api.mvc.WebSocket$.$anonfun$acceptOrResult$1(WebSocket.scala:205)
	at play.api.mvc.WebSocket$$$Lambda$1765/0x0000000800e61040.apply(Unknown Source)
	at play.api.mvc.WebSocket$$anon$1.apply(WebSocket.scala:42)
	at play.core.server.AkkaHttpServer.executeHandler(AkkaHttpServer.scala:370)
	at play.core.server.AkkaHttpServer.handleRequest(AkkaHttpServer.scala:325)
	at play.core.server.AkkaHttpServer.$anonfun$createServerBinding$1(AkkaHttpServer.scala:189)
	at play.core.server.AkkaHttpServer$$Lambda$934/0x0000000800aa0840.apply(Unknown Source)
	at akka.stream.impl.fusing.MapAsync$$anon$30.onPush(Ops.scala:1261)
	at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:523)

Any thoughts or guidance is highly appreciated.

[fdimuccio]

Hi @ProChirathF, without further info it looks like working as intended. A size of 1048576 could be quite a lot for a buffer depending on the element you are storing and the number of actors instantiated.

From what I understood clients are not able to cope with the outgoing message rate and the entire flow is back pressured all the way up till your Actor wrapped by the ActorFlow. Usually you would end up implementing rate-limiting mechanisms within your Actor to stop producing messages if downstream is back pressuring, but unluckily ActorFlow doesn't give any mean to detect it.

What I can suggest is to try out akka.stream.scaladsl.Source#actorRefWithBackpressure that will allow you to use an Actor to implement your main logic but also gives you a way to control downstream back pressure. The downside is that you have to replace the ActorFlow with a plain flow (or roll your own ActorFlow implementation that uses akka.stream.scaladsl.Source#actorRefWithBackpressure instead of akka.stream.scaladsl.Source#actorRef).

I hope this helps.

Since this doesn't look like related to play2-sockjs, I'll be closing the issue. Feel free to re-open it if you have more findings that could lead to a memory leak within paly2-sockjs.

[ProChirathF]

@fdimuccio I have Implemented a custom flow using Source.actorRefWithAck (akka v2.5), which sends a 'ack' back to the user WebSocket actor after emitting the message to the downstream. This bachpressure mechanism together with a sensible inputBuffer on the source allows message throttling. Thank you for your advice!.
My code is in java, I am happy to submit my flow. I think it would be quite beneficial if this flow is available with the play-sockjs api ( open issue #26). Play Framework does not have a answer for this issue.
playframework/playframework#6246

Thanks