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()
                .subscribeOn(Schedulers.parallel()),
            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.