From 624dde15fcab669a1dc5175cee95f559bff7705f Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 15 Aug 2022 16:20:35 -0400 Subject: [PATCH] Pulsar template schema detection improvements PulsarTemplate can infer schema type for basic primitive types. For complex types like JSON, AVRO etc, it needs to support the ability for the user to provide the schema information. --- .../src/main/asciidoc/pulsar.adoc | 11 +++++++++ .../pulsar/core/PulsarTemplate.java | 12 +++++++++- .../pulsar/core/SchemaUtils.java | 3 +-- .../pulsar/core/PulsarTemplateTests.java | 24 +++++++++++++++++++ 4 files changed, 47 insertions(+), 3 deletions(-) diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc index 180feec0..3284dd3f 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc @@ -45,6 +45,17 @@ template.send(msg, (messageBuilder -> messageBuilder.key(myMessageKey))); ---- ==== +If you are using simple Java primitive types, then the framework auto-detects the schema for you, and you do not need to specify any schema types for publishing the data. +However, if you are using any complex types such as `JSON`, `AVRO`, `PROTOBUF`, etc. then you need to set the proper schema type on the `PulsarTemplate` before invoking any send operations as shown below. + +==== +[source, java] +---- +pulsarTemplate.setSchema(Schema.JSON(Foo.class)); +---- +==== + + When using partitioned topics, then you can use custom partitioning routing when publishing records. For this purpose, you can provide an implementation of a `MessageRouter` and pass it along with both `send` and `sendAsync` methods. Note that, when using a `MessageRouter`, you must set the `spring.pulsar.producer.messageRoutingMode` property to `custom`. diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java index cfd29481..5b563818 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java @@ -46,6 +46,8 @@ public class PulsarTemplate implements PulsarOperations { private final List interceptors; + private Schema schema; + /** * Constructs a template instance. * @param producerFactory the producer factory used to create the backing Pulsar @@ -103,8 +105,16 @@ public class PulsarTemplate implements PulsarOperations { private Producer prepareProducerForSend(String topic, T message, MessageRouter messageRouter) throws PulsarClientException { - Schema schema = SchemaUtils.getSchema(message); + Schema schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message); return this.producerFactory.createProducer(topic, schema, messageRouter, this.interceptors); } + /** + * Setter for schema. + * @param schema provides the {@link Schema} used on this template + */ + public void setSchema(Schema schema) { + this.schema = schema; + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java index 689dc23e..e3540b82 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java @@ -17,7 +17,6 @@ package org.springframework.pulsar.core; import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.impl.schema.JSONSchema; /** * Utility class for Pulsar schema inference. @@ -49,7 +48,7 @@ public final class SchemaUtils { case "java.time.LocalDate" -> (Schema) Schema.LOCAL_DATE; case "java.time.LocalDateTime" -> (Schema) Schema.LOCAL_DATE_TIME; case "java.time.LocalTime" -> (Schema) Schema.LOCAL_TIME; - default -> (Schema) JSONSchema.of(message.getClass()); + default -> (Schema) Schema.BYTES; }; } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java index 69ff53f3..65afb903 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java @@ -29,6 +29,7 @@ import java.time.Duration; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.stream.Stream; @@ -45,6 +46,7 @@ import org.apache.pulsar.client.api.TopicMetadata; import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Named; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; @@ -118,6 +120,25 @@ class PulsarTemplateTests extends AbstractContainerBaseTests { } } + @Test + void sendMessageWithSpecificSchemaTest() throws Exception { + String topic = "smt-specific-schema-topic"; + try (PulsarClient client = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build()) { + try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic) + .subscriptionName("test-specific-schema-subscription").subscribe()) { + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, + Collections.singletonMap("topicName", topic)); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); + pulsarTemplate.setSchema(Schema.JSON(Foo.class)); + Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); + pulsarTemplate.send(foo); + + CompletableFuture> receiveMsgFuture = consumer.receiveAsync(); + assertThat(receiveMsgFuture).isCompleted().isCompletedWithValueMatching(m -> m.getValue().equals(foo)); + } + } + } + private static Stream sendMessageTestProvider() { return Stream.of( @@ -216,4 +237,7 @@ class PulsarTemplateTests extends AbstractContainerBaseTests { } + record Foo(String foo, String bar) { + } + }