Take 2 : Pulsar force schema fetch on pulsar-client-thread for all sc…

…hema using AbstractMultiVersionReader Fixes #2513
smallrye · Mar 6, 2024 · 94b3fa1 · 94b3fa1
1 parent 249d074
commit 94b3fa1
Showing 1 changed file with 14 additions and 2 deletions.
diff --git a/...ing-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/PulsarIncomingChannel.java b/...ing-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/PulsarIncomingChannel.java
@@ -13,6 +13,10 @@
 import org.apache.pulsar.client.api.*;
 import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.impl.schema.AvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema;
 import org.eclipse.microprofile.reactive.messaging.Message;
 
 import io.opentelemetry.api.GlobalOpenTelemetry;
@@ -112,7 +116,7 @@ public PulsarIncomingChannel(PulsarClient client, Vertx vertx, Schema<T> schema,
                     .until(m -> closed.get())
                     .plug(msgMulti -> {
                         // Calling getValue on the pulsar-client-internal thread to make sure the SchemaInfo is fetched
-                        if (schema.requireFetchingSchemaInfo()) {
+                        if (schemaRequiresBLockingFetch(schema)) {
                             return msgMulti.onItem().invoke(org.apache.pulsar.client.api.Message::getValue);
                         } else {
                             return msgMulti;
@@ -136,7 +140,7 @@ public PulsarIncomingChannel(PulsarClient client, Vertx vertx, Schema<T> schema,
                     .filter(m -> m.size() > 0)
                     .plug(msgMulti -> {
                         // Calling getValue on the pulsar-client-internal thread to make sure the SchemaInfo is fetched
-                        if (schema.requireFetchingSchemaInfo()) {
+                        if (schemaRequiresBLockingFetch(schema)) {
                             return msgMulti.onItem().invoke(msg -> msg.forEach(org.apache.pulsar.client.api.Message::getValue));
                         } else {
                             return msgMulti;
@@ -169,6 +173,14 @@ public PulsarIncomingChannel(PulsarClient client, Vertx vertx, Schema<T> schema,
                 .buildConsumerInstrumenter(PulsarTraceTextMapGetter.INSTANCE);
     }
 
+    private static <T> boolean schemaRequiresBLockingFetch(Schema<T> schema) {
+        return schema.requireFetchingSchemaInfo()
+                || schema instanceof AvroSchema
+                || schema instanceof GenericAvroSchema
+                || schema instanceof GenericJsonSchema
+                || schema instanceof GenericProtobufNativeSchema;
+    }
+
     public void incomingTrace(PulsarMessage<T> pulsarMessage) {
         PulsarIncomingMessageMetadata metadata = pulsarMessage.getMetadata(PulsarIncomingMessageMetadata.class).get();
         TracingUtils.traceIncoming(instrumenter, pulsarMessage, new PulsarTrace.Builder()