Spring-RSocket: unexpected IllegalReferenceCountException on requestChannel response error

[mostroverkhov]

trafficstars

Terminating channel response immediately with error leads to unexpected refcnt exception, plus 2 other exceptions (delivered through different code paths) which make little sense in local "catch all" handler - original outbound error should only be delivered to remote side, and not manifest itself locally.

2021-09-08 20:26:07.881 ERROR 80757 --- [or-http-epoll-2] reactor.core.publisher.Operators         : Operator called default onErrorDropped

io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1
        at io.netty.util.internal.ReferenceCountUpdater.toLiveRealRefCnt(ReferenceCountUpdater.java:74)
        at io.netty.util.internal.ReferenceCountUpdater.release(ReferenceCountUpdater.java:138)
        at io.netty.util.AbstractReferenceCounted.release(AbstractReferenceCounted.java:76)
        at org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.lambda$handleAndReply$5(MessagingRSocket.java:176)

2021-09-08 20:26:07.882 ERROR 80757 --- [or-http-epoll-2] reactor.core.publisher.Operators         : Operator called default onErrorDropped

java.util.concurrent.CancellationException: Outbound has terminated with an error
        at io.rsocket.core.RequestChannelResponderSubscriber.onError(RequestChannelResponderSubscriber.java:765)
        at reactor.core.publisher.FluxSwitchOnFirst$SwitchOnFirstControlSubscriber.onError(FluxSwitchOnFirst.java:877)

2021-09-08 20:26:07.874 ERROR 80757 --- [or-http-epoll-2] o.s.m.h.i.reactive.InvocableHelper       : No exception handling method

java.lang.RuntimeException: plain error
        at example.service.Service.channel(Service.java:16)

Reproducer with 2.5.4

./server.sh (contains above stacktraces), channel_error.sh

[tukez]

I'm running into the same bug. Here is a full stack trace.

2022-10-17 12:26:19.947 [reactor-http-epoll-2] ERROR reactor.core.publisher.Operators - Operator called default onErrorDropped
io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1
	at io.netty.util.internal.ReferenceCountUpdater.toLiveRealRefCnt(ReferenceCountUpdater.java:83)
	at io.netty.util.internal.ReferenceCountUpdater.release(ReferenceCountUpdater.java:147)
	at io.netty.util.AbstractReferenceCounted.release(AbstractReferenceCounted.java:76)
	at org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.lambda$handleAndReply$5(MessagingRSocket.java:176)
	at reactor.core.publisher.FluxDoFinally$DoFinallySubscriber.runFinally(FluxDoFinally.java:146)
	at reactor.core.publisher.FluxDoFinally$DoFinallySubscriber.cancel(FluxDoFinally.java:135)
	at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.cancel(FluxOnAssembly.java:654)
	at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.cancel(MonoIgnoreElements.java:104)
	at reactor.core.publisher.FluxConcatArray$ConcatArraySubscriber.cancel(FluxConcatArray.java:286)
	at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.cancel(FluxOnAssembly.java:654)
	at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.cancel(FluxContextWrite.java:141)
	at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.cancel(FluxOnAssembly.java:654)
	at reactor.core.publisher.Operators.terminate(Operators.java:1240)
	at reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.cancel(MonoFlatMapMany.java:131)
	at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.cancel(FluxOnAssembly.java:654)
	at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.cancel(FluxOnAssembly.java:654)
	at io.rsocket.core.RequestStreamResponderSubscriber.handleCancel(RequestStreamResponderSubscriber.java:300)
	at io.rsocket.core.RSocketResponder.cleanUpSendingSubscriptions(RSocketResponder.java:193)
	at io.rsocket.core.RSocketResponder.doOnDispose(RSocketResponder.java:172)
	at io.rsocket.core.RSocketResponder.tryTerminate(RSocketResponder.java:106)
	at io.rsocket.core.RSocketResponder.tryTerminateOnConnectionClose(RSocketResponder.java:99)
	at reactor.core.publisher.LambdaMonoSubscriber.onComplete(LambdaMonoSubscriber.java:135)
	at org.springframework.security.config.annotation.web.configuration.SecurityReactorContextConfiguration$SecurityReactorContextSubscriber.onComplete(SecurityReactorContextConfiguration.java:176)
	at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.onComplete(FluxOnAssembly.java:549)
	at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.onComplete(FluxOnAssembly.java:549)
	at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.onComplete(FluxOnAssembly.java:549)
	at reactor.core.publisher.SinkEmptyMulticast$VoidInner.complete(SinkEmptyMulticast.java:238)
	at reactor.core.publisher.SinkEmptyMulticast.tryEmitEmpty(SinkEmptyMulticast.java:70)
	at reactor.core.publisher.SinkEmptySerialized.tryEmitEmpty(SinkEmptySerialized.java:46)
	at io.rsocket.internal.BaseDuplexConnection.dispose(BaseDuplexConnection.java:51)
	at io.rsocket.transport.netty.WebsocketDuplexConnection.lambda$new$0(WebsocketDuplexConnection.java:54)
	at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:578)
	at io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:571)
	at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:550)
	at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:491)
	at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:616)
	at io.netty.util.concurrent.DefaultPromise.setSuccess0(DefaultPromise.java:605)
	at io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:104)
	at io.netty.channel.DefaultChannelPromise.trySuccess(DefaultChannelPromise.java:84)
	at io.netty.channel.AbstractChannel$CloseFuture.setClosed(AbstractChannel.java:1164)
	at io.netty.channel.AbstractChannel$AbstractUnsafe.doClose0(AbstractChannel.java:755)
	at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:731)
	at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:620)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.close(DefaultChannelPipeline.java:1352)
	at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:622)
	at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:606)
	at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:472)
	at io.netty.channel.DefaultChannelPipeline.close(DefaultChannelPipeline.java:957)
	at io.netty.channel.AbstractChannel.close(AbstractChannel.java:244)
	at reactor.netty.DisposableChannel.dispose(DisposableChannel.java:72)
	at reactor.netty.channel.ChannelOperations.dispose(ChannelOperations.java:203)
	at reactor.netty.transport.ServerTransport$ChildObserver.onStateChange(ServerTransport.java:474)
	at reactor.netty.channel.ChannelOperations.terminate(ChannelOperations.java:478)
	at reactor.netty.http.server.WebsocketServerOperations.lambda$onInboundNext$1(WebsocketServerOperations.java:154)
	at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:578)
	at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:552)
	at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:491)
	at io.netty.util.concurrent.DefaultPromise.addListener(DefaultPromise.java:184)
	at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:95)
	at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:30)
	at reactor.netty.http.server.WebsocketServerOperations.sendCloseNow(WebsocketServerOperations.java:262)
	at reactor.netty.http.server.WebsocketServerOperations.sendCloseNow(WebsocketServerOperations.java:245)
	at reactor.netty.http.server.WebsocketServerOperations.onInboundNext(WebsocketServerOperations.java:154)
	at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:93)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.rsocket.transport.netty.server.BaseWebsocketServerTransport$PongHandler.channelRead(BaseWebsocketServerTransport.java:60)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:800)
	at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:487)
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:385)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:833)

[snicoll]

@mostroverkhov in which order this needs to be invoked? The server doesn't start because it expects to bind something on port 7000:

Caused by: reactor.netty.ChannelBindException: Failed to bind on [0.0.0.0:7000]
        Suppressed: java.lang.Exception: #block terminated with an error

[spring-projects-issues]

If you would like us to look at this issue, please provide the requested information. If the information is not provided within the next 7 days this issue will be closed.

[spring-projects-issues]

Closing due to lack of requested feedback. If you would like us to look at this issue, please provide the requested information and we will re-open the issue.