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.

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`.

spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java

@@ -46,6 +46,8 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
 
 	private final List<ProducerInterceptor> interceptors;
 
+	private Schema<T> schema;
+
 	/**
 	 * Constructs a template instance.
 	 * @param producerFactory the producer factory used to create the backing Pulsar
@@ -103,8 +105,16 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
 
 	private Producer<T> prepareProducerForSend(String topic, T message, MessageRouter messageRouter)
 			throws PulsarClientException {
-		Schema<T> schema = SchemaUtils.getSchema(message);
+		Schema<T> 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<T> schema) {
+		this.schema = schema;
+	}
+
 }

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<T>) Schema.LOCAL_DATE;
 			case "java.time.LocalDateTime" -> (Schema<T>) Schema.LOCAL_DATE_TIME;
 			case "java.time.LocalTime" -> (Schema<T>) Schema.LOCAL_TIME;
-			default -> (Schema<T>) JSONSchema.of(message.getClass());
+			default -> (Schema<T>) Schema.BYTES;
 		};
 	}
 

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<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic)
+					.subscriptionName("test-specific-schema-subscription").subscribe()) {
+				PulsarProducerFactory<Foo> producerFactory = new DefaultPulsarProducerFactory<>(client,
+						Collections.singletonMap("topicName", topic));
+				PulsarTemplate<Foo> pulsarTemplate = new PulsarTemplate<>(producerFactory);
+				pulsarTemplate.setSchema(Schema.JSON(Foo.class));
+				Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID());
+				pulsarTemplate.send(foo);
+
+				CompletableFuture<Message<Foo>> receiveMsgFuture = consumer.receiveAsync();
+				assertThat(receiveMsgFuture).isCompleted().isCompletedWithValueMatching(m -> m.getValue().equals(foo));
+			}
+		}
+	}
+
 	private static Stream<Arguments> sendMessageTestProvider() {
 		return Stream.of(
 
@@ -216,4 +237,7 @@ class PulsarTemplateTests extends AbstractContainerBaseTests {
 
 	}
 
+	record Foo(String foo, String bar) {
+	}
+
 }