Messaging Context propagation: Replace emitOn that requests upfront with operator that calls executor synchronously.

Fixes #47670

(cherry picked from commit daa3e28)

Author: ozangunalp

extensions/smallrye-reactive-messaging/runtime/src/main/java/io/quarkus/smallrye/reactivemessaging/runtime/ConnectorContextPropagationDecorator.java

@@ -2,6 +2,7 @@
 
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.Executor;
 
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
@@ -11,6 +12,10 @@
 import org.eclipse.microprofile.reactive.messaging.Message;
 
 import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.helpers.ParameterValidation;
+import io.smallrye.mutiny.operators.multi.AbstractMultiOperator;
+import io.smallrye.mutiny.operators.multi.MultiOperatorProcessor;
+import io.smallrye.mutiny.subscription.MultiSubscriber;
 import io.smallrye.reactive.messaging.PublisherDecorator;
 import io.smallrye.reactive.messaging.SubscriberDecorator;
 
@@ -32,7 +37,7 @@ public ConnectorContextPropagationDecorator(
     public Multi<? extends Message<?>> decorate(Multi<? extends Message<?>> publisher, List<String> channelName,
             boolean isConnector) {
         if (isConnector) {
-            return publisher.emitOn(tc.currentContextExecutor());
+            return new ContextPropagationOperator<>(publisher, tc);
         }
         return publisher;
     }
@@ -42,4 +47,42 @@ public int getPriority() {
         // Before the io.smallrye.reactive.messaging.providers.locals.ContextDecorator which has the priority 0
         return -100;
     }
+
+    public static class ContextPropagationOperator<T> extends AbstractMultiOperator<T, T> {
+
+        private final ThreadContext tc;
+
+        /**
+         * Creates a new {@link AbstractMultiOperator} with the passed {@link Multi} as upstream.
+         *
+         * @param upstream the upstream, must not be {@code null}
+         */
+        public ContextPropagationOperator(Multi<? extends T> upstream, ThreadContext tc) {
+            super(upstream);
+            this.tc = tc;
+        }
+
+        @Override
+        public void subscribe(MultiSubscriber<? super T> downstream) {
+            ParameterValidation.nonNullNpe(downstream, "subscriber");
+            upstream.subscribe().withSubscriber(new ContextPropagationProcessor<>(downstream, tc));
+        }
+
+        static final class ContextPropagationProcessor<T> extends MultiOperatorProcessor<T, T> {
+
+            private final Executor tcExecutor;
+
+            public ContextPropagationProcessor(MultiSubscriber<? super T> downstream, ThreadContext tc) {
+                super(downstream);
+                this.tcExecutor = tc.currentContextExecutor();
+            }
+
+            @Override
+            public void onItem(T item) {
+                // Even though the executor is called, this is a synchronous call
+                tcExecutor.execute(() -> super.onItem(item));
+            }
+
+        }
+    }
 }