Копировать контекст Reactor в MD C в случае TimeoutMainSubscriber

Question

Я следовал коду на https://github.com/archie-swif/webflux-mdc/blob/master/src/main/java/com/example/webfluxmdc/MdcContextLifter.java, чтобы поместить данные из Reactor Context в MD C, используя перехватчики.

Решение отлично работает в большинстве случаев, кроме когда Mono истекает и TimeoutMainSubscriber используется.

Ниже приведен небольшой тест, который я использовал.

Mono<Integer> integerMono = Mono.just(1)
            .doOnEach(logger.addToContext(i -> ImmutableMap.of("index", i)))
            .flatMap(___ -> Mono.fromSupplier(() -> {
                try {
                    Thread.sleep(6000);
                } catch (InterruptedException e) {
                    return 0;
                }
                return 1;
            }))
            .timeout(Duration.ofMillis(5000), Schedulers.parallel())
            .doOnEach(logger.info("testMDCLogging"))
            .doOnError(logger.error("testMDCErrorLogging"))
            .subscriberContext(logger.initContext());

При проверке распространения Context через лифтеры MD C, onNext() onError() и onComplete() не вызываются в TimeoutMainSubscriber (где контекст копируется в MD C) вызывается только currentContext().

Как распространить контекст и скопировать в MD C для использования в doOnError и doOnEach в случае Mono таймаута? Я не хочу перемещать копию в currentContext(), потому что этот метод вызывается несколько раз и, на мой взгляд, не является оптимальным решением.

Answer 1

Вы пытались переопределить метод onError в классе MdcContextLifter?

    @Override
    public void onError(Throwable t) {
        copyToMdc(coreSubscriber.currentContext()); // similar to onNext()
        coreSubscriber.onError(t);
    }