Propagating ContextLocals when sending Kafka Messages using Emitter

[pk-public]

Hi.

I need to propagate tenant ID as a header in every Kafka Message that is being sent in the context of a tenant.

I have a quarkus project written in kotlin (with coroutines), and I'm sending messages using Emitter with sendSuspending from quarkus-smallrye-reactive-messaging-kotlin - however the behavior is the same for other methods, reactive or not.

My idea was to propagate the tenant ID into a kafka metadata header in a PublisherDecorator (or OutgoingInterceptor), however, if I store the value in a Vertx duplicated context using ContextLocals, like so:

        ContextLocals.put("events-ContextLocals", "sendEvent")
        eventsEmitter.sendSuspending(event)

it's not accessible in the Multi stream handled in PublisherDecorator.decorate.

What's interesting though, is that some quarkus-related contexts are propagated to the decorator stream just fine, like:

"io.quarkus.vertx.core.runtime.VertxMDC"
"io.quarkus.opentelemetry.runtime.QuarkusContextStorage.otelContext" 

But others are not, like:

"io.quarkus.vertx.cdi-current-context" -> {RequestContext$RequestContextState@16501} 
"VertxRoute" -> "/"
"UrlPathTemplate" -> "/simulator/init"

My question is - what's the correct approach to propagate context variables to the publisher interceptor, and how does quarkus implementation of VertxMDC / OTEL achieve this?
I've been trying to find the implementation in the code, but I've failed to do that - found QuarkusContextStorage, but didn't help much.

[pk-public]

Inspired by the OpenTelemetryMpContextPropagationProvider, I've also tried implementing the ThreadContextProvider, but doesn't propagate the value as expected either (the provider is registered and being executed) - this even caused VertxMDC and QuarkusContextStorage.otelContext not to propagate ...

public class TenantThreadContextProvider implements ThreadContextProvider {

    public static final String TENANT_KEY = "tenant";

    @NotNull
    private static ThreadContextSnapshot emptyContextSnapshot() {
        return () -> () -> {
        };
    }


    public static Context getVertxContext() {
        Context context = Vertx.currentContext();
        if (context != null && VertxContext.isOnDuplicatedContext()) {
            return context;
        } else if (context != null) {
            Context dc = VertxContext.createNewDuplicatedContext(context);
            setContextSafe(dc, true);
            return dc;
        }
        return null;
    }

    private static void setTenantContext(Object tenant) {
        final Context vertxContext = getVertxContext();
        if (vertxContext != null)
            vertxContext.putLocal(TENANT_KEY, tenant);
    }

    @Override
    public ThreadContextSnapshot currentContext(Map<String, String> props) {
        final Context context = getVertxContext();
        if (context == null)
            return emptyContextSnapshot();

        final Object tenant = context.getLocal(TENANT_KEY);
        if (tenant == null)
            return emptyContextSnapshot();
        else {
            return () -> {
                Context context1 = getVertxContext();
                Object currentTenant = context1 != null ? context1.getLocal(TENANT_KEY) : null;
                setTenantContext(tenant);
                return () -> setTenantContext(currentTenant);
            };
        }
    }

    @Override
    public ThreadContextSnapshot clearedContext(Map<String, String> props) {
            return emptyContextSnapshot();
    }

    @Override
    public String getThreadContextType() {
        return TENANT_KEY;
    }
}

[pk-public]

After many hours of debugging quarkus / vertx / smallrye source code, I realized that PublisherDecorator is used on the receiver side, not the sender side.
I should have implemented OutgoingInterceptor instead, which I btw did in the beginning but at some point replaced with PublisherDecorator and forgot about that 🤦
I haven't noticed the issue, cause I had both - producer and consumer configured in the same application so it was executing as it should, but obviously the context was missing.