Incidental Context Propagation With In-Memory Reactive Messaging Channels

[gilday]

Describe the bug

A Quarkus application that uses an Emitter to send messages to a reactive messaging pipeline incidentally propagates context to the Incoming method. Of course, this only happens when the channel is "in-memory". When the channel uses a connector - for example, Kafka - then, as expected, propagation does not occur, because the context is lost when the message is sent to Kafka.

This creates a leaky abstraction with respect to context propagation in @Incoming methods. These methods unexpectedly inherit context from the Emitter's scope only when the channel uses an in-memory connector. I would expect an @Incoming method to never inherit context, so that its relationship to the current context is consistent regardless of the connector being used.

Expected behavior

The @Incoming method should execute with a cleared context regardless of whether it receives messages from an external system (e.g. Kafka) or an in-memory data structure.

Actual behavior

Context is propagated from the Quarkus REST resource to the Reactive Messaging consumer. This denies the Reactive Messaging consumer the ability to start its own request context to use while consume messages. For example, if the @Incoming method were also decorated with @ActivateRequestScope, then the method would use the request scope created by Quarkus REST, and that scope will be destroyed when Quarkus REST has finished serving the request.

How to Reproduce?

This small Quarkus application reproduces the issue. Its test passes, because it uses a ManagedExecutor to clear the context before calling the Emitter. This is a workaround to the described issue. Removing this workaround and instead executing the commented out code return emitter.send(body); (no ManagedExecutor used) demonstrates the problem.

@Path("/hello")
class GreetingResource {

  @Channel("greeting")
  Emitter<String> emitter;

  @Inject
  @ManagedExecutorConfig(maxAsync = 2, maxQueued = 3, cleared = CDI)
  ManagedExecutor executor;

  @POST
  @Consumes(MediaType.TEXT_PLAIN)
  public CompletionStage<Void> hello(String body) {
    // context incidentally propagated, use ManagedExecutor to clear context at the emitter
    // return emitter.send(body);
    return CompletableFuture.supplyAsync(() -> emitter.send(body), executor).join();
  }
}

@ApplicationScoped
class GreetingProcessor {

  @Inject RequestContextController controller;

  @Inject Logger logger;

  @Incoming("greeting")
  void process(String name) {
    if (!controller.activate()) {
      throw new IllegalStateException("Request context is already active");
    }
    logger.infof("Hello, %s", name);
  }
}

quarkus-emitter-new-request-context.zip

Output of uname -a or ver

Darwin MacBookPro 24.2.0 Darwin Kernel Version 24.2.0: Fri Dec 6 19:01:59 PST 2024; root:xnu-11215.61.5~2/RELEASE_ARM64_T6000 arm64

Output of java -version

openjdk version "21.0.6" 2025-01-21 LTS OpenJDK Runtime Environment Temurin-21.0.6+7 (build 21.0.6+7-LTS) OpenJDK 64-Bit Server VM Temurin-21.0.6+7 (build 21.0.6+7-LTS, mixed mode, sharing)

Quarkus version or git rev

3.18.3

Build tool (ie. output of mvnw --version or gradlew --version)

Apache Maven 3.9.6 (bc0240f3c744dd6b6ec2920b3cd08dcc295161ae) Maven home: /Users/jgilday/.m2/wrapper/dists/apache-maven-3.9.6-bin/3311e1d4/apache-maven-3.9.6 Java version: 21.0.6, vendor: Eclipse Adoptium, runtime: /Library/Java/JavaVirtualMachines/temurin-21.jdk/Contents/Home Default locale: en_US, platform encoding: UTF-8 OS name: "mac os x", version: "15.2", arch: "aarch64", family: "mac"

Additional information

No response

[quarkus-bot]

/cc @FroMage (context-propagation), @cescoffier (reactive-messaging), @manovotn (context-propagation), @ozangunalp (reactive-messaging)

[ozangunalp]

#46174 will resolve this and bunch of other problems with context propagation in Quarkus messaging

[michalvavrik]

#46174 will resolve this and bunch of other problems with context propagation in Quarkus messaging

please consider linking this issue to #46174 so that the issue is automatically closed

[ryandens]

I was able to test out the reproducer project here with a build of #46174 on my machine. Adapting the Emitter to be a ContextualEmitter and annotating the producer method with @CurrentThreadContext(propagated = {}) does indeed fix this issue. Thanks for the extensive docs - it wasn't immediately clear to be that i should use the ContextualEmitter when I want to emit a message without context propagation occurring. Is it correct to say that a ContextualEmitter should be used when you need to respect Context Propagation configuration?

Additionally, while I'm really quite interested in seeing this issue fixed in the long term, is there a known workaround for forcing context to be cleared in 3.18.3 or later? I imagine that this fix will target 3.20.0 or 3.21.0, so looking for advice on upgrade options in the short term. The workaround suggested in this issue seems to use a ManagedExecutor only seems to work on short-lived non-blocking message consumers.

When using a blocking executor with a thread pool of size 2 and hitting the API endpoint 3 times concurrently, the 3rd message to get processed triggers the illegal state exception.

@Path("/hello")
class GreetingResource {

  @Channel("greeting")
  Emitter<String> emitter;

  @Inject Logger logger;

  @Inject
  UriInfo uriInfo;

  @Inject
  @ManagedExecutorConfig(maxAsync = 2, maxQueued = 3, propagated = {})
  ManagedExecutor executor;


  @POST
  @Consumes(MediaType.TEXT_PLAIN)
  @CurrentThreadContext(propagated = {})
  public CompletionStage<Void> hello(String body) {
    logger.infof("Received request: %s", uriInfo.getPath());
    return CompletableFuture.supplyAsync(() -> emitter.send(body), executor).join();
  }
}

@ApplicationScoped
class GreetingProcessor {

  @Inject RequestContextController controller;

  @Inject Logger logger;

  @Incoming("greeting")
  @Blocking(value = "greeting-pool", ordered = false)
  void process(String name) throws InterruptedException{
    if (!controller.activate()) {
      throw new IllegalStateException("Request context is already active");
    }
    logger.infof("Hello, %s before sleep", name);
    try {
        Thread.sleep(10_000);
        logger.infof("Hello, %s after sleep", name);
    } finally {
      controller.deactivate();
    }
  }
}

and updating the test:

@QuarkusTest
class GreetingResourceTest {
  @Test
  void testHelloEndpoint() throws ExecutionException, InterruptedException {
    CompletableFuture<ValidatableResponse> future1 = CompletableFuture.supplyAsync(() ->
            given().body("world").when().post("/hello").then()
    );
    CompletableFuture<ValidatableResponse> future2 = CompletableFuture.supplyAsync(() ->
            given().body("world1").when().post("/hello").then()
    );
    CompletableFuture<ValidatableResponse> future3 = CompletableFuture.supplyAsync(() ->
            given().body("world2").when().post("/hello").then()
    );

    CompletableFuture.allOf(future1, future2, future3).join();

    future1.get().statusCode(204);
    future2.get().statusCode(204);
    future3.get().statusCode(204);
  }
}

[FroMage]

I was able to test out the reproducer project here with a build of #46174 on my machine. Adapting the Emitter to be a ContextualEmitter and annotating the producer method with @CurrentThreadContext(propagated = {}) does indeed fix this issue. Thanks for the extensive docs - it wasn't immediately clear to be that i should use the ContextualEmitter when I want to emit a message without context propagation occurring. Is it correct to say that a ContextualEmitter should be used when you need to respect Context Propagation configuration?

@ozangunalp could you explain to me why ContextualEmitter is required? The above code looks like the emitter should already have zero contexts captured, so I'm not sure why it is not working. You surely know.

[ozangunalp]

@ryandens thanks for looking into this.

I've written some explanation on the description of the PR: #46174 (comment) (section with Note on previous PR flaky tests and ContextualEmitter)

I'll try to rephrase it,

• For the propagation of request-scoped context, cleared/not propagated doesn't mean the absence of request-scoped context activation, it is setting a fresh empty state for request-scoped beans.
• So when you do a @CurrentThreadContext(propagated = {}) or use a @ManagedExecutorConfig(propagated = {}) it makes the contextual call with an empty but active request-scoped context.
• Reactive messaging emitter sends, even for internal channels, are not single-stage async completion stages. Each message is dispatched (so consumed) within its own Vert.x context, which by default duplicates the state of the caller.
• When you try not to propagate the request-scoped context, with some form of propagated = {}, the message is emitted in the contextual call but not guaranteed to be consumed during that call, before the empty state is deactivated/destroyed.
• That is why in some cases you've "active request context", in other cases you've not active request context.
• The way ContextualEmitter resolves this is by making sure the context propagation is active on a single call where the message context is captured, and that the contextual call is returned (deactivating/destroying the request context) before emitting the message.
• If an empty but active request context is nevertheless required on the consumer method, @ActivateRequestContext can still be used.

Not related to this question but for completeness, there were other cases where the context propagation plan didn't work at all, which are fixed by changes to the WorkerPoolRegistry to ensure messages are dispatched on the correct Vert.x context.

@FroMage does this make sense?

[ryandens]

This makes sense! this also lines up with my anecdotal experience - i was nervous about seeing a request context active on my worker thread because I thought that meant my request handling thread would destroy the context when the request was done processing but that my worker thread would expect the context to be active. it sounds like from your comment that's not a concern because although a request context is propagated to the worker thread, it is not the same request context as the one processing the worker HTTP request and emitted the message on the reactive messaging channel?

[FroMage]

@ozangunalp OK thanks for the explanation. I did notice recently that the CDI cleared context would start a new request context. I'm not sure how that's really helping any scenario, but well… it is what it is.

So here, it will start an empty context, and what happens is that it will also destroy it before your consumers have a chance to use it? This is not very useful. I wonder if we've really thought this use-case through in CP, @manovotn. This is not a scenario we've tested a lot, I think. I'm not sure how it composes, but it's possible that we're creating and destroying lots of empty contexts and perhaps we have some bits that capture these empty contexts but can't use them before they're destroyed.

[ozangunalp]

@FroMage replied in #46174 (comment)