Pulsar force schema fetch on pulsar-client-thread for all schema requ…

…iring schema info fetching Fixes #2493
smallrye · Feb 21, 2024 · bfdbf19 · bfdbf19
1 parent f810ab8
commit bfdbf19
Showing 1 changed file with 2 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
@@ -115,8 +115,8 @@ 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 instanceof AutoConsumeSchema || schema instanceof KeyValueSchema) {
-                            return msgMulti.onItem().call(msg -> Uni.createFrom().item(msg::getValue));
+                        if (schema.requireFetchingSchemaInfo()) {
+                            return msgMulti.onItem().invoke(org.apache.pulsar.client.api.Message::getValue);
                         } else {
                             return msgMulti;
                         }