WebFlux retryBackoff() not works with exchange()

[altro3]

If you use retrieve() method all works fine:

      boolean send(String event) {

          Response response = webClient.post()
                  .body(fromObject(event))
                  .retrieve()
                  .bodyToMono(Response.class)
                  .retryBackoff(numRetries, firstBackoff, maxBackoff, jitterFactor)
                  .doOnError(t -> log.error("Some problems with request", t))
                  .onErrorReturn(ERROR)
                  .block();

          if (response != null && response.error) {
              log.error("Can't send event to OpenBroker");
              return false;
          }

          return true;
    }

    public static class Response {
        private final boolean error;
    }

for test you can use OkHttp MockWebServer (setup numRetries = 2 before):

    void test() {

       MockWebServer  server = new MockWebServer();
        server.start(InetAddress.getByName("localhost", "8888");

        MockResponse response = new MockResponse();
        response.setResponseCode(HttpStatus.INTERNAL_SERVER_ERROR.value());

        server.enqueue(response);
        server.enqueue(response);
        server.enqueue(response);
        assertFalse(send("test"));
    }

In logs you can see that webClient send 3 request and next thrown IllegalStateException:

jun. 07, 2019 4:38:38 PM okhttp3.mockwebserver.MockWebServer$4 processOneRequest
INFO: MockWebServer[8888] received request: POST / HTTP/1.1 and responded: HTTP/1.1 500 Server Error
16:38:38.949 DEBUG [reactor-http-nio-4] o.s.w.r.f.c.ExchangeFunctions [LogFormatUtils.java:91] [1dbeedff] Response 500 INTERNAL_SERVER_ERROR
16:38:39.023 DEBUG [parallel-2] o.s.w.r.f.c.ExchangeFunctions [LogFormatUtils.java:91] [1dbeedff] HTTP POST http://localhost:8888
jun. 07, 2019 4:38:39 PM okhttp3.mockwebserver.MockWebServer$4 processOneRequest
INFO: MockWebServer[8888] received request: POST / HTTP/1.1 and responded: HTTP/1.1 500 Server Error
16:38:39.028 DEBUG [reactor-http-nio-4] o.s.w.r.f.c.ExchangeFunctions [LogFormatUtils.java:91] [1dbeedff] Response 500 INTERNAL_SERVER_ERROR
16:38:39.079 DEBUG [parallel-4] o.s.w.r.f.c.ExchangeFunctions [LogFormatUtils.java:91] [1dbeedff] HTTP POST http://localhost:8888
jun. 07, 2019 4:38:39 PM okhttp3.mockwebserver.MockWebServer$4 processOneRequest
INFO: MockWebServer[8888] received request: POST / HTTP/1.1 and responded: HTTP/1.1 500 Server Error
16:38:39.138 DEBUG [reactor-http-nio-4] o.s.w.r.f.c.ExchangeFunctions [LogFormatUtils.java:91] [1dbeedff] Response 500 INTERNAL_SERVER_ERROR
16:38:39.140 DEBUG [reactor-http-nio-4] o.s.w.r.f.c.ExchangeFunctions [ExchangeFunctions.java:105] [1dbeedff] Cancel signal (to close connection)
16:38:39.143 ERROR [reactor-http-nio-4] i.k.c.s.o.OpenBrokerNoClient [OpenBrokerNoClient.java:304] Some problems with request to Open broker
java.lang.IllegalStateException: Retries exhausted: 2/2
    at reactor.core.publisher.FluxRetryWhen.lambda$null$0(FluxRetryWhen.java:260)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:350)
    at reactor.core.publisher.FluxIndex$IndexSubscriber.onNext(FluxIndex.java:95)
    at reactor.core.publisher.DirectProcessor$DirectInner.onNext(DirectProcessor.java:333)
    at reactor.core.publisher.DirectProcessor.onNext(DirectProcessor.java:142)
    at reactor.core.publisher.SerializedSubscriber.onNext(SerializedSubscriber.java:89)
    at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onError(FluxRetryWhen.java:160)
    <...>
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:906)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.springframework.web.reactive.function.client.WebClientResponseException$InternalServerError: 500 Internal Server Error
    at org.springframework.web.reactive.function.client.WebClientResponseException.create(WebClientResponseException.java:196)
    at org.springframework.web.reactive.function.client.DefaultWebClient$DefaultResponseSpec.lambda$createResponseException$13(DefaultWebClient.java:495)
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100)
    ... 45 common frames omitted
16:38:39.143 ERROR [main] i.k.c.s.o.OpenBrokerNoClient [OpenBrokerNoClient.java:310] Can't send event to OpenBroker
jun. 07, 2019 4:38:39 PM okhttp3.mockwebserver.MockWebServer$3 acceptConnections

If you try use exchange() method with retryBackoff() you will see that send only 1 request:

      boolean send(String event) {

          Response response = webClient.post()
                  .body(fromObject(event))
                  .exchange()
                  .flatMap(clientResponse -> clientResponse.bodyToMono(Response.class))
                  .retryBackoff(numRetries, firstBackoff, maxBackoff, jitterFactor)
                  .doOnError(t -> log.error("Some problems with request", t))
                  .onErrorReturn(ERROR)
                  .block();

          if (response != null && response.error) {
              log.error("Can't send event to OpenBroker");
              return false;
          }

          return true;
    }

    public static class Response {
        private final boolean error;
    }

Run same test and you will see in logs:

jun. 07, 2019 4:51:57 PM okhttp3.mockwebserver.MockWebServer$4 processOneRequest
INFO: MockWebServer[8888] received request: POST / HTTP/1.1 and responded: HTTP/1.1 500 Server Error
16:51:57.132 DEBUG [reactor-http-nio-4] o.s.w.r.f.c.ExchangeFunctions [LogFormatUtils.java:91] [7c4ca87c] Response 500 INTERNAL_SERVER_ERROR
jun. 07, 2019 4:51:57 PM okhttp3.mockwebserver.MockWebServer$3 acceptConnections
INFO: MockWebServer[8888] done accepting connections: Interrupted function call: accept failed
16:51:57.151 DEBUG [main] o.s.t.c.s.AbstractDirtiesContextTestExecutionListener [AbstractDirtiesContextTestExecutionListener.java:107] After test method: context [DefaultTestContext@6581dc0a testClass = Test, testInstance = Test@34b27915, testMethod = test@Test, testException = org.opentest4j.AssertionFailedError: expected: <false> but was: <true>, mergedContextConfiguration = [MergedContextConfiguration@39de3d36 testClass = Test, locations = '{}', classes = '', contextInitializerClasses = '[class org.springframework.boot.test.context.ConfigFileApplicationContextInitializer]', activeProfiles = '{test}', propertySourceLocations = '{}', propertySourceProperties = '{}', contextCustomizers = set[org.springframework.boot.test.autoconfigure.properties.PropertyMappingContextCustomizer@0, org.springframework.boot.test.autoconfigure.web.servlet.WebDriverContextCustomizerFactory$Customizer@1583741e, org.springframework.boot.test.context.filter.ExcludeFilterContextCustomizer@1efe439d, org.springframework.boot.test.json.DuplicateJsonObjectContextCustomizerFactory$DuplicateJsonObjectContextCustomizer@12d2ce03, org.springframework.boot.test.mock.mockito.MockitoContextCustomizer@0], contextLoader = 'org.springframework.test.context.support.DelegatingSmartContextLoader', parent = [null]], attributes = map[[empty]]], class annotated with @DirtiesContext [false] with mode [null], method annotated with @DirtiesContext [false] with mode [null].

org.opentest4j.AssertionFailedError: expected: <false> but was: <true>

No retries in this case.

Is this bug or I doing something wrong?

[DeCaMil]

It did retry, but the logs are saying a 500 error was received. You need to identify cause of the remote error. The remote has some sort of issue with the request you sent.

[altro3]

@DeCaMil Thanks for answer! But no, I specifically told mockServer to respond with an error 500 to the request, so that the retry mechanism worked. Look again:

        MockResponse response = new MockResponse();
        response.setResponseCode(HttpStatus.INTERNAL_SERVER_ERROR.value());

And look what I see in the logs in the end: in the first case there are 3 requests, in the second - only one. The problem is that using exchange did not work the retry mechanism.

[snicoll]

@altro3 sorry for the delay. Can you please move that code snippet into an actual project that we can use to run the test? That way, we'll be sure to test the same thing.

You can share the project with us by attaching a zip to this issue or by pushing the code to a separate GitHub repository. Thanks!

[altro3]

@snicoll Hi! Don't worry, it's only 4,5 years :-D, it could have been worse ;-). Well, I need to run this test again with the latest versions of the spring boot, because... This problem has been around for a long time and may have already been fixed. When I have free time, I'll do it

[spring-projects-issues]

If you would like us to look at this issue, please provide the requested information. If the information is not provided within the next 7 days this issue will be closed.

[spring-projects-issues]

Closing due to lack of requested feedback. If you would like us to look at this issue, please provide the requested information and we will re-open the issue.