Xử lý lỗi, timeout và retry

Đúng, try/catch quanh code reactive gần như vô dụng: lỗi xảy ra sau này, khi dữ liệu chạy qua chuỗi, lúc đó method đã return từ lâu. Lỗi là một tín hiệu onError chạy dọc chuỗi, nên ta xử lý bằng operator.

1. Bài toán

Checkout gọi payment-service. Thực tế:

• Thỉnh thoảng trả 503 (đang deploy).
• Có lúc treo 30s vì ngân hàng chậm.
• Gợi ý sản phẩm (recommendation-service) lỗi thì checkout vẫn phải chạy.

2. Vấn đề

try {
    return paymentClient.pay(req);   // chỉ trả về Mono, chưa gọi gì
} catch (Exception e) {              // không bao giờ bắt được lỗi HTTP
    ...
}

• Không timeout: một request treo giữ connection, kết nối dồn lên.
• Không retry: lỗi tạm thời 1 giây cũng làm hỏng đơn hàng.
• Retry ngay lập tức, liên tục: dội thêm tải vào service đang yếu (retry storm).

3. Giải pháp

4. Code từng bước

Bước 1: Dependency: spring-boot-starter-webflux (đã có reactor.util.retry.Retry).

Bước 2: Config timeout ở tầng connection (bổ sung cho timeout() ở chuỗi):

@Bean
WebClient paymentWebClient(WebClient.Builder builder) {
    HttpClient http = HttpClient.create()
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000)
            .responseTimeout(Duration.ofSeconds(3));
    return builder.baseUrl("http://payment-service")
            .clientConnector(new ReactorClientHttpConnector(http))
            .build();
}

Bước 3: Code Java

public Mono<Payment> pay(PaymentRequest req) {
    return paymentWebClient.post().uri("/payments")
            .header("Idempotency-Key", req.orderId().toString())   // retry an toàn
            .bodyValue(req)
            .retrieve()
            .bodyToMono(Payment.class)
            .timeout(Duration.ofSeconds(3))                         // mỗi lần thử tối đa 3s
            .retryWhen(Retry.backoff(3, Duration.ofMillis(200))     // 200ms, 400ms, 800ms (+jitter)
                    .filter(this::isTransient)                      // chỉ retry lỗi tạm thời
                    .onRetryExhaustedThrow((spec, signal) -> signal.failure()))
            .doOnError(e -> log.warn("Payment failed order={}: {}", req.orderId(), e.toString()))
            .onErrorMap(e -> new PaymentUnavailableException("Payment unavailable", e));
}

private boolean isTransient(Throwable e) {
    return e instanceof TimeoutException
        || e instanceof WebClientRequestException                       // lỗi kết nối
        || (e instanceof WebClientResponseException r && r.getStatusCode().is5xxServerError());
}

// Phần không quan trọng: lỗi thì bỏ qua
public Mono<List<Product>> recommendations(Long userId) {
    return recoClient.forUser(userId)
            .timeout(Duration.ofMillis(500))
            .onErrorReturn(List.of());
}

// Fallback sang nguồn khác
public Mono<Price> price(Long productId) {
    return priceClient.live(productId)
            .onErrorResume(WebClientResponseException.ServiceUnavailable.class,
                           e -> priceCache.get(productId));
}

Map lỗi nghiệp vụ ra HTTP status:

@RestControllerAdvice
public class ApiErrorHandler {
    @ExceptionHandler(PaymentUnavailableException.class)
    @ResponseStatus(HttpStatus.SERVICE_UNAVAILABLE)
    public Map<String, String> handle(PaymentUnavailableException e) {
        return Map.of("error", e.getMessage());
    }
}

Bước 4: Test

@Test
void retriesThenSucceeds() {
    AtomicInteger attempts = new AtomicInteger();
    Mono<String> flaky = Mono.defer(() -> attempts.incrementAndGet() < 3
            ? Mono.error(new TimeoutException())
            : Mono.just("SUCCESS"));

    StepVerifier.create(flaky.retryWhen(Retry.backoff(3, Duration.ofMillis(10))))
            .expectNext("SUCCESS")
            .verifyComplete();
    assertEquals(3, attempts.get());
}

# tắt payment-service rồi:
curl -i -X POST localhost:8080/api/checkout -d '{"userId":1,"cartId":10}' \
  -H "Content-Type: application/json"     # sau ~vài giây: 503

5. Cách hoạt động

subscribe -> POST /payments --x 503
             chờ ~200ms
resubscribe -> POST /payments --x timeout 3s
             chờ ~400ms
resubscribe -> POST /payments --> 200 OK -> onNext(Payment)

Hết lượt retry:
  onError(503) -> doOnError(log) -> onErrorMap(PaymentUnavailable) -> 503 cho client

Retry hoạt động bằng cách subscribe lại từ đầu chuỗi phía trên nó. Vì WebClient là cold publisher nên mỗi lần subscribe là một HTTP request mới. Vị trí operator quan trọng: timeout đặt trước retryWhen là timeout cho từng lần thử; đặt sau là timeout tổng.

6. So sánh

7. Khi nào nên dùng / không nên dùng

• Luôn có timeout cho mọi lời gọi ra ngoài.
• Chỉ retry lỗi tạm thời (5xx, timeout, connection). Không retry 400/401/404.
• Chỉ retry thao tác idempotent hoặc có idempotency key. Retry POST /payments không có key có thể trừ tiền 2 lần.
• Retry nhiều tầng (gateway + checkout + payment, mỗi tầng 3 lần) nhân lên 27 lần. Chỉ retry ở một tầng.
• Cần circuit breaker thì dùng Resilience4j (bài 4.6.6), có operator cho Reactor.

Đúng rồi, về kết quả thì onErrorReturn(x) tương đương onErrorResume(e -> Mono.just(x)). Khác biệt ở chỗ onErrorResume linh hoạt hơn:

• Nhìn exception để quyết định: lỗi 404 thì trả rỗng, lỗi khác thì ném tiếp bằng Mono.error(e).
• Fallback sang lời gọi async khác (cache, service dự phòng).

.onErrorResume(e -> e instanceof WebClientResponseException.NotFound
        ? Mono.empty()
        : Mono.error(e))

Cẩn thận: nuốt lỗi ở chỗ quan trọng (như payment) làm đơn "thành công giả". Payment lỗi thì nên báo lỗi, không fallback bừa.

Đúng, doOnError chỉ "nhìn" lỗi đi qua (side effect), giống finally có log. Nó không thay đổi tín hiệu. Muốn đổi kết quả phải dùng onErrorResume/onErrorReturn/onErrorMap. Họ doOnXxx (doOnNext, doOnSuccess, doFinally) đều là side effect, dùng cho log và metric, không nên chứa logic chính hay gọi blocking.

Q1. Xử lý lỗi trong Reactor thế nào?

🇻🇳 Tiếng Việt:

Lỗi là tín hiệu onError chạy dọc chuỗi nên try/catch không bắt được. Em dùng onErrorReturn cho giá trị mặc định, onErrorResume để fallback sang Publisher khác, onErrorMap để đổi sang exception nghiệp vụ, và doOnError chỉ để log. Sau đó @RestControllerAdvice map exception ra HTTP status.

🇬🇧 English:

An error is an onError signal travelling down the chain, so try/catch does not catch it. I use onErrorReturn for a default value, onErrorResume to fall back to another publisher, onErrorMap to translate into a business exception, and doOnError only for logging. A @RestControllerAdvice then maps exceptions to HTTP status codes.

Q2. Retry trong Reactor nên cấu hình ra sao?

🇻🇳 Tiếng Việt:

Dùng retryWhen với Retry.backoff: số lần giới hạn, chờ tăng dần có jitter, filter chỉ retry lỗi tạm thời như 5xx và timeout. Chỉ retry thao tác idempotent hoặc có idempotency key, và chỉ retry ở một tầng để tránh retry storm. Retry hoạt động bằng cách subscribe lại nguồn.

🇬🇧 English:

Use retryWhen with Retry.backoff: a bounded number of attempts, exponential delay with jitter, and a filter so only transient errors like 5xx and timeouts are retried. Only retry idempotent operations or ones with an idempotency key, and retry at a single layer to avoid retry storms. Retry works by resubscribing to the source.

Q3. timeout đặt trước hay sau retryWhen?

🇻🇳 Tiếng Việt:

Đặt trước retryWhen thì đó là timeout cho từng lần thử, vì retry subscribe lại cả đoạn phía trên. Đặt sau thì là timeout tổng cho mọi lần thử. Thường em đặt cả hai: per-attempt ngắn và một giới hạn tổng.

🇬🇧 English:

Placed before retryWhen it is a per-attempt timeout, because retry resubscribes everything upstream. Placed after, it is an overall timeout across all attempts. I usually use both: a short per-attempt timeout and an overall limit.

Cách nhớ ngắn gọn

• Return = giá trị; Resume = Publisher khác; Map = đổi lỗi; doOn = chỉ nhìn.
• Mọi call ra ngoài phải có timeout.
• Retry: backoff + jitter + filter lỗi tạm thời + idempotent.
• Vị trí operator quyết định ý nghĩa.