RSocketRequester and RSocketResponder leak Reactor subscriptions when disconnecting

[mdindoffer]

Expected Behavior

RSocketRequester and RSocketResponder should dispose of all associated resources when terminating.

Actual Behavior

RSocketRequester and RSocketResponder do not dispose of internal Flux subscriptions, creating a leak.

Steps to Reproduce

1. Start a RequesterResponderLeakReproducerServer https://gist.github.com/mdindoffer/c0e65bd9075ca7bdc8b639ebbf2c0fde#file-requesterresponderleakreproducerserver-java
2. Start a RequesterResponderLeakReproducerClient https://gist.github.com/mdindoffer/c0e65bd9075ca7bdc8b639ebbf2c0fde#file-requesterresponderleakreproducerclient-java
3. Repeat step nr.2 ad lib
4. Observe the number of active subscriptions increasing on the server side:

14-06-2022 15:05:02.337 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
14-06-2022 15:05:04.352 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
14-06-2022 15:05:06.363 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
14-06-2022 15:05:08.385 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketRequester.<init>(RSocketRequester.java:99)=2
io.rsocket.core.RSocketResponder.<init>(RSocketResponder.java:87)=2
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
14-06-2022 15:05:10.415 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketRequester.<init>(RSocketRequester.java:99)=2
io.rsocket.core.RSocketResponder.<init>(RSocketResponder.java:87)=2
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
14-06-2022 15:05:12.459 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketRequester.<init>(RSocketRequester.java:99)=2
io.rsocket.core.RSocketResponder.<init>(RSocketResponder.java:87)=2
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
14-06-2022 15:05:14.475 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketRequester.<init>(RSocketRequester.java:97)=1
io.rsocket.core.RSocketRequester.<init>(RSocketRequester.java:99)=3
io.rsocket.core.RSocketResponder.<init>(RSocketResponder.java:87)=3
io.rsocket.core.RSocketResponder.<init>(RSocketResponder.java:91)=1
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
io.rsocket.internal.BaseDuplexConnection.<init>(BaseDuplexConnection.java:30)=2
io.rsocket.keepalive.KeepAliveHandler$DefaultKeepAliveHandler.start(KeepAliveHandler.java:30)=1
io.rsocket.keepalive.KeepAliveSupport.start(KeepAliveSupport.java:72)=1
io.rsocket.transport.netty.TcpDuplexConnection.<init>(TcpDuplexConnection.java:52)=4
io.rsocket.transport.netty.server.TcpServerTransport.lambda$start$0(TcpServerTransport.java:115)=6
io.rsocket.transport.netty.server.TcpServerTransport.lambda$start$0(TcpServerTransport.java:119)=5
14-06-2022 15:05:16.495 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketRequester.<init>(RSocketRequester.java:99)=4
io.rsocket.core.RSocketResponder.<init>(RSocketResponder.java:87)=4
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
14-06-2022 15:05:18.509 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketRequester.<init>(RSocketRequester.java:99)=4
io.rsocket.core.RSocketResponder.<init>(RSocketResponder.java:87)=4
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
14-06-2022 15:05:20.553 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketRequester.<init>(RSocketRequester.java:99)=6
io.rsocket.core.RSocketResponder.<init>(RSocketResponder.java:87)=6
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
14-06-2022 15:05:22.573 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketRequester.<init>(RSocketRequester.java:99)=6
io.rsocket.core.RSocketResponder.<init>(RSocketResponder.java:87)=6
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
14-06-2022 15:05:24.592 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketRequester.<init>(RSocketRequester.java:99)=6
io.rsocket.core.RSocketResponder.<init>(RSocketResponder.java:87)=6
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1
14-06-2022 15:05:26.636 [main] INFO  RequesterResponderLeakReproducerServer - Active Reactor subscriptions: 
io.rsocket.core.RSocketRequester.<init>(RSocketRequester.java:99)=8
io.rsocket.core.RSocketResponder.<init>(RSocketResponder.java:87)=8
io.rsocket.core.RSocketServer$2.lambda$get$2(RSocketServer.java:317)=1

Possible Solution

Calling dispose on the Disposable when terminating.

Your Environment

• JDK 17
• RSocket 1.1.2

[OlegDokuka]

hello @mdindoffer!

First of all, is there any memory leak that causes a problem?

To explain my question given the above context: GC collects the mentioned subscription, since neither RSocket Requester, nor Responder is stored on the server after the connection is closed. That said, even though the disposable created by -> connection.receive().subscribe(this::handleFrame, e -> {}) is unstored, the actual garbage is collected after all.

Also, to clarify, the cancellation of the subscription for connection.receive() is noop (https://github.com/rsocket/rsocket-java/blob/master/rsocket-core/src/main/java/io/rsocket/core/ClientServerInputMultiplexer.java#L244)

[mdindoffer]

@OlegDokuka No, the memory is reclaimed afaict, sorry for the confusion. The problem is just the subscription. We are using a modified version of the linked gist to monitor live application and hunt for leaks. An ever-growing number of subscriptions is suspicious and causes false-positive alerts. We found numerous cases of memory/resource leaks with this tooling, and the only false positives were these two here.

While we could just suppress these two subscription counters, it would be much better to actually fix the cause upstream - hence eagerly calling dispose when the subscriptions are no longer needed. Is there a reason to not dispose them explicitly? Relying on GC seems fragile at best.

[OlegDokuka]

gotcha, makes sense to me

[OlegDokuka]

@mdindoffer as it turned out, the issue it selves is related to reactor netty and should be fixed with #2441 and 1.1.3 release

[OlegDokuka]

delayed due to ongoing investigation in reactor-netty

[mdindoffer]

Any news regarding this?

[violetagg]

Any news regarding this?

this is still to be fixed