Buffer LEAK detected by ResourceLeakDetector in AuthenticationPayloadExchangeConverter

[kevinat]

Describe the bug

Stack trace:

[ERROR] io.netty.util.ResourceLeakDetector LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
Created at:
	io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:385)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:173)
	io.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:107)
	org.springframework.security.rsocket.authentication.AuthenticationPayloadExchangeConverter.authentication(AuthenticationPayloadExchangeConverter.java:74)
	org.springframework.security.rsocket.authentication.AuthenticationPayloadExchangeConverter.lambda$convert$1(AuthenticationPayloadExchangeConverter.java:66)
	reactor.core.publisher.FluxFlatMap.trySubscribeScalarMap(FluxFlatMap.java:151)
	reactor.core.publisher.MonoFlatMap.subscribeOrReturn(MonoFlatMap.java:53)
	reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:57)
	reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
	reactor.core.publisher.Mono.subscribe(Mono.java:4046)
	reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:173)
	reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56)
	reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
	reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
	reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
	io.rsocket.core.RSocketResponder.handleRequestResponse(RSocketResponder.java:345)
	io.rsocket.core.RSocketResponder.handleFrame(RSocketResponder.java:200)
	reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160)
	io.rsocket.core.ClientServerInputMultiplexer$InternalDuplexConnection.onNext(ClientServerInputMultiplexer.java:248)
	io.rsocket.core.ClientServerInputMultiplexer.onNext(ClientServerInputMultiplexer.java:129)
	io.rsocket.core.ClientServerInputMultiplexer.onNext(ClientServerInputMultiplexer.java:48)
	io.rsocket.core.SetupHandlingDuplexConnection.onNext(SetupHandlingDuplexConnection.java:118)
	io.rsocket.core.SetupHandlingDuplexConnection.onNext(SetupHandlingDuplexConnection.java:19)
	reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
	reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:267)
	reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:377)
	reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:381)
	reactor.netty.http.server.HttpServerOperations.onInboundNext(HttpServerOperations.java:559)
	reactor.netty.http.server.WebsocketServerOperations.onInboundNext(WebsocketServerOperations.java:161)
	reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:94)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	io.rsocket.transport.netty.server.BaseWebsocketServerTransport$PongHandler.channelRead(BaseWebsocketServerTransport.java:63)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
	io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
	io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
	io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.base/java.lang.Thread.run(Thread.java:834)

To Reproduce
Enable RSocket security, run for a while, when make rsocket request again, error stack prints.

Expected behavior

Maybe this ByteBuf should be release?

74:		ByteBuf rawAuthentication = ByteBufAllocator.DEFAULT.buffer().writeBytes(authenticationMetadata);

[jzheaux]

According to https://netty.io/wiki/reference-counted-objects.html, allocated buffers need to be proactively released, so I imagine that the code needs to be changed to look something like:

ByteBuf rawAuthentication = ByteBufAllocator.DEFAULT.buffer();
try {
  // ... produce Authentication
} finally {
    rawAuthentication.release();
}

Would you be able to submit a pull request?

[kevinat]

Pull request submit: #9671