How to reinvoke WebClient's ExchangeFilterFunction on retry

Question

When using reactor's retry(..) operator WebClient exchange filter functions are not triggered on retry. I understand why, but the issue is when a function (like bellow) generates an authentication token with an expiry time. It might happen, while a request is being "retried" the token expires because the Exchange function is not re-invoked during the retry. Is there a way how to re-generate a token for each retry?

Following AuthClientExchangeFunction generates an authentication token (JWT) with an expiration.

public class AuthClientExchangeFunction implements ExchangeFilterFunction {


    private final TokenProvider tokenProvider;

    public IntraAuthWebClientExchangeFunction(TokenProvider tokenProvider) {
        this.tokenProvider = tokenProvider;
    }

    @Override
    public Mono filter(ClientRequest request, ExchangeFunction next) {
        String jwt = tokenProvider.getToken();
        return next.exchange(withBearer(request, jwt));

    }

    private ClientRequest withBearer(ClientRequest request, String jwt){
        return ClientRequest.from(request)
                            .headers(headers -> headers.set(HttpHeaders.AUTHORIZATION, "Bearer "+ jwt))
                            .build();
    }
}

Lets say that a token is valid for 2999 ms -> Each retry request fails due to 401.

WebClient client = WebClient.builder()
                            .filter(new AuthClientExchangeFunction(tokenProvider))
                            .build();        

 client.get()
       .uri("/api")
       .retrieve()
       .bodyToMono(String.class)
       .retryBackoff(1, Duration.ofMillis(3000)) ;

Edit Here is an executable example

@SpringBootTest
@RunWith(SpringRunner.class)
public class RetryApplicationTests {


    private static final MockWebServer server  = new MockWebServer();

    private final RquestCountingFilterFunction requestCounter = new RquestCountingFilterFunction();

    @AfterClass
    public static void shutdown() throws IOException {
        server.shutdown();
    }

    @Test
    public void test() {

        server.enqueue(new MockResponse().setResponseCode(500).setBody("{}"));
        server.enqueue(new MockResponse().setResponseCode(500).setBody("{}"));
        server.enqueue(new MockResponse().setResponseCode(500).setBody("{}"));
        server.enqueue(new MockResponse().setResponseCode(200).setBody("{}"));

        WebClient webClient = WebClient.builder()
                                       .baseUrl(server.url("/api").toString())
                                       .filter(requestCounter)
                                       .build();

        Mono responseMono1 = webClient.get()
                                              .uri("/api")
                                              .retrieve()
                                              .bodyToMono(String.class)
                                              .retryBackoff(3, Duration.ofMillis(1000)) ;

        StepVerifier.create(responseMono1).expectNextCount(1).verifyComplete();

        assertThat(requestCounter.count()).isEqualTo(4);
    }



    static class RquestCountingFilterFunction implements ExchangeFilterFunction {

        final Logger log = LoggerFactory.getLogger(getClass());
        final AtomicInteger counter = new AtomicInteger();

        @Override
        public Mono filter(ClientRequest request, ExchangeFunction next) {
            log.info("Sending {} request to {} {}", counter.incrementAndGet(), request.method(), request.url());
            return next.exchange(request);
        }

        int count() {
            return counter.get();
        }
    }

}

output

MockWebServer[44855] starting to accept connections
Sending 1 request to GET http://localhost:44855/api/api
MockWebServer[44855] received request: GET /api/api HTTP/1.1 and responded: HTTP/1.1 500 Server Error
MockWebServer[44855] received request: GET /api/api HTTP/1.1 and responded: HTTP/1.1 500 Server Error
MockWebServer[44855] received request: GET /api/api HTTP/1.1 and responded: HTTP/1.1 500 Server Error
MockWebServer[44855] received request: GET /api/api HTTP/1.1 and responded: HTTP/1.1 200 OK

org.junit.ComparisonFailure: 
Expected :4
Actual   :1

user_7652010 · Answer

You need to update your spring-boot version to 2.2.0.RELEASE. retry() will not invoke exchange function in previous version.

I've tested this using a simple code (In Kotlin).

@Component
class AnswerPub {

    val webClient = WebClient.builder()
        .filter(PrintExchangeFunction())
        .baseUrl("https://jsonplaceholder.typicode.com").build()

    fun productInfo(): Mono {
        return webClient
            .get()
            .uri("/todos2/1")
            .retrieve()
            .bodyToMono(User::class.java)
            .retry(2) { it is Exception }
    }

    data class User(
        val id: String,
        val userId: String,
        val title: String,
        val completed: Boolean
    )

}

class PrintExchangeFunction : ExchangeFilterFunction {
    override fun filter(request: ClientRequest, next: ExchangeFunction): Mono {
        println("Filtered")
        return next.exchange(request)
    }

}

And the console output looked like:

2019-10-29 09:31:55.912  INFO 12206 --- [           main] o.s.b.web.embedded.netty.NettyWebServer  : Netty started on port(s): 8080
2019-10-29 09:31:55.917  INFO 12206 --- [           main] c.e.s.SpringWfDemoApplicationKt          : Started SpringWfDemoApplicationKt in 3.19 seconds (JVM running for 4.234)
Filtered
Filtered
Filtered

So in my case, the exchange function is invoked every single time.

How to reinvoke WebClient's ExchangeFilterFunction on retry

1 Answers