WebClient .retrieve() + bodyToMono() causes "Only one connection receive subscriber allowed" on 4xx and 5xx [SPR-17615]

[spring-projects-issues]

Vladislav Kisel opened SPR-17615 and commented

On 4xx and 5xx status, using WebClient "retrieve" + "bodyTo..." causes swallowed exception:

java.lang.IllegalStateException: Only one connection receive subscriber allowed. 

Example (kotlin):

WebClient.builder().build().get()
        .uri("https://jira.atlassian.com/rest/api/latest/issue/QWERTY-123")
        .retrieve()
        .bodyToMono(String::class.java)
        .test()
        .expectError(WebClientResponseException::class.java)
        .verify()

 

Probably exception handler subscribes to the connection itself so "bodyTo..." causes this exception. 

Anyway, WebClientResponseException is being correctly thrown and IllegalStateException is just swallowed and printed to the output

---

Affects: 5.1.3

Issue Links:

• WebClient logs "Only one connection receive subscriber allowed" when response status is an error [SPR-17564] #22096 WebClient logs "Only one connection receive subscriber allowed" when response status is an error ("duplicates")

[spring-projects-issues]

Brian Clozel commented

What is the test method you're using? Is that a custom Kotlin extension method?

Could you provide us with a small project we can run?

[spring-projects-issues]

Vladislav Kisel commented

This is an extension from reactor-test, the same as:

StepVerifier.create(WebClient.builder().build().get()
        .uri("https://jira.atlassian.com/rest/api/latest/issue/QWERTY-123")
        .retrieve()
        .bodyToMono(String::class.java))
        .expectError(WebClientResponseException::class.java)
        .verify() 

I believe Spring test will illustrate it as well: WebClientIntegrationTests#shouldGetErrorSignalOn404

[spring-projects-issues]

Brian Clozel commented

Thanks for the pointer indeed, I forgot about that one.

I've tried to reproduce this but couldn't. I'm using Spring Boot 2.1.1.RELEASE (with Framework 5.1.3.RELEASE), reactor-netty 0.8.3.RELEASE and Kotlin 1.3.11.

My test is green and I'm getting the following log:

10:55:30.105 [main] DEBUG io.netty.util.internal.logging.InternalLoggerFactory - Using SLF4J as the default logging framework
10:55:30.129 [main] DEBUG io.netty.util.internal.PlatformDependent - Platform: MacOS
10:55:30.131 [main] DEBUG io.netty.util.internal.PlatformDependent0 - -Dio.netty.noUnsafe: false
10:55:30.131 [main] DEBUG io.netty.util.internal.PlatformDependent0 - Java version: 8
10:55:30.132 [main] DEBUG io.netty.util.internal.PlatformDependent0 - sun.misc.Unsafe.theUnsafe: available
10:55:30.132 [main] DEBUG io.netty.util.internal.PlatformDependent0 - sun.misc.Unsafe.copyMemory: available
10:55:30.132 [main] DEBUG io.netty.util.internal.PlatformDependent0 - java.nio.Buffer.address: available
10:55:30.133 [main] DEBUG io.netty.util.internal.PlatformDependent0 - direct buffer constructor: available
10:55:30.133 [main] DEBUG io.netty.util.internal.PlatformDependent0 - java.nio.Bits.unaligned: available, true
10:55:30.133 [main] DEBUG io.netty.util.internal.PlatformDependent0 - jdk.internal.misc.Unsafe.allocateUninitializedArray(int): unavailable prior to Java9
10:55:30.133 [main] DEBUG io.netty.util.internal.PlatformDependent0 - java.nio.DirectByteBuffer.<init>(long, int): available
10:55:30.133 [main] DEBUG io.netty.util.internal.PlatformDependent - sun.misc.Unsafe: available
10:55:30.133 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.tmpdir: /var/folders/72/f_r9mz9x2fg9cnvfb548pqbh0000gn/T (java.io.tmpdir)
10:55:30.134 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.bitMode: 64 (sun.arch.data.model)
10:55:30.134 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.maxDirectMemory: 3817865216 bytes
10:55:30.134 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.uninitializedArrayAllocationThreshold: -1
10:55:30.135 [main] DEBUG io.netty.util.internal.CleanerJava6 - java.nio.ByteBuffer.cleaner(): available
10:55:30.135 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.noPreferDirect: false
10:55:30.146 [main] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.leakDetection.level: simple
10:55:30.146 [main] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.leakDetection.targetRecords: 4
10:55:30.148 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.numHeapArenas: 16
10:55:30.148 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.numDirectArenas: 16
10:55:30.151 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.pageSize: 8192
10:55:30.151 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.maxOrder: 11
10:55:30.151 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.chunkSize: 16777216
10:55:30.151 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.tinyCacheSize: 512
10:55:30.151 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.smallCacheSize: 256
10:55:30.151 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.normalCacheSize: 64
10:55:30.151 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.maxCachedBufferCapacity: 32768
10:55:30.151 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.cacheTrimInterval: 8192
10:55:30.151 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.useCacheForAllThreads: true
10:55:30.158 [main] DEBUG io.netty.util.internal.InternalThreadLocalMap - -Dio.netty.threadLocalMap.stringBuilder.initialSize: 1024
10:55:30.158 [main] DEBUG io.netty.util.internal.InternalThreadLocalMap - -Dio.netty.threadLocalMap.stringBuilder.maxSize: 4096
10:55:30.169 [main] DEBUG io.netty.util.NetUtil - -Djava.net.preferIPv4Stack: false
10:55:30.170 [main] DEBUG io.netty.util.NetUtil - -Djava.net.preferIPv6Addresses: false
10:55:30.171 [main] DEBUG io.netty.util.NetUtil - Loopback interface: lo0 (lo0, 0:0:0:0:0:0:0:1%lo0)
10:55:30.172 [main] DEBUG io.netty.util.NetUtil - Failed to get SOMAXCONN from sysctl and file /proc/sys/net/core/somaxconn. Default: 128
10:55:30.174 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
10:55:30.195 [main] DEBUG reactor.netty.tcp.TcpResources - [http] resources will use the default LoopResources: DefaultLoopResources {prefix=reactor-http, daemon=true, selectCount=8, workerCount=8}
10:55:30.195 [main] DEBUG reactor.netty.tcp.TcpResources - [http] resources will use the default ConnectionProvider: PooledConnectionProvider {name=http, poolFactory=reactor.netty.resources.ConnectionProvider$$Lambda$68/1502335674@5a755cc1}
10:55:30.320 [main] DEBUG reactor.netty.resources.DefaultLoopEpoll - Default Epoll support : false
10:55:30.321 [main] DEBUG reactor.netty.resources.DefaultLoopKQueue - Default KQueue support : false
10:55:30.326 [main] DEBUG io.netty.channel.MultithreadEventLoopGroup - -Dio.netty.eventLoopThreads: 16
10:55:30.362 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.noKeySetOptimization: false
10:55:30.362 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.selectorAutoRebuildThreshold: 512
10:55:30.385 [main] DEBUG io.netty.util.internal.PlatformDependent - org.jctools-core.MpscChunkedArrayQueue: available
10:55:30.701 [main] DEBUG org.springframework.web.reactive.function.client.ExchangeFunctions - [69653e16] HTTP GET https://jira.atlassian.com/rest/api/latest/issue/QWERTY-123
10:55:30.724 [main] DEBUG io.netty.handler.ssl.OpenSsl - netty-tcnative not in the classpath; OpenSslEngine will be unavailable.
10:55:30.940 [main] DEBUG io.netty.handler.ssl.JdkSslContext - Default protocols (JDK): [TLSv1.2, TLSv1.1, TLSv1] 
10:55:30.940 [main] DEBUG io.netty.handler.ssl.JdkSslContext - Default cipher suites (JDK): [TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384, TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256, TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256, TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA, TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA, TLS_RSA_WITH_AES_128_GCM_SHA256, TLS_RSA_WITH_AES_128_CBC_SHA, TLS_RSA_WITH_AES_256_CBC_SHA]
10:55:30.965 [main] DEBUG reactor.netty.resources.PooledConnectionProvider - Creating new client pool [http] for jira.atlassian.com:443
10:55:31.014 [main] DEBUG io.netty.channel.DefaultChannelId - -Dio.netty.processId: 1772 (auto-detected)
10:55:31.018 [main] DEBUG io.netty.channel.DefaultChannelId - -Dio.netty.machineId: ac:de:48:ff:fe:00:11:22 (auto-detected)
10:55:31.043 [main] DEBUG io.netty.buffer.ByteBufUtil - -Dio.netty.allocator.type: pooled
10:55:31.043 [main] DEBUG io.netty.buffer.ByteBufUtil - -Dio.netty.threadLocalDirectBufferSize: 0
10:55:31.043 [main] DEBUG io.netty.buffer.ByteBufUtil - -Dio.netty.maxThreadLocalCharBufferSize: 16384
10:55:31.062 [reactor-http-nio-4] DEBUG reactor.netty.resources.PooledConnectionProvider - [id: 0x0702771a] Created new pooled channel, now 0 active connections and 1 inactive connections
10:55:31.094 [reactor-http-nio-4] DEBUG io.netty.buffer.AbstractByteBuf - -Dio.netty.buffer.checkAccessible: true
10:55:31.094 [reactor-http-nio-4] DEBUG io.netty.buffer.AbstractByteBuf - -Dio.netty.buffer.checkBounds: true
10:55:31.095 [reactor-http-nio-4] DEBUG io.netty.util.ResourceLeakDetectorFactory - Loaded default ResourceLeakDetector: io.netty.util.ResourceLeakDetector@76dd0173
10:55:31.110 [reactor-http-nio-4] DEBUG reactor.netty.tcp.SslProvider - [id: 0x0702771a] SSL enabled using engine SSLEngineImpl and SNI jira.atlassian.com:443
10:55:31.116 [reactor-http-nio-4] DEBUG reactor.netty.channel.BootstrapHandlers - [id: 0x0702771a] Initialized pipeline DefaultChannelPipeline{(reactor.left.sslHandler = io.netty.handler.ssl.SslHandler), (reactor.left.sslReader = reactor.netty.tcp.SslProvider$SslReadHandler), (BootstrapHandlers$BootstrapInitializerHandler#0 = reactor.netty.channel.BootstrapHandlers$BootstrapInitializerHandler), (SimpleChannelPool$1#0 = io.netty.channel.pool.SimpleChannelPool$1), (reactor.left.httpCodec = io.netty.handler.codec.http.HttpClientCodec), (reactor.left.decompressor = io.netty.handler.codec.http.HttpContentDecompressor), (reactor.right.reactiveBridge = reactor.netty.channel.ChannelOperationsHandler)}
10:55:31.433 [reactor-http-nio-4] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.maxCapacityPerThread: 4096
10:55:31.433 [reactor-http-nio-4] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.maxSharedCapacityFactor: 2
10:55:31.433 [reactor-http-nio-4] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.linkCapacity: 16
10:55:31.433 [reactor-http-nio-4] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.ratio: 8
10:55:31.443 [reactor-http-nio-4] DEBUG reactor.netty.resources.PooledConnectionProvider - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] Registering pool release on close event for channel
10:55:31.443 [reactor-http-nio-4] DEBUG reactor.netty.resources.PooledConnectionProvider - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] Channel connected, now 1 active connections and 0 inactive connections
10:55:31.963 [reactor-http-nio-4] DEBUG io.netty.handler.ssl.SslHandler - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] HANDSHAKEN: TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256
10:55:31.964 [reactor-http-nio-4] DEBUG reactor.netty.resources.PooledConnectionProvider - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] onStateChange(PooledConnection{channel=[id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443]}, [connected])
10:55:31.973 [reactor-http-nio-4] DEBUG reactor.netty.resources.PooledConnectionProvider - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] onStateChange(GET{uri=/, connection=PooledConnection{channel=[id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443]}}, [configured])
10:55:31.975 [reactor-http-nio-4] DEBUG reactor.netty.http.client.HttpClientConnect - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] Handler is being applied: {uri=https://jira.atlassian.com/rest/api/latest/issue/QWERTY-123, method=GET}
10:55:31.983 [reactor-http-nio-4] DEBUG reactor.netty.channel.ChannelOperationsHandler - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] Writing object DefaultHttpRequest(decodeResult: success, version: HTTP/1.1)
GET /rest/api/latest/issue/QWERTY-123 HTTP/1.1
accept-encoding: gzip
user-agent: ReactorNetty/0.8.3.RELEASE
host: jira.atlassian.com
accept: */*
10:55:31.995 [reactor-http-nio-4] DEBUG reactor.netty.channel.ChannelOperationsHandler - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] Writing object EmptyLastHttpContent
10:55:31.996 [reactor-http-nio-4] DEBUG reactor.netty.resources.PooledConnectionProvider - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] onStateChange(GET{uri=/rest/api/latest/issue/QWERTY-123, connection=PooledConnection{channel=[id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443]}}, [request_sent])
10:55:32.249 [reactor-http-nio-4] DEBUG reactor.netty.http.client.HttpClientOperations - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] Received response (auto-read:false) : [Date=Fri, 21 Dec 2018 09:55:32 GMT, Content-Type=application/json;charset=UTF-8, Transfer-Encoding=chunked, Connection=keep-alive, Cache-Control=no-cache, no-store, no-transform, Content-Security-Policy=frame-ancestors 'self', Set-Cookie=atlassian.xsrf.token=AKVY-YUFR-9LM7-97AB_50eabffb44cf6dd6e55913c2d9721e3a0d90aac0_lout; Path=/; Secure, X-ANODEID=i-02f4367aef9e46b81:10.104.240.132, X-AREQUESTID=595x4937766x1, X-ASEN=TestSEN, X-AUSERNAME=anonymous, X-Content-Type-Options=nosniff, X-Frame-Options=SAMEORIGIN, X-XSS-Protection=1; mode=block]
10:55:32.249 [reactor-http-nio-4] DEBUG reactor.netty.resources.PooledConnectionProvider - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] onStateChange(GET{uri=/rest/api/latest/issue/QWERTY-123, connection=PooledConnection{channel=[id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443]}}, [response_received])
10:55:32.253 [reactor-http-nio-4] DEBUG org.springframework.web.reactive.function.client.ExchangeFunctions - [69653e16] Response 404 NOT_FOUND
10:55:32.300 [reactor-http-nio-4] DEBUG reactor.netty.channel.FluxReceive - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] Subscribing inbound receiver [pending: 0, cancelled:false, inboundDone: false]
10:55:32.316 [reactor-http-nio-4] DEBUG reactor.netty.http.client.HttpClientOperations - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] Received last HTTP packet
10:55:32.317 [reactor-http-nio-4] DEBUG reactor.netty.resources.PooledConnectionProvider - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] onStateChange(GET{uri=/rest/api/latest/issue/QWERTY-123, connection=PooledConnection{channel=[id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443]}}, [disconnecting])
10:55:32.317 [reactor-http-nio-4] DEBUG reactor.netty.resources.PooledConnectionProvider - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] Releasing channel
10:55:32.317 [reactor-http-nio-4] DEBUG reactor.netty.resources.PooledConnectionProvider - [id: 0x0702771a, L:/192.168.150.90:51174 - R:jira.atlassian.com/13.52.5.95:443] Channel cleaned, now 0 active connections and 1 inactive connections

Do you have a sample project that I can run to reproduce this?

[spring-projects-issues]

Vladislav Kisel commented

Sure, https://github.com/encircled/reactive-jira-client  - MockTest class

 

P.S. hotspot 1.8_152, MacOS

[spring-projects-issues]

Brian Clozel commented

Ignore me Vladislav Kisel, this is a duplicate of #22096 which is already fixed for the next Spring Framework version.

Thanks!

[spring-projects-issues]

Vladislav Kisel commented

Oh, great. Sorry have not found that one myself

Thanks