带SocketTimeout重试的Reactor流单元测试：重试逻辑未触发排查

问题描述 #

我编写了一个调用foobarMethod的Reactor流，该方法可能抛出SocketTimeoutException类型的Mono异常。预期触发异常时重试2次，手动运行代码时功能正常，但JUnit测试中无法触发重试逻辑。请问这是单元测试环境问题还是测试代码编写有误？

服务代码 #

public void foobarServiceMethod() {
   foobarClient.foobarMethod(foobarParameter)
            .onErrorMap(ex -> {
                log.error("Code has failed");
                if (ex instanceof SocketTimeoutException) {
                    return new SocketTimeoutException("");
                } else {
                    return ex;
                }
            })
            .retryWhen(
                    Retry.fixedDelay(3, Duration.ofSeconds(1))
                            .filter(SocketTimeoutException.class::isInstance))
            .subscribe(foobarResponse -> handleFoobarResoponse(foobarResponse));
}

测试代码 #

@Test
void foobarRest(CapturedOutput capturedOutput) throws SocketTimeoutException, InterruptedException {
    foobarParameter = "test";
    when(foobarClient.foobarMethod(any()))
            .thenReturn(Mono.error(new SocketTimeoutException("First timeout")))
            .thenReturn(Mono.error(new SocketTimeoutException("First timeout")));

    foobarMethod(foobarParameter);

    assertEquals(2, capturedOutputSplit.contains("Code has failed"));
}

问题排查与修复 #

你的问题出在测试代码编写错误，和环境无关，核心问题有三个：

1. 异步流未等待执行完成 #

Reactor流是异步执行的，测试方法调用foobarMethod后直接进入断言逻辑，此时重试流程可能还没执行完，测试就已经结束了，导致日志未完全打印。

2. Mock返回的错误次数不匹配 #

你配置的Retry.fixedDelay(3, Duration.ofSeconds(1))意味着会重试3次，加上初始调用总共会发起4次请求，但你的Mock只返回了2次错误，无法触发完整的重试流程。如果你的预期是重试2次（总调用3次），应该把重试次数改为2。

3. 断言逻辑错误 #

capturedOutputSplit.contains("Code has failed")返回的是布尔值，不能直接和数字2做相等判断，你需要统计该字符串在输出中出现的实际次数。

修正后的测试代码 #

方式1：手动等待（简单但不够优雅） #

@Test
void foobarRest(CapturedOutput capturedOutput) throws InterruptedException {
    foobarParameter = "test";
    // 模拟3次错误（对应重试2次，总调用3次）
    when(foobarClient.foobarMethod(any()))
            .thenReturn(Mono.error(new SocketTimeoutException("First timeout")))
            .thenReturn(Mono.error(new SocketTimeoutException("Second timeout")))
            .thenReturn(Mono.error(new SocketTimeoutException("Third timeout")));

    foobarServiceMethod();
    // 等待重试流程完成（2次重试+初始调用，间隔1秒，留冗余时间）
    Thread.sleep(3000);

    // 统计错误日志出现次数
    long errorCount = Arrays.stream(capturedOutput.getOut().split("\n"))
            .filter(line -> line.contains("Code has failed"))
            .count();
    assertEquals(3, errorCount); // 初始调用+2次重试，共3次错误日志
}

方式2：使用StepVerifier（Reactor测试标准方式） #

先修改服务方法，让它返回Mono而非直接subscribe，方便测试：

public Mono<FoobarResponse> foobarServiceMethod() {
   return foobarClient.foobarMethod(foobarParameter)
            .onErrorMap(ex -> {
                log.error("Code has failed");
                if (ex instanceof SocketTimeoutException) {
                    return new SocketTimeoutException("");
                } else {
                    return ex;
                }
            })
            .retryWhen(
                    Retry.fixedDelay(2, Duration.ofSeconds(1)) // 重试2次，总调用3次
                            .filter(SocketTimeoutException.class::isInstance));
}

测试方法：

@Test
void foobarRest(CapturedOutput capturedOutput) {
    foobarParameter = "test";
    when(foobarClient.foobarMethod(any()))
            .thenReturn(Mono.error(new SocketTimeoutException("First timeout")))
            .thenReturn(Mono.error(new SocketTimeoutException("Second timeout")))
            .thenReturn(Mono.error(new SocketTimeoutException("Third timeout")));

    // 使用StepVerifier订阅并验证流的执行
    StepVerifier.create(foobarServiceMethod())
            .expectError(SocketTimeoutException.class)
            .verify(Duration.ofSeconds(3)); // 等待流执行完成

    // 统计错误日志次数
    long errorCount = Arrays.stream(capturedOutput.getOut().split("\n"))
            .filter(line -> line.contains("Code has failed"))
            .count();
    assertEquals(3, errorCount);
}

内容的提问来源于stack exchange，提问作者Hywel Griffiths