How to implement a retry strategy from serverB to serverC using Spring WebFlux when building LLM gateway?

Retry mechanism for building LLM gateway using Spring WebFlux

When building an LLM gateway, communication between services needs to be handled and ensure that when a service is unavailable, it is possible to switch to the backup service seamlessly. This article will explore how to achieve this using Spring WebFlux, especially if gateway to Server B communication fails, how to retry and connect to Server C.

Scene description

Our LLM gateway call link is: Client-> Gateway-> Server B. If the gateway connection to Server B fails, we want the gateway to be able to retry and connect to Server C. This requires that the gateway can capture the error response code of Server B and automatically switch to Server C on failure.

Code analysis and improvement solutions

Let's first look at the original sseHttp method, which handles gateway requests to Server B or Server C:

Flux<response> responseFlux = webClient.create(url)
               .post()
               .headers(httpHeaders -> setHeaders(httpHeaders, headers))
               .contentType(MediaType.APPLICATION_JSON)
               .bodyValue(jsonBody)
               .retrieve()
               .onStatus(status -> status != HttpStatus.OK, response -> {
                   // Error handling logic})
               // ...Other logic...</response>

In order to implement the retry strategy, we need to capture the error response code of Server B and switch to Server C when an error occurs. There are some problems with previous attempts: simple try-catch cannot catch errors inside Flux ; the subscribe method is non-blocking, resulting in the error handling logic not taking effect in time.

Best Practice: Utilize retryWhen and onErrorResume

To solve the above problem, we should take advantage of retryWhen and onErrorResume operators provided by Spring WebFlux.

First, modify the sseHttp method and add retry logic:

Flux<response> sseHttp(String url) {
    return webClient.create(url)
            .post()
            .headers(httpHeaders -> setHeaders(httpHeaders, headers))
            .contentType(MediaType.APPLICATION_JSON)
            .bodyValue(jsonBody)
            .retrieve()
            .onStatus(HttpStatus::isError, clientResponse -> {
                // Record error logs to facilitate debugging return Mono.error(new WebClientResponseException("Server returned error status: " clientResponse.rawStatusCode(), clientResponse.rawStatusCode(), clientResponse.headers().asHttpHeaders(), clientResponse.bodyToMono(String.class).block(), null));
            })
            .bodyToFlux(typeRef)
            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
                    .filter(throwable -> throwable instanceof WebClientResponseException)
                    .onRetryExhaustedThrow((spec, signal) -> new GatewayException("Failed to connect to both Server B and Server C after multiple retries.")));
}</response>

This code uses onStatus to process HTTP error status codes and retry with retryWhen , retry up to 3 times, each time interval of 1 second. filter ensures that only exceptions of type WebClientResponseException are retryed. If the number of retrys is exhausted, GatewayException is thrown.

Then, where sseHttp is called, use onErrorResume to handle the failure of Server B and switch to Server C:

Mono<response> responseMono = sseHttp(serverBUrl)
       .onErrorResume(WebClientResponseException.class, ex -> {
           log.warn("Failed to connect to Server B: {}", ex.getMessage()); // Log error log return sseHttp(serverCUrl);
       })
       .next();</response>

This code first tries to connect to Server B, and if WebClientResponseException occurs, it tries to connect to Server C. The next() method ensures that only one result is returned.

Handle multiple successful responses

If both Server B and Server C successfully return data, we need to make sure that only one response is processed. An AtomicBoolean variable can be used to track whether the response has been processed successfully:

AtomicBoolean success = new AtomicBoolean(false);

Flux<response> sseHttp(String url) {
    // ... (previous code) ...
    .doOnNext(response -> {
        if (success.compareAndSet(false, true)) {
            // Processing a successful response}
    })
    // ... (rest of the code) ...
}</response>

Through the above improvements, we have implemented a more robust retry mechanism that can effectively handle communication failures between services and ensure high availability of LLM gateways. Remember to add sufficient logging to facilitate troubleshooting.

The above is the detailed content of How to implement a retry strategy from serverB to serverC using Spring WebFlux when building LLM gateway?. For more information, please follow other related articles on the PHP Chinese website!