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<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;
+	}
+
 }
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<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;
 		};
 	}
 
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<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) {
+	}
+
 }