Propagate thread context when using vertx EventBus

[CosminBadea1]

I'm trying to get context propagation working when using event bus. I've created a simple flow following this guide https://quarkus.io/guides/reactive-event-bus.

The code looks something like this:

@Inject
EventBus bus;

@Inject
@ManagedExecutorConfig(propagated = "SLF4J_MDC")
ManagedExecutor managedExecutor;

@GET
@Produces(MediaType.TEXT_PLAIN)
@Path("{name}")
public Uni<String> hello(String name) {
    MDC.put("key", "someValue");
    Context context = Vertx.currentContext();
    ContextLocals.put("key", "Local Context");

    return bus.<String>request("greeting", name)
            .emitOn(managedExecutor)      // I also tried with - context::runOnContext
            .runSubscriptionOn(managedExecutor)
            .onItem().transform(Message::body);
}

@ConsumeEvent(value = "greeting")
@CurrentThreadContext(propagated = {"SLF4J_MDC"})
public String greeting(String name) {
    log.info("MDC2 ->>> {}", MDC.get("key"));                            // null
    log.info("from Context Local {}", ContextLocals.<String>get("key")); // null

    return "Hello " + name.toUpperCase();
}

SLF4J_MDC is a simple ThreadContextProvider implementation that I have registered in order to propagate the MDC values. I already tested this mechanism with regular async code and it worked, but not with eventBus.

Endpoint works as expected, but the values I am trying to log are null.

What is the right way to use context propagation in this scenario? Is it possible or is it not intended to be used in conjunction with event bus requests?

[ashutoshk2206]

Take a look here - https://reactiverse.io/reactiverse-contextual-logging/

Contextual data is not propagated over the EventBus. In this case, it must be:

1. added to message headers on the sender side
2. retrieved from headers and saved again on the receiver side

This process can be automated with EventBus interceptors:

vertx.eventBus().addOutboundInterceptor(event -> {
  String requestId = ContextualData.get("requestId");
  if (requestId != null) {
    event.message().headers().add("requestId", requestId);
  }
  event.next();
});

vertx.eventBus().addInboundInterceptor(event -> {
  String requestId = event.message().headers().get("requestId");
  if (requestId != null) {
    ContextualData.put("requestId", requestId);
  }
  event.next();
});

[CosminBadea1]

Thanks for the response @ashutoshk2206 . Unfortunately, I went back to Spring in the meantime 😂

[tugcesahin]

Hey @ashutoshk2206, I am trying to propagate MDC context to event bus consumers in Quarkus and I applied your solution above but it doesn't work for me. I am using slf4j as logging library which is not supported according to https://reactiverse.io/reactiverse-contextual-logging/#_supported_logging_libraries. Do you have any suggestions for that? Thank you.

Event is produced like:

bus.send("address", response.getId())
		.addOutboundInterceptor(event -> {
			for (MDCKey key : MDCKey.values()) {
				String val = MDC.get(key.value());
				if (StringUtils.isNotEmpty(val)) {
					event.message().headers().add(key.value(), val);
				}
			}
			event.next();
		});

And on consumer side, I fill the MDC context from event message headers after that read the MDC and see that MDC context is set, but outside of the interceptor; the correlationId from MDC but it is logged as null:

@ConsumeEvent(value = "address", blocking = true)
public void method(String val) {
         bus.addInboundInterceptor(event -> {
			for (MDCKey key : MDCKey.values()) {
				String val = event.message().headers().get(key.value());
				if (StringUtils.isNotEmpty(val)) {
					MDC.put(key.value(), val);
					logger.info("mdc with key {} value {}", key.value(), val);
				}	
			}
			event.next();
		});
	String corrId = MDC.get(MDCKeys.CORRELATION_ID);
	logger.info("corrId is {}", corrId);

[tugcesahin]

Update: If I assign the MDC context to a map inside interceptor and set that map to MDC, all is working as expected but I don't understand why setting only the MDC is not enough. Working piece of code:

bus.addInboundInterceptor(event -> {
	for (MDCKey key : MDCKey.values()) {
		String val = event.message().headers().get(key.value());
		if (StringUtils.isNotEmpty(val)) {
			MDC.put(key.value(), val);
			logger.info("mdc with key {} value {}", key.value(), val);
		}	
	}
	map = MDC.getCopyOfContextMap(); // assing MDC context map to a global map 
	event.next();
});
for (MDCKey key : MDCKey.values()) { // before setting context map to MDC, no MDC values are logged
	logger.info("mdc with key {} value {} after interceptor", key.value(), MDC.get(key.value()));
}
MDC.setContextMap(map); // MDC context map is set
for (MDCKey key : MDCKey.values()) { // after setting context map to MDC, all MDC values are logged as expected
	logger.info("mdc with key {} value {} after context map is set", key.value(), MDC.get(key.value()));
}

[tugcesahin]

I saw that in inbound interceptor, vert.x-eventloop-thread is opened, but after that interceptor, consumer continues its work in vert.x-worker-thread. That's why I lost MDC context but I can't figure out how to solve it as well.

[ashutoshk2206]

Hi @tugcesahin, MDC is for ThreadContext, so you cannot access data from MDC in vertx current context. Hence MDC is not reliable for vertx apps. You will not observe any issue if you run single request at a time. Issue starts happening when you do a load test with multiple concurrent request, thats where MDC data will get overridden in event loop threads.

If your consumer side of event bus is a blocking operation (or if apply @Blocking annotation on consume method), vertx assigns a worker thread to it. In this case the interceptor won't work as it can't access the data set in event loop context. To solve this, you will need to pass the context data as part of the message (may be in header), retrieve it in the consume side and use it.

[holomekc]

Quarkus provides a VertxMDC. Maybe this is suitable?