RSocket and WebClient with Security refCount: 0

[ghahramani]

I first created this issue in the spring boot repo spring-projects/spring-boot#24833 but after digging more and creating a new project to see where the problem is, I found out this is about spring-security

Hey everyone, I am facing a strange problem, cannot figure out what is going on. It is related to RSocket and WebClient together.
So what I am trying to achieve is, Creating a stream of data from retrieving the data via RSocket and stream that data to an endpoint via WebClient. So This is my code.

Spring Boot: 2.4.1
Dependency Management: 1.0.10.RELEASE
Spring Cloud Dependencies: 2020.0.0

RSocket:

@MessageMapping(Api.Message.UPLOAD_FILE)
    fun uploadFile(
        @Header(Api.Header.MIME_FILE_NAME)
        fileName: String,

        @Validated
        @Payload
        data: Flux<DataBuffer>
    ) = service.upload(fileName, data, "test")

WebClient:

fun upload(fileName: String, data: Flux<DataBuffer>, token: String? = null) =
        webClient
            .post()
            .uri(
                UriComponentsBuilder
                    .fromUriString("https://something.com")
                    .queryParam("filename", fileName)
                    .also { builder -> token?.let { value -> builder.queryParam(token, value) } }
                    .build(true)
                    .toUri()
            )
            .header(HttpHeaders.AUTHORIZATION, "Bearer Test")
            .contentType(MediaType.asMediaType(MimeType.valueOf("application/binary")))
            .body(data, DataBuffer::class.java)
            .exchangeToMono { response ->
                if (response.statusCode().isError) {
                    handleResponseError(response)
                } else {
                    response.bodyToMono()
                }
            }

But the problem is, it seems when I use the Publisher that came from RSocket, the body that wants to send to WebClient will be empty and zero-length and I get the error below:

org.springframework.web.reactive.function.client.WebClientRequestException: refCnt: 0; nested exception is io.netty.util.IllegalReferenceCountException: refCnt: 0
cause -> IllegalReferenceCountException: refCnt: 0

But if I change the body to any other Publisher, it works with no problem

So changing this:

.body(data, DataBuffer::class.java)

To this or anything else:

.body(Flux.just("Test","Test2"), String::class.java)

Update 1:
I found a strange thing, when the rsocket security gets involved I get that error, but if I remove the RSocket dependencies and remove the code below, it started working

@EnableRSocketSecurity
class RSocketSecurityConfiguration {

    private val logger: Logger = LoggerFactory.getLogger(javaClass)

    @Bean
    fun messageHandler(strategies: RSocketStrategies): RSocketMessageHandler {
        val handler = RSocketMessageHandler()
        handler.argumentResolverConfigurer.addCustomResolver(AuthenticationPrincipalArgumentResolver())
        handler.rSocketStrategies = strategies
        logger.info("Configuring RSocket handler")
        return handler
    }

    @Bean
    fun authorization(
        security: RSocketSecurity,
        authenticationManager: ReactiveAuthenticationManager
    ): PayloadSocketAcceptorInterceptor = security
        .authorizePayload { spec: AuthorizePayloadsSpec ->
            logger.info("Configuring RSocket authorization and jwt")
            spec
                .route("api.**").authenticated()
                .anyExchange().permitAll()
        }
        .jwt { customizer -> customizer.authenticationManager(authenticationManager) }
        .build()

}

How to reproduce it:

1. Clone https://github.com/ghahramani/spring-security-rsocket-bug
2. run ./gradlew bootRun
3. There are two endpoins /working and /not-working you can call them either via curl (curl http://localhost:9999/not-working or curl http://localhost:9999/working) or just browse those URLs in your browser

You should see the error like this : (/not-working endpoint)

[
     {
          "fileName": "not-passed",
          "body": "refCnt: 0; nested exception is io.netty.util.IllegalReferenceCountException: refCnt: 0"
    }
]

and see the data below for /working endpoint:

[
    {
         "fileName": "TestFileName",
         "body": "TestTest"
     }
]

This is the stacktrace with reactive log

2021-01-14 23:07:45.964  INFO 83120 --- [           main] uration$$EnhancerBySpringCGLIB$$ed6308fa : Configuring RSocket authorization and jwt
2021-01-14 23:07:45.972  INFO 83120 --- [           main] uration$$EnhancerBySpringCGLIB$$ed6308fa : Configuring RSocket handler
2021-01-14 23:07:46.258  INFO 83120 --- [           main] o.s.b.web.embedded.netty.NettyWebServer  : Netty started on port(s): 9999
2021-01-14 23:07:46.268  INFO 83120 --- [           main] o.s.b.rsocket.netty.NettyRSocketServer   : Netty RSocket started on port(s): 9998
2021-01-14 23:07:46.280  INFO 83120 --- [           main] com.example.demo.DemoApplicationKt       : Started DemoApplicationKt in 1.688 seconds (JVM running for 2.277)
2021-01-14 23:07:49.736  INFO 83120 --- [ctor-http-nio-6] reactor.Flux.SwitchIfEmpty.1             : onSubscribe(FluxSwitchIfEmpty.SwitchIfEmptySubscriber)
2021-01-14 23:07:49.747  INFO 83120 --- [ctor-http-nio-6] reactor.Flux.SwitchIfEmpty.1             : request(128)
2021-01-14 23:07:49.749  INFO 83120 --- [ctor-http-nio-6] reactor.Flux.SwitchIfEmpty.1             : onNext([116, 101, 115, 116])
2021-01-14 23:07:49.755 ERROR 83120 --- [ctor-http-nio-6] reactor.Flux.SwitchIfEmpty.1             : onError(io.netty.util.IllegalReferenceCountException: refCnt: 0)
2021-01-14 23:07:49.759 ERROR 83120 --- [ctor-http-nio-6] reactor.Flux.SwitchIfEmpty.1             : 

io.netty.util.IllegalReferenceCountException: refCnt: 0
	at io.rsocket.util.ByteBufPayload.ensureAccessible(ByteBufPayload.java:208) ~[rsocket-core-1.1.0.jar:na]
	at io.rsocket.util.ByteBufPayload.sliceData(ByteBufPayload.java:155) ~[rsocket-core-1.1.0.jar:na]
	at org.springframework.messaging.rsocket.PayloadUtils.retainDataAndReleasePayload(PayloadUtils.java:54) ~[spring-messaging-5.3.2.jar:5.3.2]
	at org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.retainDataAndReleasePayload(MessagingRSocket.java:184) ~[spring-messaging-5.3.2.jar:5.3.2]
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Operators$ConditionalSubscriberAdapter.onNext(Operators.java:2429) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxSwitchOnFirst$AbstractSwitchOnFirstMain.onNext(FluxSwitchOnFirst.java:161) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxConcatArray$ConcatArraySubscriber.onNext(FluxConcatArray.java:177) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:711) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.drain(FluxFlatMap.java:587) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.request(FluxFlatMap.java:346) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.request(Operators.java:2118) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxSwitchOnFirst$AbstractSwitchOnFirstMain.request(FluxSwitchOnFirst.java:261) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:162) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxPeek$PeekSubscriber.request(FluxPeek.java:137) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:162) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:162) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:2154) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onSubscribe(FluxOnErrorResume.java:74) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:92) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:92) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxPeek$PeekSubscriber.onSubscribe(FluxPeek.java:170) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:92) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxSwitchOnFirst$SwitchOnFirstConditionalMain.subscribe(FluxSwitchOnFirst.java:391) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Flux.subscribe(Flux.java:8147) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.netty.channel.MonoSendMany.subscribe(MonoSendMany.java:102) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:154) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Mono.subscribe(Mono.java:4046) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxConcatIterable$ConcatIterableSubscriber.onComplete(FluxConcatIterable.java:147) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxConcatIterable.subscribe(FluxConcatIterable.java:60) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.MonoFromFluxOperator.subscribe(MonoFromFluxOperator.java:81) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.netty.http.client.HttpClientConnect$HttpIOHandlerObserver.onStateChange(HttpClientConnect.java:399) ~[reactor-netty-http-1.0.2.jar:1.0.2]
	at reactor.netty.ReactorNetty$CompositeConnectionObserver.onStateChange(ReactorNetty.java:612) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.netty.resources.DefaultPooledConnectionProvider$DisposableAcquire.onStateChange(DefaultPooledConnectionProvider.java:195) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.netty.resources.DefaultPooledConnectionProvider$PooledConnection.onStateChange(DefaultPooledConnectionProvider.java:466) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.netty.channel.ChannelOperationsHandler.channelActive(ChannelOperationsHandler.java:63) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:230) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:216) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelActive(AbstractChannelHandlerContext.java:209) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelActive(CombinedChannelDuplexHandler.java:412) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.ChannelInboundHandlerAdapter.channelActive(ChannelInboundHandlerAdapter.java:69) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.CombinedChannelDuplexHandler.channelActive(CombinedChannelDuplexHandler.java:211) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:230) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:216) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelActive(AbstractChannelHandlerContext.java:209) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelActive(DefaultChannelPipeline.java:1398) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:230) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:216) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.DefaultChannelPipeline.fireChannelActive(DefaultChannelPipeline.java:895) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.fulfillConnectPromise(AbstractNioChannel.java:305) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:335) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:707) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
	at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]

2021-01-14 23:07:49.762  WARN 83120 --- [ctor-http-nio-6] r.netty.http.client.HttpClientConnect    : [id: 0xaf1eaaa3, L:/127.0.0.1:64411 - R:localhost/127.0.0.1:9999] The connection observed an error

io.netty.util.IllegalReferenceCountException: refCnt: 0
	at io.rsocket.util.ByteBufPayload.ensureAccessible(ByteBufPayload.java:208) ~[rsocket-core-1.1.0.jar:na]
	at io.rsocket.util.ByteBufPayload.sliceData(ByteBufPayload.java:155) ~[rsocket-core-1.1.0.jar:na]
	at org.springframework.messaging.rsocket.PayloadUtils.retainDataAndReleasePayload(PayloadUtils.java:54) ~[spring-messaging-5.3.2.jar:5.3.2]
	at org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.retainDataAndReleasePayload(MessagingRSocket.java:184) ~[spring-messaging-5.3.2.jar:5.3.2]
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Operators$ConditionalSubscriberAdapter.onNext(Operators.java:2429) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxSwitchOnFirst$AbstractSwitchOnFirstMain.onNext(FluxSwitchOnFirst.java:161) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxConcatArray$ConcatArraySubscriber.onNext(FluxConcatArray.java:177) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:711) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.drain(FluxFlatMap.java:587) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.request(FluxFlatMap.java:346) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.request(Operators.java:2118) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxSwitchOnFirst$AbstractSwitchOnFirstMain.request(FluxSwitchOnFirst.java:261) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:162) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxPeek$PeekSubscriber.request(FluxPeek.java:137) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:162) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:162) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:2154) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onSubscribe(FluxOnErrorResume.java:74) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:92) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:92) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxPeek$PeekSubscriber.onSubscribe(FluxPeek.java:170) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:92) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxSwitchOnFirst$SwitchOnFirstConditionalMain.subscribe(FluxSwitchOnFirst.java:391) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Flux.subscribe(Flux.java:8147) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.netty.channel.MonoSendMany.subscribe(MonoSendMany.java:102) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:154) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Mono.subscribe(Mono.java:4046) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxConcatIterable$ConcatIterableSubscriber.onComplete(FluxConcatIterable.java:147) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxConcatIterable.subscribe(FluxConcatIterable.java:60) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.MonoFromFluxOperator.subscribe(MonoFromFluxOperator.java:81) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.netty.http.client.HttpClientConnect$HttpIOHandlerObserver.onStateChange(HttpClientConnect.java:399) ~[reactor-netty-http-1.0.2.jar:1.0.2]
	at reactor.netty.ReactorNetty$CompositeConnectionObserver.onStateChange(ReactorNetty.java:612) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.netty.resources.DefaultPooledConnectionProvider$DisposableAcquire.onStateChange(DefaultPooledConnectionProvider.java:195) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.netty.resources.DefaultPooledConnectionProvider$PooledConnection.onStateChange(DefaultPooledConnectionProvider.java:466) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.netty.channel.ChannelOperationsHandler.channelActive(ChannelOperationsHandler.java:63) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:230) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:216) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelActive(AbstractChannelHandlerContext.java:209) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelActive(CombinedChannelDuplexHandler.java:412) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.ChannelInboundHandlerAdapter.channelActive(ChannelInboundHandlerAdapter.java:69) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.CombinedChannelDuplexHandler.channelActive(CombinedChannelDuplexHandler.java:211) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:230) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:216) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelActive(AbstractChannelHandlerContext.java:209) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelActive(DefaultChannelPipeline.java:1398) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:230) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:216) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.DefaultChannelPipeline.fireChannelActive(DefaultChannelPipeline.java:895) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.fulfillConnectPromise(AbstractNioChannel.java:305) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:335) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:707) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
	at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]

2021-01-14 23:07:49.770 ERROR 83120 --- [ctor-http-nio-6] o.s.m.h.i.reactive.InvocableHelper       : No exception handling method

org.springframework.web.reactive.function.client.WebClientRequestException: refCnt: 0; nested exception is io.netty.util.IllegalReferenceCountException: refCnt: 0
	at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.lambda$wrapException$9(ExchangeFunctions.java:137) ~[spring-webflux-5.3.2.jar:5.3.2]
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
	|_ checkpoint ⇢ Request to POST http://localhost:9999/api/v2/uploads.json?filename=TestFileName [DefaultWebClient]
Stack trace:
		at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.lambda$wrapException$9(ExchangeFunctions.java:137) ~[spring-webflux-5.3.2.jar:5.3.2]
		at reactor.core.publisher.MonoErrorSupplied.subscribe(MonoErrorSupplied.java:70) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.Mono.subscribe(Mono.java:4046) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:103) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:221) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:221) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:221) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.MonoNext$NextSubscriber.onError(MonoNext.java:93) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onError(MonoFlatMapMany.java:204) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.SerializedSubscriber.onError(SerializedSubscriber.java:124) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.whenError(FluxRetryWhen.java:224) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.FluxRetryWhen$RetryWhenOtherSubscriber.onError(FluxRetryWhen.java:273) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:413) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onNext(FluxConcatMap.java:250) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.EmitterProcessor.drain(EmitterProcessor.java:491) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.EmitterProcessor.tryEmitNext(EmitterProcessor.java:299) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.SinkManySerialized.tryEmitNext(SinkManySerialized.java:97) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onError(FluxRetryWhen.java:189) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.MonoCreate$DefaultMonoSink.error(MonoCreate.java:189) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.netty.http.client.HttpClientConnect$HttpObserver.onUncaughtException(HttpClientConnect.java:359) ~[reactor-netty-http-1.0.2.jar:1.0.2]
		at reactor.netty.ReactorNetty$CompositeConnectionObserver.onUncaughtException(ReactorNetty.java:605) ~[reactor-netty-core-1.0.2.jar:1.0.2]
		at reactor.netty.resources.DefaultPooledConnectionProvider$DisposableAcquire.onUncaughtException(DefaultPooledConnectionProvider.java:213) ~[reactor-netty-core-1.0.2.jar:1.0.2]
		at reactor.netty.resources.DefaultPooledConnectionProvider$PooledConnection.onUncaughtException(DefaultPooledConnectionProvider.java:476) ~[reactor-netty-core-1.0.2.jar:1.0.2]
		at reactor.netty.http.client.HttpClientOperations.onOutboundError(HttpClientOperations.java:562) ~[reactor-netty-http-1.0.2.jar:1.0.2]
		at reactor.netty.channel.ChannelOperations.onError(ChannelOperations.java:242) ~[reactor-netty-core-1.0.2.jar:1.0.2]
		at reactor.core.publisher.MonoIgnoreElements$IgnoreElementsSubscriber.onError(MonoIgnoreElements.java:83) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onError(Operators.java:2023) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.Operators$MonoSubscriber.onError(Operators.java:1831) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.core.publisher.MonoIgnoreThen$ThenAcceptInner.onError(MonoIgnoreThen.java:315) ~[reactor-core-3.4.1.jar:3.4.1]
		at reactor.netty.channel.MonoSendMany$SendManyInner.run(MonoSendMany.java:348) ~[reactor-netty-core-1.0.2.jar:1.0.2]
		at reactor.netty.channel.MonoSendMany$SendManyInner.trySchedule(MonoSendMany.java:419) ~[reactor-netty-core-1.0.2.jar:1.0.2]
		at reactor.netty.channel.MonoSendMany$SendManyInner.trySuccess(MonoSendMany.java:562) ~[reactor-netty-core-1.0.2.jar:1.0.2]
		at reactor.netty.channel.MonoSendMany$SendManyInner.trySuccess(MonoSendMany.java:114) ~[reactor-netty-core-1.0.2.jar:1.0.2]
		at io.netty.util.concurrent.PromiseCombiner.tryPromise(PromiseCombiner.java:170) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.PromiseCombiner.access$600(PromiseCombiner.java:35) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.PromiseCombiner$1.operationComplete0(PromiseCombiner.java:62) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.PromiseCombiner$1.operationComplete(PromiseCombiner.java:44) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:578) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:552) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:491) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:616) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.DefaultPromise.setSuccess0(DefaultPromise.java:605) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:104) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.internal.PromiseNotificationUtil.trySuccess(PromiseNotificationUtil.java:48) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.ChannelOutboundBuffer.safeSuccess(ChannelOutboundBuffer.java:717) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.ChannelOutboundBuffer.remove(ChannelOutboundBuffer.java:272) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.ChannelOutboundBuffer.removeBytes(ChannelOutboundBuffer.java:352) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.socket.nio.NioSocketChannel.doWrite(NioSocketChannel.java:431) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.AbstractChannel$AbstractUnsafe.flush0(AbstractChannel.java:941) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.flush0(AbstractNioChannel.java:354) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.AbstractChannel$AbstractUnsafe.flush(AbstractChannel.java:905) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.DefaultChannelPipeline$HeadContext.flush(DefaultChannelPipeline.java:1372) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:750) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.AbstractChannelHandlerContext.invokeFlush(AbstractChannelHandlerContext.java:742) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:728) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.flush(CombinedChannelDuplexHandler.java:531) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.ChannelOutboundHandlerAdapter.flush(ChannelOutboundHandlerAdapter.java:125) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.CombinedChannelDuplexHandler.flush(CombinedChannelDuplexHandler.java:356) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.AbstractChannelHandlerContext.invokeFlush0(AbstractChannelHandlerContext.java:750) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.AbstractChannelHandlerContext.invokeFlush(AbstractChannelHandlerContext.java:742) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:728) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at reactor.netty.channel.MonoSendMany$SendManyInner$AsyncFlush.run(MonoSendMany.java:758) ~[reactor-netty-core-1.0.2.jar:1.0.2]
		at io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:164) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:497) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
		at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
Caused by: io.netty.util.IllegalReferenceCountException: refCnt: 0
	at io.rsocket.util.ByteBufPayload.ensureAccessible(ByteBufPayload.java:208) ~[rsocket-core-1.1.0.jar:na]
	at io.rsocket.util.ByteBufPayload.sliceData(ByteBufPayload.java:155) ~[rsocket-core-1.1.0.jar:na]
	at org.springframework.messaging.rsocket.PayloadUtils.retainDataAndReleasePayload(PayloadUtils.java:54) ~[spring-messaging-5.3.2.jar:5.3.2]
	at org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.retainDataAndReleasePayload(MessagingRSocket.java:184) ~[spring-messaging-5.3.2.jar:5.3.2]
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Operators$ConditionalSubscriberAdapter.onNext(Operators.java:2429) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxSwitchOnFirst$AbstractSwitchOnFirstMain.onNext(FluxSwitchOnFirst.java:161) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxConcatArray$ConcatArraySubscriber.onNext(FluxConcatArray.java:177) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:711) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.drain(FluxFlatMap.java:587) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.request(FluxFlatMap.java:346) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.request(Operators.java:2118) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxSwitchOnFirst$AbstractSwitchOnFirstMain.request(FluxSwitchOnFirst.java:261) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:162) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxPeek$PeekSubscriber.request(FluxPeek.java:137) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:162) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:162) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:2154) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onSubscribe(FluxOnErrorResume.java:74) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:92) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:92) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxPeek$PeekSubscriber.onSubscribe(FluxPeek.java:170) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:92) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxSwitchOnFirst$SwitchOnFirstConditionalMain.subscribe(FluxSwitchOnFirst.java:391) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Flux.subscribe(Flux.java:8147) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.netty.channel.MonoSendMany.subscribe(MonoSendMany.java:102) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:154) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.Mono.subscribe(Mono.java:4046) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxConcatIterable$ConcatIterableSubscriber.onComplete(FluxConcatIterable.java:147) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.FluxConcatIterable.subscribe(FluxConcatIterable.java:60) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.MonoFromFluxOperator.subscribe(MonoFromFluxOperator.java:81) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) ~[reactor-core-3.4.1.jar:3.4.1]
	at reactor.netty.http.client.HttpClientConnect$HttpIOHandlerObserver.onStateChange(HttpClientConnect.java:399) ~[reactor-netty-http-1.0.2.jar:1.0.2]
	at reactor.netty.ReactorNetty$CompositeConnectionObserver.onStateChange(ReactorNetty.java:612) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.netty.resources.DefaultPooledConnectionProvider$DisposableAcquire.onStateChange(DefaultPooledConnectionProvider.java:195) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.netty.resources.DefaultPooledConnectionProvider$PooledConnection.onStateChange(DefaultPooledConnectionProvider.java:466) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at reactor.netty.channel.ChannelOperationsHandler.channelActive(ChannelOperationsHandler.java:63) ~[reactor-netty-core-1.0.2.jar:1.0.2]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:230) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:216) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelActive(AbstractChannelHandlerContext.java:209) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelActive(CombinedChannelDuplexHandler.java:412) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.ChannelInboundHandlerAdapter.channelActive(ChannelInboundHandlerAdapter.java:69) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.CombinedChannelDuplexHandler.channelActive(CombinedChannelDuplexHandler.java:211) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:230) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:216) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelActive(AbstractChannelHandlerContext.java:209) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelActive(DefaultChannelPipeline.java:1398) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:230) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:216) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.DefaultChannelPipeline.fireChannelActive(DefaultChannelPipeline.java:895) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.fulfillConnectPromise(AbstractNioChannel.java:305) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:335) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:707) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493) ~[netty-transport-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.55.Final.jar:4.1.55.Final]
	at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]

Update 2:
So, I changed the body to this (To ignore the error):

.body(
                data
                    .log()
                    .onErrorContinue { t, u ->
                        println(t.message)
                        println(u)
                    },
                ByteArray::class.java
            )

Surprisingly, I could see the result (incomplete result) like below:

[
     {
          "fileName": "TestFileName",
         "body": "testdat2"
     }
]

I am saying incomplete result because I sent (Flux.just("data", "test", "dat2")), so the first data is missing, I can clearly see in the log the error on the first onNext only

2021-01-14 23:20:33.739  INFO 84905 --- [ctor-http-nio-6] reactor.Flux.SwitchIfEmpty.3             : onSubscribe(FluxSwitchIfEmpty.SwitchIfEmptySubscriber)
2021-01-14 23:20:33.739  INFO 84905 --- [ctor-http-nio-6] reactor.Flux.SwitchIfEmpty.3             : request(128)
2021-01-14 23:20:33.740  INFO 84905 --- [ctor-http-nio-6] reactor.Flux.SwitchIfEmpty.3             : onNext([116, 101, 115, 116])
2021-01-14 23:20:33.740  INFO 84905 --- [ctor-http-nio-6] reactor.Flux.SwitchIfEmpty.3             : onNext([100, 97, 116, 50])
refCnt: 0
io.rsocket.util.ByteBufPayload@1ad41beb
2021-01-14 23:20:33.741  INFO 84905 --- [ctor-http-nio-6] reactor.Flux.SwitchIfEmpty.3             : onComplete()
New Data: testdat2
Filename: TestFileName

So, something is happening in core security library for the first call and cannot find the data, that's why it throws exception.

[ghahramani]

Update 3:

Solution (Maybe):

I have no idea why this is happening but I could fix it via using unicast. Here is my WebClient part. Have a look into using Flux.merge and Sinks.many().unicast().onBackpressureBuffer<ByteArray>()

@MessageMapping("not-working")
    fun notWorking(@Payload data: Flux<ByteArray>): Flux<Result> {
        val sinks = Sinks.many().unicast().onBackpressureBuffer<ByteArray>()
        return Flux.merge(
            data
                .doOnNext { result -> sinks.tryEmitNext(result) }
                .doFinally { sinks.tryEmitComplete() }
                .flatMap { Mono.empty<Result>() }
                .log(),
            webClient
                .post()
                .uri(
                    UriComponentsBuilder
                        .fromUriString("http://localhost:9999/api/v2/uploads.json")
                        .queryParam("filename", "TestFileName")
                        .build(true)
                        .toUri()
                )
                .contentType(MediaType("application", "binary"))
                .body(
                    sinks.asFlux(),
                    ByteArray::class.java
                )
                .retrieve()
                .bodyToFlux()
        )
    }

So this works, but I am not sure this is the best solution as it is just a workaround. I think the framework should not throw that error. I let the spring security guys decide about it.

[ghahramani]

Update 4:

It seems when I use subscribeOn for the first part of the merge (Code below) on another thread subscribeOn(Schedulers.parallel()), I see the error sometimes, maybe in 10% of the time. Not sure it is related to the thing is happening in spring security core

data
                .doOnNext { result -> sinks.tryEmitNext(result) }
                .doFinally { sinks.tryEmitComplete() }
                .flatMap { Mono.empty<Result>() }
                .log()
                .subscribeOn(Schedulers.parallel())