WebClient logs "Only one connection receive subscriber allowed" when response status is an error [SPR-17564]

[spring-projects-issues]

Fethullah Misir opened SPR-17564 and commented

WebClient throws an IllegalStateException with:

Only one connection receive subscriber allowed

 when the server response status is 4xx / 5xx.

The issue is not reproducible with Spring Boot 2.1.0.RELEASE / WebFlux 5.1.2.RELEASE. 

 

I  reproduced the issue with the test below:

@Test
public void testRetrieveDoesThrowMultipleSubscriberError() {
   stubFor( get( urlEqualTo( "/notFound" ) )
         .willReturn( aResponse().withHeader( HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE )
                                 .withBody( "{}" )
                                 .withStatus( HttpStatus.NOT_FOUND_404 ) ) );

   StepVerifier.create( webClient.get()
                                 .uri( "/notFound" )
                                 .accept( MediaType.APPLICATION_JSON )
                                 .retrieve()

                                 .bodyToMono( String.class ) )
               .expectError( WebClientException.class )
               .verify();
}
 

The test pasess, however the below stack trace is logged: 

Caused by: java.lang.IllegalStateException: Only one connection receive subscriber allowed.
        ... 143 common frames omitted
        Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Assembly trace from producer [reactor.core.publisher.FluxMap] :
        reactor.core.publisher.Flux.map(Flux.java:5655)
        reactor.netty.ByteBufFlux.fromInbound(ByteBufFlux.java:70)
        reactor.netty.channel.ChannelOperations.receive(ChannelOperations.java:225)
        org.springframework.http.client.reactive.ReactorClientHttpResponse.getBody(ReactorClientHttpResponse.java:64)
        org.springframework.web.reactive.function.BodyExtractors.consumeAndCancel(BodyExtractors.java:268)
        org.springframework.web.reactive.function.BodyExtractors.lambda$skipBodyAsMono$21(BodyExtractors.java:264)
.....
.....
        io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:500)
        io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:462)
        io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:897)
Error has been observed by the following operator(s):
        |_      Flux.map ? reactor.netty.ByteBufFlux.fromInbound(ByteBufFlux.java:70)
        |_      Flux.doOnSubscribe ? org.springframework.http.client.reactive.ReactorClientHttpResponse.getBody(ReactorClientHttpResponse.java:65)
 

 

I attached a sample, to reproduce the issue.

It seams that the issue is related to the default status handler. When registering a custom one, the stacktrace is not logged and it works as expected.

StepVerifier.create( webClient.get()
                              .uri( "/notFound" )
                              .accept( MediaType.APPLICATION_JSON )
                              .retrieve()
                              .onStatus( status -> true, response -> Mono.error(IllegalStateException::new ) )
                              .bodyToMono( String.class ) )
            .expectError( IllegalStateException.class )
            .verify(); 

 

 

 

 

---

Affects: 5.1.3

Attachments:

• webclient-multi-subscriber.zip (3.42 kB)

Issue Links:

• WebClient .retrieve() + bodyToMono() causes "Only one connection receive subscriber allowed" on 4xx and 5xx [SPR-17615] #22147 WebClient .retrieve() + bodyToMono() causes "Only one connection receive subscriber allowed" on 4xx and 5xx ("is duplicated by")
• WebClient throws "Only one connection receive subscriber allowed." when response generates a WebClientResponseException [SPR-17576] #22108 WebClient throws "Only one connection receive subscriber allowed." when response generates a WebClientResponseException ("is duplicated by")
• The onstatus method of webclient causes a memory leak. [SPR-17473] #22005 The onstatus method of webclient causes a memory leak.

Referenced from: commits 7a5f8e0

[spring-projects-issues]

Rossen Stoyanchev commented

This is as a result of the work for #22005. We to ensure the body is consumed, and that may result in attempting to consume it twice, and hence the error. We then ignore the error, so in that sense it doesn't do any harm, but the error is logged by Reactor Core.

[spring-projects-issues]

Rossen Stoyanchev commented

The same issue was also reported in Reactor Netty #540.

[spring-projects-issues]

Rossen Stoyanchev commented

There is a fix available in master.

[spring-projects-issues]

Fethullah Misir commented

 

In an application on that i'am working on, the issue does cause the unit tests to fail. 

The use-case is to do a fallback using the same webclient when the first requests results in an error.

I can reproduce the same issue with the below test: 

 

@Test
public void testRetrieveDoesThrowMultipleSubscriberError() {
   //Stub 404
   stubFor( get( urlEqualTo( "/notFound" ) )
         .willReturn( aResponse().withHeader( HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE )
                                 .withBody( "{}" )
                                 .withStatus( HttpStatus.NOT_FOUND_404 ) ) );
   //Stub 200
   stubFor( get( urlEqualTo( "/ok" ) )
         .willReturn( aResponse().withHeader( HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE )
                                 .withBody( "{}" )
                                 .withStatus( HttpStatus.OK_200 ) ) );

   StepVerifier.create( webClient.get()
                                 .uri( "/notFound" )
                                 .accept( MediaType.APPLICATION_JSON )
                                 .retrieve()
                                 .bodyToMono( String.class )
                                 // ---> fallback
                                 .onErrorResume( error -> webClient.get().uri( "/ok" )
                                                                   .accept( MediaType.APPLICATION_JSON )
                                                                   .retrieve()
                                                                   .bodyToMono( String.class ) ) )
               .expectNext( "{}" )
               .verifyComplete();
} 

//results in: 

java.lang.AssertionError: expectation "expectNext({})" failed (expected: onNext({}); actual: onError(reactor.core.Exceptions$BubblingException: java.lang.IllegalStateException: Only one connection receive subscriber allowed.))
 at reactor.test.ErrorFormatter.assertionError(ErrorFormatter.java:105) at reactor.test.ErrorFormatter.failPrefix(ErrorFormatter.java:94)

I think the reuse of the WebClient causes the error. When replacing the fallback with 

.onErrorResume( error -> Mono.just("{}" )

the test passes successfully (the reactor error is still logged).

 

I also tried your fix with the latest snapshot build.

[INFO] +- org.springframework:spring-webflux:jar:5.1.4.BUILD-SNAPSHOT:compile
[INFO] |  +- io.projectreactor:reactor-core:jar:3.2.3.RELEASE:compile
[INFO] |  |  \- org.reactivestreams:reactive-streams:jar:1.0.2:compile
 

The excpetion is still logged and additionally the test does not terminate anymore.

(I ran the initial test without the fallback)

reactor-http-nio-42018-12-05 09:07:50 ERROR reactor.core.publisher.Operators - Operator called default onErrorDropped
java.lang.IllegalStateException: Only one connection receive subscriber allowed. 

 

 

[spring-projects-issues]

Fethullah Misir commented

Forget about my last comment. I missed to update the spring-web artifact to the latest build. 

The reactor error is gone and my tests are passing now. Awesome, thank you! 

 

 

[spring-projects-issues]

Rossen Stoyanchev commented

Thanks for checking! I did verify the fix with your test, and with the sample from the Reactor Netty issue, but always good to know it works in your environment.