Reactor Context Propagation in CorePublisherRateLimiterOperator#

Overview#

CorePublisherRateLimiterOperator is the shared implementation class used by both MonoRateLimiter and FluxRateLimiter to apply rate limiting in the Reactor operator stack. When RateLimiterOperator.of(rateLimiter) is applied to a Mono or Flux, the Publisher.apply(Publisher<T>) method routes to one of these wrappers, which each delegate subscribe(CoreSubscriber<? super T>) to CorePublisherRateLimiterOperator 1.

Subscription Flow#

The subscribe method calls rateLimiter.reservePermission(permits) to obtain a wait duration:

• waitDuration == 0 — permit available immediately; source.subscribe(new RateLimiterSubscriber<>(rateLimiter, actual)) is called synchronously.
• waitDuration > 0 — permit will be available after a delay; delaySubscription(actual, waitDuration) is called.
• waitDuration < 0 — limit exceeded; Operators.error(actual, RequestNotPermitted) is emitted.

The Context-Propagation Bug (Issue #2540)#

The delaySubscription method creates a separate Mono.delay(...) and subscribes to it with a plain lambda:

Mono.delay(Duration.ofNanos(waitDuration))
    .subscribe(delay -> source.subscribe(
        new RateLimiterSubscriber<>(rateLimiter, actual)));

This LambdaMonoSubscriber has an empty Reactor Context because actual.currentContext() is not passed to it 2. When two conditions hold simultaneously, this causes ThreadLocal values to be cleared before source is subscribed:

1. Hooks.enableAutomaticContextPropagation() is active — Reactor restores ThreadLocals from a subscriber's Reactor Context when crossing async boundaries, using clearMissing=true, which resets any ThreadLocal key registered in ContextRegistry that is absent from the subscriber's context.
2. An Operators.lift hook is registered via Hooks.onEachOperator — e.g., the OpenTelemetry Java agent's ContextPropagationOperator.registerOnEachOperator(). This hook wraps Mono.delay's subscriber in a ThreadLocal-restoring decorator built from the (empty) lambda subscriber's context.

Without a lift hook, the parallel scheduler thread retains whatever ThreadLocals it captured at scheduling time and nothing clears them. With the lift hook, Reactor explicitly restores ThreadLocals from the empty context with clearMissing=true at the point the delay fires, so registered accessors (e.g., an OTel trace context or a tenant ID) are reset to null 2.

Affected versions: Resilience4j 2.1.0–2.4.0 (same code path). The fix is not yet merged as of this writing 2.

Reproducer#

The failure is isolated to requests that the rate limiter delays (permit exhausted) with a lift hook present. The issue report includes a self-contained JUnit 5 reproducer:

noDelayKeepsThreadLocalWithLiftHook → passes
delayedSubscriptionKeepsThreadLocal → passes (no lift hook)
delayedSubscriptionKeepsThreadLocalWithLiftHook → FAILS: expected <A> but was <null>

Proposed Fixes#

Both proposals were verified against the reproducer 2:

Option 1 — Pass actual.currentContext() to the lambda subscriber:

Mono.delay(Duration.ofNanos(waitDuration))
    .subscribe(
        delay -> source.subscribe(new RateLimiterSubscriber<>(rateLimiter, actual)),
        e -> Operators.error(actual, e),
        null,
        actual.currentContext());

This is the minimal change: the fourth argument to Mono.subscribe(...) supplies the initial context to the LambdaMonoSubscriber, so the lift hook builds its ThreadLocal-restoring wrapper from the correct (non-empty) context.

Note: Mono.delay(...).contextWrite(actual.currentContext()).subscribe(lambda) does not fix the issue — contextWrite only affects upstream operators, not the subscriber's own context 2.

Option 2 — Move the delay into the subscribe chain using Flux.from(source).delaySubscription(Duration.ofNanos(waitDuration)), keeping the delay as part of the assembled reactive pipeline where Reactor handles context threading automatically.

Scope#

Both MonoRateLimiter and FluxRateLimiter delegate through CorePublisherRateLimiterOperator, so both are affected by the same delaySubscription code path 3 4.

Related: Bulkhead Resolved a Similar Problem Differently#

The analogous blocked-thread problem in MonoBulkhead/FluxBulkhead was resolved in PR #2506 by adding Bulkhead#acquirePermissionAsync() and using Operators.DeferredSubscription (the same primitive FluxDelaySubscription uses) to defer downstream subscription without spawning a side-chain. That PR explicitly includes a Context propagation test (BulkheadOperatorQueueingTest) 5. The RateLimiter operator does not yet have an equivalent non-blocking acquisition path.

Key Files#