ZERO_COPY and memory leaks

[segabriel]

Hello guys!
I got started updating to the latest rsokcet version and try to use ZERO_COPY. But time to time I catch memory leaks.

Here is my gists example --> https://gist.github.com/segabriel/b8a6a189ff89a9560a279cd97cc4a8fe

I used the following versions:

    <reactor.version>Dysprosium-RELEASE</reactor.version>
    <rsocket.version>1.0.0-RC5</rsocket.version>
    <netty.version>4.1.37.Final</netty.version>
    <reactor-netty.version>0.9.0.RELEASE</reactor-netty.version>

An example of exception:

E 2020-01-13T14:16:51,757 i.n.u.ResourceLeakDetector LEAK: ByteBuf.release() was not called before it's garbage-collected. See http://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
#1:
	io.netty.util.ReferenceCountUtil.release(ReferenceCountUtil.java:88)
	reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:210)
	reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:329)
	reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:348)
	reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:91)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:321)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:308)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:424)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:792)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:424)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:326)
	io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:748)
#2:
	io.netty.buffer.AdvancedLeakAwareByteBuf.readShort(AdvancedLeakAwareByteBuf.java:412)
	io.rsocket.frame.FrameHeaderFlyweight.frameType(FrameHeaderFlyweight.java:98)
	io.rsocket.frame.decoder.ZeroCopyPayloadDecoder.apply(ZeroCopyPayloadDecoder.java:18)
	io.rsocket.frame.decoder.ZeroCopyPayloadDecoder.apply(ZeroCopyPayloadDecoder.java:13)
	io.rsocket.RSocketRequester.handleFrame(RSocketRequester.java:572)
	io.rsocket.RSocketRequester.handleIncomingFrames(RSocketRequester.java:516)
	reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160)
	reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242)
	reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drainRegular(FluxGroupBy.java:554)
	reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drain(FluxGroupBy.java:630)
	reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.onNext(FluxGroupBy.java:670)
	reactor.core.publisher.FluxGroupBy$GroupByMain.onNext(FluxGroupBy.java:205)
	reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114)
	reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114)
	reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:206)
	reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:329)
	reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:348)
	reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:91)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:321)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:308)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:424)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:792)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:424)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:326)
	io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:748)
#3:
	io.netty.buffer.AdvancedLeakAwareByteBuf.skipBytes(AdvancedLeakAwareByteBuf.java:532)
	io.rsocket.frame.FrameHeaderFlyweight.frameType(FrameHeaderFlyweight.java:97)
	io.rsocket.frame.decoder.ZeroCopyPayloadDecoder.apply(ZeroCopyPayloadDecoder.java:18)
	io.rsocket.frame.decoder.ZeroCopyPayloadDecoder.apply(ZeroCopyPayloadDecoder.java:13)
	io.rsocket.RSocketRequester.handleFrame(RSocketRequester.java:572)
	io.rsocket.RSocketRequester.handleIncomingFrames(RSocketRequester.java:516)
	reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160)
	reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242)
	reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drainRegular(FluxGroupBy.java:554)
	reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drain(FluxGroupBy.java:630)
	reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.onNext(FluxGroupBy.java:670)
	reactor.core.publisher.FluxGroupBy$GroupByMain.onNext(FluxGroupBy.java:205)
	reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114)
	reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114)
	reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:206)
	reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:329)
	reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:348)
	reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:91)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:321)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:308)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:424)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:792)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:424)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:326)
	io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:748)
#4:
	io.netty.buffer.AdvancedLeakAwareByteBuf.skipBytes(AdvancedLeakAwareByteBuf.java:532)
	io.rsocket.frame.FrameHeaderFlyweight.frameType(FrameHeaderFlyweight.java:97)
	io.rsocket.RSocketRequester.handleIncomingFrames(RSocketRequester.java:512)
	reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160)
	reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242)
	reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drainRegular(FluxGroupBy.java:554)
	reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drain(FluxGroupBy.java:630)
	reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.onNext(FluxGroupBy.java:670)
	reactor.core.publisher.FluxGroupBy$GroupByMain.onNext(FluxGroupBy.java:205)
	reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114)
	reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114)
	reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:206)
	reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:329)
	reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:348)
	reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:91)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:321)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:308)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:424)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:792)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:424)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:326)
	io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:748)
#5:
	io.netty.buffer.AdvancedLeakAwareByteBuf.slice(AdvancedLeakAwareByteBuf.java:76)
	io.rsocket.frame.FrameLengthFlyweight.frame(FrameLengthFlyweight.java:51)
	io.rsocket.transport.netty.TcpDuplexConnection.decode(TcpDuplexConnection.java:95)
	reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100)
	reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114)
	reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:206)
	reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:329)
	reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:348)
	reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:91)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:321)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:308)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:424)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:792)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:424)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:326)
	io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:748)
#6:
	io.netty.buffer.AdvancedLeakAwareByteBuf.skipBytes(AdvancedLeakAwareByteBuf.java:532)
	io.rsocket.frame.FrameLengthFlyweight.frame(FrameLengthFlyweight.java:50)
	io.rsocket.transport.netty.TcpDuplexConnection.decode(TcpDuplexConnection.java:95)
	reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100)
	reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114)
	reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:206)
	reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:329)
	reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:348)
	reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:91)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:321)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:308)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:424)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:792)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:424)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:326)
	io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:748)
#7:
	Hint: 'reactor.right.reactiveBridge' will handle the message from this point.
	io.netty.channel.DefaultChannelPipeline.touch(DefaultChannelPipeline.java:116)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:357)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:321)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:308)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:424)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:792)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:424)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:326)
	io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:748)
Created at:
	io.netty.buffer.SimpleLeakAwareByteBuf.unwrappedDerived(SimpleLeakAwareByteBuf.java:143)
	io.netty.buffer.SimpleLeakAwareByteBuf.retainedSlice(SimpleLeakAwareByteBuf.java:57)
	io.netty.buffer.AdvancedLeakAwareByteBuf.retainedSlice(AdvancedLeakAwareByteBuf.java:95)
	io.netty.handler.codec.LengthFieldBasedFrameDecoder.extractFrame(LengthFieldBasedFrameDecoder.java:508)
	io.netty.handler.codec.LengthFieldBasedFrameDecoder.decode(LengthFieldBasedFrameDecoder.java:439)
	io.netty.handler.codec.LengthFieldBasedFrameDecoder.decode(LengthFieldBasedFrameDecoder.java:334)
	io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:500)
	io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:439)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:792)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:424)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:326)
	io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:748)
: 22 leak records were discarded because the leak record count is targeted to 4. Use system property io.netty.leakDetection.targetRecords to increase the limit. [reactor-tcp-epoll-1]

[mostroverkhov]

Logs show buffers are leaked on the receiving side, and there is race between cancellation (blockFirst) and received stream on client so messages could be dropped by Reactor - maybe try release them with Hooks.onNextDropped(ReferenceCountUtil::safeRelease);

[mostroverkhov]

(noticed intent was to update actual software)
For reasons outlined at the bottom of #739 It is better to stick with 0.12.2-RC4 (unless you want to hate your life) until next version is released and known to be stable.

[OlegDokuka]

@segabriel I have tried to run this sample but the issue was not reproducing on my machine™️. I suspect it is indeed somewhere in the hooks

I will doublecheck that later but right now it is a minor issue

[rstoyanchev]

This is easy to reproduce with the sample.

Logs show buffers are leaked on the receiving side, and there is race between cancellation (blockFirst) and received stream on client so messages could be dropped by Reactor - maybe try release them with Hooks.onNextDropped(ReferenceCountUtil::safeRelease);

This is correct. The last access record within RSocket is in RSocketRequester#handleFrame where it pushes into the UnicastProcessor. From there it may be dropped or it could get slightly ahead and get into the queue but drain can exit. Hooks.onNextDropped helps with the former but doesn't cover the latter. @simonbasle what do you think? doOnDiscard hook, something else?

There are probably other issues lurking where operators like flatMap cache items. For that there needs be doOnDiscard inserted in RSocket Java. Another potential case of a leak is where a Payload is taken as input, and if the response publisher is cancelled before the input is sent.

We'll need some buffer leak tests to prove and fix these.

[OlegDokuka]

@rstoyanchev Did you reproduce that with RC5 or the snapshot one?

I mean I don't see any "easy" reproducible leaks on the snapshot

[OlegDokuka]

In any case. Should be addressed in #761. I have addressed that case in the PR