WebClient subscribing twice

[Sam-Kruglov]

Expected behavior

Only hit reactor.netty.channel.FluxReceive#startReceiver once

Actual behavior

Hits twice, where second time throws silently the "Only one connection receive subscriber allowed."

In production, sometimes this exception is reported from reactor.core.publisher.Operators : Operator called default onErrorDropped ....
Locally I could only see this exception at the point of throwing inside FluxReceive#startReceiver with a debugger. Later in reactor.core.publisher.MonoFlatMap.FlatMapMain#onError it was done = false, so the exception was never shown anywhere.

Steps to reproduce

My calling code is the following:

webClient.get()
      .uri("myUri")
      .retrieve()
      .onStatus(HttpStatus::isError, response ->
                          response.bodyToMono(ErrorResponse.class)
                                  .log("response.bodyToMono")
                                  .flatMap(error -> Mono.error(
                                          new MyException(
                                                  MyErrorType.ofCode(error.getCode()).orElse(null),
                                                  error.getDescription()
                                          )
                                  ))
        )
      .bodyToMono(MyDto.class)

the logs:

client.bodyToMono                        : | onSubscribe([Fuseable] MonoFlatMap.FlatMapMain)
client.bodyToMono                        : | request(unbounded)
response.bodyToMono                      : | onSubscribe([Fuseable] MonoSingle.SingleSubscriber)
response.bodyToMono                      : | request(unbounded)
response.bodyToMono                      : | onNext(ErrorResponse(code=500, description=message))
response.bodyToMono                      : | cancel()
client.bodyToMono                        : | onError(com.bla.MyException: message)
client.bodyToMono                        :  the exception stacktrace

Reactor Core version

org.springframework:spring-webflux:5.1.6.RELEASE

JVM version (e.g. java -version)

openjdk version "11.0.1" 2018-10-16

========

The workaround for me is using an ExchangeFilterFunction https://stackoverflow.com/a/48984852/6166627 instead of onStatus. With this approach, I only hit the start receiver once with a debugger. Hopefully, it's gonna fix itself.

Anyway, this ERROR log seems to not affect anything actually, the code still throws MyException

---

originally posted in reactor but it's off-topic there reactor/reactor-core#1747
Similar: #22096 #22284

[poutsma]

I wrote a test to reproduce this problem (see 9f7dd9f), but it seems to work fine. Closing this for now, but let us now how we can improve the test to reproduce this issue.

[gmuth]

I'd like to confirm the behaviour reported by Sam (my used version is 5.1.7.RELEASE).
To be more precise: the logging message appears twice.
This happens within the IDE (IntelliJ). There might be an issue with annotation processing - it's just a guess. We use a Lombok injected Logger via @slf4j.
As long as the onStatus-lambda (called twice) does not change state this is not an issue (like in my case).

[rstoyanchev]

@gmuth please provide a sample if you'd like us to look further. As you can see above we did try the above snippet, a test was written, but could not reproduce it.

[DurgeshAlex]

I am also facing the same issue with spring-webflux 5.2.9.RELEASE and using on spring boot version 2.1.11 and it seems to call the the endpoint twice. Below is the dependencies i am using

org.springframework
spring-webflux
5.2.9.RELEASE


io.projectreactor
reactor-core
3.3.10.RELEASE


io.projectreactor.netty
reactor-netty
0.9.12.RELEASE

[gmuth]

good point I haven't checked. Maybe the onStatus() lambda calls are handled correct triggered by according endpoint calls.

1. someone should confirm that indeed the endpoint is called twice (e.g. by checking the server side).
2. we should find out why the "call-endpoint" operation is called multiple times.

[DurgeshAlex]

Yeahh i created dummy endpoint in local server itself which is being called by webclient and kept the log and debug point in the endpoint i created. And as result debug point got invoked twice and log printed twice . That concluded that endpoint is being called twice.
And also i added a filter while creating the the WebClient instance. Some thing like this

And it seems that filter got invoked only once.
This is the log i am seeing

In the log above you can see that two 200 Ok response i got. with two different threads.
Hope it helps you.

[DurgeshAlex]

Twice calling got fixed by removing the subscribe() i think issue is their with subscribe function.

[ShimelesT]

I am having similar issue. I have list of Mono waiting response from multiple servers. And collectively block them after I subscribed to each of them individually. And I have to subscribe.

1. For each server
   create client
   make a get request
   subscribe to the mono
2. block all mono
3. generate status summary from each server

[bclozel]

You should (almost) never subscribe() to a publisher in a Spring application.

1. if you're writing a non-reactive application, you can call block() on publisher instances and get the value directly
2. if you're writing a reactive application, subscribe and block should never be called and you should use reactor operators instead to compose publishers if necessary

If you're reading this issue and you're still confused, I'd suggest writing a new question on StackOverflow showing a code snippet and explaining what you're trying to achieve.

[skyblackhawk]

We are a team of over 50 developers added to it system cloud architects, ans so on ... and we see in the log of SMS server receiving multiple request from the same webclient. We have verified it step by step enabling log on concurrent jakarta jdk21.0.6, (with only Weflux 3.3.4 reactive multiple projects API REST), before with tomcat and now with netty component and in the SMS server log receive multiple requests!

We have analyzed multiple time this case and finally we solved it using in webclient:
ExecutorService executorService = Executors.newSingleThreadExecutor();
CompletableFuture object as the response
injecting in async blocking forced.

More details on our solution that works fine here in stackoverflow to force mono request from webclient in webflux versus SMS service for this scenario. Using a single thread and get response in async when receive response.

I'm not joking webclient in reactive with tomcat or netty (we are migrated in netty) send more request I suppose it is done to respect reactive manifesto in webflux.

I'm sure about what I wrote before.
For this reason we are using ExecutorService with a single thread in webclient to avoid SMS server will send multiple OTP SMS to the customer.

[skyblackhawk]

Other question, why with retrieve it doesn't work and in exchange it works fine.
I agree with this generic definition:"The retrieve() method decodes the ClientResponse object and hands you the ready-made object for your use. It doesn't have a very nice api for handling exceptions.

However on the other hand the exchange() method hands you the ClientResponse object itself along with the response status and headers. With exchange method you get fine grained control over your response objects and a better way to handle the response object and the exceptions.

If you just want to consume some api go with retrieve().

If you want a better control over your response objects, headers and exceptions, go with exchange()."

or other definition:"According to spring Webclient api documentation the difference between the two is that exchange retrieve in addition to the body other http response information like headers and status, while retrieve only returns body information.

So If you only need the body information you should use retrieve, because it is a shortcut for exchange and then get the body, but if you need other information like http status you must use exchange."

Thanks for your reply.

[skyblackhawk]

Please see the real situation of Baeldung site

[bclozel]

@skyblackhawk This looks very odd. We never managed to reproduce this case so if you can provide a minimal sample application (for example with a client sending requests to httpbin.org), I'd be happy to look into it.

Right now I think this might be an invalid combination of calling "block" within a reactive controller method. I'm not sure why things are executed twice though in that case. Again, a minimal sample would be useful to explain things. I think the executors and retrieve vs exchange are red herrings at this stage and probably hiding the real problem.

Note: I don't think it's related to rate limiting either.

[skyblackhawk]

This problem we have seen happen for some critical servers that receive a lot of request on its API REST not reactive like this external service not reactive API REST smsSender on external server.

My hypothesis is that when arrive datagram of http to the SMS server, it not reply immediatly by handshake and to respect reactive manifesto webclient try to do another request by a new thread.
We are talking that this happen waiting 100-300 ms from the first request.

SMS Server of external service is on Azure Cloud and we are on GCP, we using an intermediate server in GCP to redirect call and get response (but it's trasparent to Webflux webclient it).

The solution before posted by me and developed by our teams avoid multiple request to send multiple SMS to the customer and it works fine.

We are super sure that solution fix the problem and send only one SMS to the customer without that solution SMS server always send two SMS to the customer and it isn't very nice for an OTP SMS.

[bclozel]

My hypothesis is that when arrive datagram of http to the SMS server not reply immediatly and to respect reactive manifesto webclient try to do another request by another thread but we are talking that this happen waiting 100-300 ms from the first request.

This behavior is definitely not part of the reactive streams behavior and should not happen. If the application uses a retry() operator this could happen but I don't see how scheduling work on a different executor would make things behave differently.

We are super sure that solution fix the problem and send only one SMS to the customer without that solution SMS server always send two SMS to the customer and it isn't very nice for an OTP SMS.

It looks like there is a problem in WebClient, unfortunately our tests do not show this behavior and we can't fix something unless we can replicate and understand the problem. If you can provide a minimal sample application that shows the issue I'd be happy to work on a fix. Reading your description of the problem, this should be fairly easy. httpbin.org provides a "/delay" endpoint that should help with that and using the log() operator on WebClient should show multiple executions in the logs. Please create a new issue in Spring Framework as soon as you can reproduce this. Thanks!

[skyblackhawk]

We don't use reply(), we manage it by OnSuccess and OnError.

I'm done working.
I'll do it this weekend, if you're okay with it, I can't do it before.
I think the tests should be done with Artillery or Jmeter tools for example to simulate a server that receives a lot of requests under pressure, but it isn't responsive, from a webclient webflux.
If you use just one simple request, you won't see this behavior because we only noticed it for certain servers on Azure Cloud from GCP.
Believe me our servers in Azure and GCP scale a lot.

[skyblackhawk]

WebFlux is actually my favorite framework, you have created a really excellent concurrent multi-threaded framework.
If we compare it to the Go language, are these the only languages with these performances!

Give compliments to all the guys in the team who worked on it, they really deserve it, maybe we who use it do not often say it to all of you, but share these compliments from me too.

You all have developed a serious framework that in my opinion will explode in artificial intelligence, (Are you working on a Springboot WebFlux module for AI?) to reduce electricity consumption and with an eye on the environment that we are literally destroying for the sake of future generations.

[bclozel]

@skyblackhawk thanks for the kind words, this means a lot!
No rush on that sample application, we will be releasing maintenance versions tomorrow and the next opportunity is next month. Plenty of time to work on a sample and a fix. Cheers!