ResponseSpec.bodyToMono subscribes twice in case of empty body [SPR-17653]

[spring-projects-issues]

Pavel Kryl opened SPR-17653 and commented

I think that I have just discovered a bug in spring web client. I am writing a reactive kotlin server, retrieving data from microservices, so I have a WebClient and when the response arrives, I am doing:

webClientResponseSpec.retrieve().bodyToMono(MyBean::class.java)

If the server response is empty however (without set Content-Length header), this fails with:

java.lang.IllegalStateException: Only one connection receive subscriber allowed.java.lang.IllegalStateException: Only one connection receive subscriber allowed.
at reactor.netty.channel.FluxReceive.startReceiver(FluxReceive.java:277) [reactor-netty-0.8.2.RELEASE.jar:0.8.2.RELEASE]
at reactor.netty.channel.FluxReceive.subscribe(FluxReceive.java:127) [reactor-netty-0.8.2.RELEASE.jar:0.8.2.RELEASE]
at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at reactor.core.publisher.FluxLift.subscribe(FluxLift.java:46) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at reactor.netty.ByteBufFlux.subscribe(ByteBufFlux.java:290) [reactor-netty-0.8.2.RELEASE.jar:0.8.2.RELEASE]
at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at reactor.core.publisher.FluxLift.subscribe(FluxLift.java:46) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at reactor.core.publisher.FluxMap.subscribe(FluxMap.java:62) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at reactor.core.publisher.FluxLift.subscribe(FluxLift.java:46) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at reactor.core.publisher.Flux.subscribe(Flux.java:7734) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at reactor.core.publisher.FluxConcatArray$ConcatArraySubscriber.onComplete(FluxConcatArray.java:207) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onComplete(ScopePassingSpanSubscriber.java:90) [spring-cloud-sleuth-core-2.0.1.RELEASE.jar:2.0.1.RELEASE]
at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:81) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onComplete(ScopePassingSpanSubscriber.java:90) [spring-cloud-sleuth-core-2.0.1.RELEASE.jar:2.0.1.RELEASE]
at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onComplete(MonoIgnoreElements.java:81) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onComplete(ScopePassingSpanSubscriber.java:90) [spring-cloud-sleuth-core-2.0.1.RELEASE.jar:2.0.1.RELEASE]
at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onComplete(Operators.java:1713) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onComplete(ScopePassingSpanSubscriber.java:90) [spring-cloud-sleuth-core-2.0.1.RELEASE.jar:2.0.1.RELEASE]
at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onComplete(ScopePassingSpanSubscriber.java:90) [spring-cloud-sleuth-core-2.0.1.RELEASE.jar:2.0.1.RELEASE]
at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber.onComplete(ScopePassingSpanSubscriber.java:90) [spring-cloud-sleuth-core-2.0.1.RELEASE.jar:2.0.1.RELEASE]
at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136) [reactor-core-3.2.2.RELEASE.jar:3.2.2.RELEASE]
at reactor.netty.channel.FluxReceive.terminateReceiver(FluxReceive.java:378) [reactor-netty-0.8.2.RELEASE.jar:0.8.2.RELEASE]
at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:202) [reactor-netty-0.8.2.RELEASE.jar:0.8.2.RELEASE]
at reactor.netty.channel.FluxReceive.onInboundComplete(FluxReceive.java:343) [reactor-netty-0.8.2.RELEASE.jar:0.8.2.RELEASE]
at reactor.netty.channel.ChannelOperations.onInboundComplete(ChannelOperations.java:325) [reactor-netty-0.8.2.RELEASE.jar:0.8.2.RELEASE]
at reactor.netty.channel.ChannelOperations.terminate(ChannelOperations.java:372) [reactor-netty-0.8.2.RELEASE.jar:0.8.2.RELEASE]
at reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:522) [reactor-netty-0.8.2.RELEASE.jar:0.8.2.RELEASE]
at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:141) [reactor-netty-0.8.2.RELEASE.jar:0.8.2.RELEASE]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:102) [netty-codec-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:438) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310) [netty-codec-4.1.29.Final.jar:4.1.29.Final]
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284) [netty-codec-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:253) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1434) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:965) [netty-transport-4.1.29.Final.jar:4.1.29.Final]
at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:808) [netty-transport-native-epoll-4.1.29.Final-linux-x86_64.jar:4.1.29.Final]
at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:410) [netty-transport-native-epoll-4.1.29.Final-linux-x86_64.jar:4.1.29.Final]
at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:310) [netty-transport-native-epoll-4.1.29.Final-linux-x86_64.jar:4.1.29.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:884) [netty-common-4.1.29.Final.jar:4.1.29.Final]
at java.lang.Thread.run(Thread.java:834) [?:?]

 

In such a wrong case (when body is expected, but none arrives) I would either presume a more readable exception/error signal or null emission or complete/empty signal. Any of these three options would be better than this IllegalStateException.

Please note, that the error does not occur if I parse to String::class.java.

---

Affects: 5.1.3

[poutsma]

I am not seeing a single Spring Framework line in that stack trace, only Spring Cloud Sleuth (and Reactor). So I am guessing that Sleuth is a bit too eager when subscribing.

Pinging @marcingrzejszczak

[rstoyanchev]

This is a duplicate of #22096.