使用onErrorResume()处理错误时如何保留Reactor Context？

现象复现 #

以下测试代码可正常通过：

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import reactor.util.context.Context;

public class ReactorContextTest {
    @Test
    void test() {
        String key = "some key";
        String value = "some value";
        Mono<Void> mono = Mono.just(value)
                .flatMap(v -> Mono.<Void>empty().contextWrite(Context.of(key, v)))
//                .onErrorResume(Exception.class, t -> {
//                    System.out.println("onErrorResume() triggered...");
//                    return Mono.empty();
//                })
                ;

        StepVerifier.create(mono)
                .expectAccessibleContext()
                .contains(key, value)
                .then()
                .verifyComplete();
    }
}

取消注释onErrorResume()相关代码后，出现两个问题：

• 日志从未打印
• 测试失败，错误信息如下：

java.lang.AssertionError: Expected value some value for key some key, key not present
Context: Context0{}
Captured at: onErrorResume

问题根源 #

查看onErrorResume()的实现：

// Mono
public final Mono<T> onErrorResume(Function<? super Throwable, ? extends Mono<? extends
        T>> fallback) {
    return onAssembly(new MonoOnErrorResume<>(this, fallback));
}

核心问题总结：

• onErrorResume()会返回一个全新的Publisher
• 这个新Publisher的Context为空，且不会委托被包装的原Publisher处理上下文查询
• 无论onErrorResume()是否触发错误处理逻辑，原有Context都会丢失

注意：并非所有onErrorX()方法都有此问题，比如onErrorContinue()不会返回新的Publisher，链式调用后测试可正常通过；flatMap()虽返回新Publisher，但能正确保留Context。

解决方案 #

要在写入Context后实现错误时切换Publisher，同时保留上下文，可通过以下方式处理：

方法一：在fallback中显式继承上游上下文 #

修改onErrorResume()的fallback逻辑，让返回的Mono继承当前的上下文：

.onErrorResume(Exception.class, t -> {
    System.out.println("onErrorResume() triggered...");
    // 显式将当前上下文传递给fallback的Mono
    return Mono.empty().contextWrite(Context::currentContext);
})

方法二：调整Context写入位置（业务允许时） #

如果业务逻辑允许，可将contextWrite移到整个链式调用的最后，确保上下文在onErrorResume之后仍能被保留：

Mono<Void> mono = Mono.just(value)
        .flatMap(v -> Mono.<Void>empty())
        .onErrorResume(Exception.class, t -> {
            System.out.println("onErrorResume() triggered...");
            return Mono.empty();
        })
        .contextWrite(Context.of(key, value));

原理说明：onErrorResume的默认实现不会自动携带上游上下文，因此需要显式传递或调整上下文写入时机，确保测试能访问到预期的Context内容。

内容的提问来源于stack exchange，提问作者Sergey Zolotarev