Add custom schema mappings

See #269

- Still need to add docs

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/DefaultReactivePulsarListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -85,6 +85,7 @@ public class DefaultReactivePulsarListenerContainerFactory<T> implements Reactiv
 			ReactivePulsarListenerEndpoint<T> endpoint) {
 
 		ReactivePulsarContainerProperties<T> properties = new ReactivePulsarContainerProperties<>();
+		properties.setSchemaResolver(this.getContainerProperties().getSchemaResolver());
 
 		if (!CollectionUtils.isEmpty(endpoint.getTopics())) {
 			properties.setTopics(endpoint.getTopics());

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -31,6 +31,7 @@ import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.impl.schema.AvroSchema;
 import org.apache.pulsar.client.impl.schema.JSONSchema;
 import org.apache.pulsar.client.impl.schema.ProtobufSchema;
+import org.apache.pulsar.common.schema.KeyValue;
 import org.apache.pulsar.common.schema.KeyValueEncodingType;
 import org.apache.pulsar.common.schema.SchemaType;
 
@@ -43,7 +44,7 @@ import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
-import org.springframework.pulsar.core.SchemaUtils;
+import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.listener.Acknowledgement;
 import org.springframework.pulsar.listener.adapter.HandlerAdapter;
 import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter;
@@ -139,9 +140,14 @@ public class MethodReactivePulsarListenerEndpoint<V> extends AbstractReactivePul
 			messageParameter = parameter.get();
 		}
 
+		// TODO refactor this all out later to SchemaResolver
+		// ResolvableType messageType = resolvableType(messageParameter);
+		// Schema<?> schema = schemaResolver.getSchema(schemaType, messageType);
+
 		DefaultReactivePulsarMessageListenerContainer<?> containerInstance = (DefaultReactivePulsarMessageListenerContainer<?>) container;
 		ReactivePulsarContainerProperties<?> pulsarContainerProperties = containerInstance.getContainerProperties();
 		SchemaType schemaType = pulsarContainerProperties.getSchemaType();
+		SchemaResolver schemaResolver = pulsarContainerProperties.getSchemaResolver();
 		if (schemaType != SchemaType.NONE) {
 			switch (schemaType) {
 				case STRING -> pulsarContainerProperties.setSchema((Schema) Schema.STRING);
@@ -173,15 +179,22 @@ public class MethodReactivePulsarListenerEndpoint<V> extends AbstractReactivePul
 					pulsarContainerProperties.setSchema((Schema) messageSchema);
 				}
 				case KEY_VALUE -> {
-					Schema<?> messageSchema = getMessageKeyValueSchema(messageParameter);
+					Schema<?> messageSchema = getMessageKeyValueSchema(schemaResolver, messageParameter);
 					pulsarContainerProperties.setSchema((Schema) messageSchema);
 				}
 			}
 		}
 		else {
 			if (messageParameter != null) {
-				Schema<?> messageSchema = getMessageSchema(messageParameter,
-						(messageClass) -> SchemaUtils.getSchema(messageClass, false));
+				Schema<?> messageSchema = null;
+				ResolvableType type = resolvableType(messageParameter);
+				if (KeyValue.class.isAssignableFrom(type.getRawClass())) {
+					messageSchema = getMessageKeyValueSchema(schemaResolver, messageParameter);
+				}
+				else {
+					messageSchema = getMessageSchema(messageParameter,
+							(messageClass) -> schemaResolver.getSchema(messageClass, false));
+				}
 				if (messageSchema != null) {
 					pulsarContainerProperties.setSchema((Schema) messageSchema);
 				}
@@ -207,12 +220,12 @@ public class MethodReactivePulsarListenerEndpoint<V> extends AbstractReactivePul
 		return schemaFactory.apply(messageClass);
 	}
 
-	private Schema<?> getMessageKeyValueSchema(MethodParameter messageParameter) {
+	private Schema<?> getMessageKeyValueSchema(SchemaResolver schemaResolver, MethodParameter messageParameter) {
 		ResolvableType messageType = resolvableType(messageParameter);
 		Class<?> keyClass = messageType.resolveGeneric(0);
 		Class<?> valueClass = messageType.resolveGeneric(1);
-		Schema<? extends Class<?>> keySchema = SchemaUtils.getSchema(keyClass);
-		Schema<? extends Class<?>> valueSchema = SchemaUtils.getSchema(valueClass);
+		Schema<? extends Class<?>> keySchema = schemaResolver.getSchema(keyClass);
+		Schema<? extends Class<?>> valueSchema = schemaResolver.getSchema(valueClass);
 		return Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE);
 	}
 

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -26,7 +26,9 @@ import org.apache.pulsar.reactive.client.api.ReactiveMessageSender;
 import org.reactivestreams.Publisher;
 
 import org.springframework.core.log.LogAccessor;
-import org.springframework.pulsar.core.SchemaUtils;
+import org.springframework.lang.Nullable;
+import org.springframework.pulsar.core.DefaultSchemaResolver;
+import org.springframework.pulsar.core.SchemaResolver;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -43,15 +45,29 @@ public class ReactivePulsarTemplate<T> implements ReactivePulsarOperations<T> {
 
 	private final ReactivePulsarSenderFactory<T> reactiveMessageSenderFactory;
 
+	private final SchemaResolver schemaResolver;
+
+	@Nullable
 	private Schema<T> schema;
 
 	/**
-	 * Construct a template instance with observation configuration.
+	 * Construct a template instance that uses the default schema resolver.
 	 * @param reactiveMessageSenderFactory the factory used to create the backing Pulsar
 	 * reactive senders
 	 */
 	public ReactivePulsarTemplate(ReactivePulsarSenderFactory<T> reactiveMessageSenderFactory) {
+		this(reactiveMessageSenderFactory, new DefaultSchemaResolver());
+	}
+
+	/**
+	 * Construct a template instance with a custom schema resolver.
+	 * @param reactiveMessageSenderFactory the factory used to create the backing Pulsar
+	 * @param schemaResolver the schema resolver to use reactive senders
+	 */
+	public ReactivePulsarTemplate(ReactivePulsarSenderFactory<T> reactiveMessageSenderFactory,
+			SchemaResolver schemaResolver) {
 		this.reactiveMessageSenderFactory = reactiveMessageSenderFactory;
+		this.schemaResolver = schemaResolver;
 	}
 
 	@Override
@@ -106,10 +122,10 @@ public class ReactivePulsarTemplate<T> implements ReactivePulsarOperations<T> {
 		if (this.schema != null) {
 			/*
 			 * If the template has a schema, we can create the message sender right away
-			 * and use ReactiveMessageSender::sendMany to send them as a stream. Otherwise
-			 * we need to wait to get a message to create it and we can't share it between
-			 * messages. So we create one each time and use ReactiveMessageSender::sendOne
-			 * to send messages individually.
+			 * and use ReactiveMessageSender::sendMany to send them as a stream.
+			 * Otherwise, we need to wait to get a message to create it and we can't share
+			 * it between messages. So we create one each time and use
+			 * ReactiveMessageSender::sendOne to send messages individually.
 			 */
 			ReactiveMessageSender<T> sender = createMessageSender(topic, null, null);
 			return messages.map(MessageSpec::of).as(sender::sendMany)
@@ -134,7 +150,7 @@ public class ReactivePulsarTemplate<T> implements ReactivePulsarOperations<T> {
 
 	private ReactiveMessageSender<T> createMessageSender(String topic, T message,
 			ReactiveMessageSenderBuilderCustomizer<T> customizer) {
-		Schema<T> schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message);
+		Schema<T> schema = this.schema != null ? this.schema : this.schemaResolver.getSchema(message);
 		return this.reactiveMessageSenderFactory.createSender(topic, schema,
 				customizer == null ? Collections.emptyList() : Collections.singletonList(customizer));
 	}

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,6 +24,9 @@ import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.common.schema.SchemaType;
 
+import org.springframework.pulsar.core.DefaultSchemaResolver;
+import org.springframework.pulsar.core.SchemaResolver;
+
 /**
  * Contains runtime properties for a reactive listener container.
  *
@@ -44,6 +47,8 @@ public class ReactivePulsarContainerProperties<T> {
 
 	private SchemaType schemaType;
 
+	private SchemaResolver schemaResolver = new DefaultSchemaResolver();
+
 	private ReactivePulsarMessageHandler messageHandler;
 
 	private Duration handlingTimeout = Duration.ofMinutes(2);
@@ -84,6 +89,14 @@ public class ReactivePulsarContainerProperties<T> {
 		this.schemaType = schemaType;
 	}
 
+	public SchemaResolver getSchemaResolver() {
+		return this.schemaResolver;
+	}
+
+	public void setSchemaResolver(SchemaResolver schemaResolver) {
+		this.schemaResolver = schemaResolver;
+	}
+
 	public Collection<String> getTopics() {
 		return this.topics;
 	}

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,8 +22,10 @@ import static org.junit.jupiter.params.provider.Arguments.arguments;
 
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;
 
@@ -39,6 +41,7 @@ import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.PulsarTestContainerSupport;
 
 import reactor.core.publisher.Flux;
@@ -52,12 +55,12 @@ import reactor.core.publisher.Mono;
 class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
 
 	@Test
-	void sendMessagesWithSpecificSchemaTest() throws Exception {
-		String topic = "smt-specific-schema-topic-reactive";
+	void sendMessagesWithSpecificSchema() throws Exception {
+		String topic = "smt-specific-schema-reactive-topic";
 		try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build()) {
 			try (Consumer<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic)
-					.subscriptionName("test-specific-schema-subscription").subscribe()) {
+					.subscriptionName("smt-specific-schema-reactive-sub").subscribe()) {
 				MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
 				senderSpec.setTopicName(topic);
 				org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory<Foo> producerFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory<>(
@@ -80,6 +83,42 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
 		}
 	}
 
+	@Test
+	void sendMessagesWithSpecificSchemaAndCustomTypeMappings() throws Exception {
+		String topic = "smt-specific-schema-custom-reactive-topic";
+		try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
+				.build()) {
+			try (Consumer<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic)
+					.subscriptionName("smt-specific-schema-custom-reactive-sub").subscribe()) {
+				MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
+				senderSpec.setTopicName(topic);
+				org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory<Foo> producerFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory<>(
+						client, senderSpec, null);
+				// Custom schema resolver allows not calling setSchema on template
+				DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(
+						Collections.singletonMap(Foo.class, Schema.JSON(Foo.class)));
+				org.springframework.pulsar.reactive.core.ReactivePulsarTemplate<Foo> pulsarTemplate = new org.springframework.pulsar.reactive.core.ReactivePulsarTemplate<>(
+						producerFactory, schemaResolver);
+
+				List<Foo> foos = new ArrayList<>();
+				for (int i = 0; i < 10; i++) {
+					foos.add(new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()));
+				}
+				pulsarTemplate.send(Flux.fromIterable(foos)).subscribe();
+
+				// TODO figure out why ordering is not preserved when template does not
+				// have schema set
+				List<Foo> foos2 = new ArrayList<>();
+				for (int i = 0; i < 10; i++) {
+					CompletableFuture<Message<Foo>> receiveFuture = consumer.receiveAsync();
+					assertThat(receiveFuture).succeedsWithin(Duration.ofSeconds(3));
+					foos2.add(receiveFuture.get().getValue());
+				}
+				assertThat(foos).containsExactlyInAnyOrderElementsOf(foos2);
+			}
+		}
+	}
+
 	@ParameterizedTest(name = "{0}")
 	@MethodSource("sendMessageTestProvider")
 	void sendMessageTest(String testName, SendTestArgs testArgs) throws Exception {

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -59,11 +59,13 @@ import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.pulsar.config.PulsarClientConfiguration;
 import org.springframework.pulsar.config.PulsarClientFactoryBean;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.PulsarAdministration;
 import org.springframework.pulsar.core.PulsarProducerFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
 import org.springframework.pulsar.core.PulsarTestContainerSupport;
 import org.springframework.pulsar.core.PulsarTopic;
+import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
 import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory;
 import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry;
@@ -72,6 +74,7 @@ import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListe
 import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory;
 import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer;
 import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory;
+import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.User2;
 import org.springframework.pulsar.support.PulsarHeaders;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
@@ -492,6 +495,189 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 	}
 
+	@Nested
+	@ContextConfiguration(classes = SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.class)
+	class SchemaCustomMappingsTestCases {
+
+		static CountDownLatch jsonLatch = new CountDownLatch(3);
+		static CountDownLatch avroLatch = new CountDownLatch(3);
+		static CountDownLatch keyvalueLatch = new CountDownLatch(3);
+		static CountDownLatch protobufLatch = new CountDownLatch(3);
+
+		@Test
+		void jsonSchema() throws Exception {
+			PulsarProducerFactory<User2> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+					Collections.emptyMap());
+			PulsarTemplate<User2> template = new PulsarTemplate<>(pulsarProducerFactory);
+			template.setSchema(JSONSchema.of(User2.class));
+
+			for (int i = 0; i < 3; i++) {
+				template.send("json-custom-schema-topic", new User2("Jason", i));
+			}
+			assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+
+		@Test
+		void avroSchema() throws Exception {
+			PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+					Collections.emptyMap());
+			PulsarTemplate<User> template = new PulsarTemplate<>(pulsarProducerFactory);
+			template.setSchema(AvroSchema.of(User.class));
+
+			for (int i = 0; i < 3; i++) {
+				template.send("avro-custom-schema-topic", new User("Avi", i));
+			}
+			assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+
+		@Test
+		void keyvalueSchema() throws Exception {
+			PulsarProducerFactory<KeyValue<String, User2>> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
+					pulsarClient, Collections.emptyMap());
+			PulsarTemplate<KeyValue<String, User2>> template = new PulsarTemplate<>(pulsarProducerFactory);
+			Schema<KeyValue<String, User2>> kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class),
+					KeyValueEncodingType.INLINE);
+			template.setSchema(kvSchema);
+
+			for (int i = 0; i < 3; i++) {
+				template.send("keyvalue-custom-schema-topic", new KeyValue<>("Kevin", new User2("Kevin", 5150)));
+			}
+			assertThat(keyvalueLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+
+		@Test
+		void protobufSchema() throws Exception {
+			PulsarProducerFactory<Proto.Person> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+					Collections.emptyMap());
+			PulsarTemplate<Proto.Person> template = new PulsarTemplate<>(pulsarProducerFactory);
+			template.setSchema(ProtobufSchema.of(Proto.Person.class));
+
+			for (int i = 0; i < 3; i++) {
+				template.send("protobuf-custom-schema-topic",
+						Proto.Person.newBuilder().setId(i).setName("Paul").build());
+			}
+			assertThat(protobufLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+
+		@EnableReactivePulsar
+		@Configuration
+		static class SchemaCustomMappingsTestConfig {
+
+			@Bean
+			SchemaResolver customSchemaResolver() {
+				Map<Class<?>, Schema<?>> customMappings = new HashMap<>();
+				customMappings.put(User.class, Schema.AVRO(User.class));
+				customMappings.put(User2.class, Schema.JSON(User2.class));
+				customMappings.put(Proto.Person.class, Schema.PROTOBUF(Proto.Person.class));
+				return new DefaultSchemaResolver(customMappings);
+			}
+
+			@Bean
+			ReactivePulsarListenerContainerFactory<String> reactivePulsarListenerContainerFactory(
+					ReactivePulsarConsumerFactory<String> pulsarConsumerFactory, SchemaResolver schemaResolver) {
+				ReactivePulsarContainerProperties<String> containerProps = new ReactivePulsarContainerProperties<>();
+				containerProps.setSchemaResolver(schemaResolver);
+				return new DefaultReactivePulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProps);
+			}
+
+			@ReactivePulsarListener(id = "jsonListener", topics = "json-custom-schema-topic",
+					consumerCustomizer = "subscriptionInitialPositionEarliest")
+			Mono<Void> listenJson(User2 message) {
+				jsonLatch.countDown();
+				return Mono.empty();
+			}
+
+			@ReactivePulsarListener(id = "avroListener", topics = "avro-custom-schema-topic",
+					consumerCustomizer = "subscriptionInitialPositionEarliest")
+			Mono<Void> listenAvro(User message) {
+				avroLatch.countDown();
+				return Mono.empty();
+			}
+
+			@ReactivePulsarListener(id = "keyvalueListener", topics = "keyvalue-custom-schema-topic",
+					consumerCustomizer = "subscriptionInitialPositionEarliest")
+			Mono<Void> listenKeyvalue(KeyValue<String, User2> message) {
+				keyvalueLatch.countDown();
+				return Mono.empty();
+			}
+
+			@ReactivePulsarListener(id = "protobufListener", topics = "protobuf-custom-schema-topic",
+					consumerCustomizer = "subscriptionInitialPositionEarliest")
+			Mono<Void> listenProtobuf(Proto.Person message) {
+				protobufLatch.countDown();
+				return Mono.empty();
+			}
+
+			@Bean
+			ReactiveMessageConsumerBuilderCustomizer<?> subscriptionInitialPositionEarliest() {
+				return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+			}
+
+			record User2(String name, int age) {
+			}
+
+		}
+
+		/**
+		 * Do not convert this to a Record as Avro does not seem to work well w/ records.
+		 */
+		static class User {
+
+			private String name;
+
+			private int age;
+
+			User() {
+
+			}
+
+			User(String name, int age) {
+				this.name = name;
+				this.age = age;
+			}
+
+			public String getName() {
+				return name;
+			}
+
+			public void setName(String name) {
+				this.name = name;
+			}
+
+			public int getAge() {
+				return age;
+			}
+
+			public void setAge(int age) {
+				this.age = age;
+			}
+
+			@Override
+			public boolean equals(Object o) {
+				if (this == o) {
+					return true;
+				}
+				if (o == null || getClass() != o.getClass()) {
+					return false;
+				}
+				User user = (User) o;
+				return age == user.age && Objects.equals(name, user.name);
+			}
+
+			@Override
+			public int hashCode() {
+				return Objects.hash(name, age);
+			}
+
+			@Override
+			public String toString() {
+				return "User{" + "name='" + name + '\'' + ", age=" + age + '}';
+			}
+
+		}
+
+	}
+
 	@Nested
 	@ContextConfiguration(classes = ReactivePulsarListenerTests.PulsarHeadersTest.PulsarListenerWithHeadersConfig.class)
 	class PulsarHeadersTest {

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,6 +28,7 @@ import org.springframework.pulsar.annotation.EnablePulsar;
 import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
 import org.springframework.pulsar.config.PulsarListenerBeanNames;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
+import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.listener.PulsarContainerProperties;
 import org.springframework.pulsar.observation.PulsarListenerObservationConvention;
 import org.springframework.util.unit.DataSize;
@@ -55,9 +56,11 @@ public class PulsarAnnotationDrivenConfiguration {
 	ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
 			ObjectProvider<PulsarConsumerFactory<Object>> consumerFactoryProvider,
 			ObjectProvider<ObservationRegistry> observationRegistryProvider,
-			ObjectProvider<PulsarListenerObservationConvention> observationConventionProvider) {
+			ObjectProvider<PulsarListenerObservationConvention> observationConventionProvider,
+			SchemaResolver schemaResolver) {
 
 		PulsarContainerProperties containerProperties = new PulsarContainerProperties();
+		containerProperties.setSchemaResolver(schemaResolver);
 		containerProperties.setSubscriptionType(this.pulsarProperties.getConsumer().getSubscriptionType());
 		containerProperties.setObservationConvention(observationConventionProvider.getIfUnique());
 

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java

@@ -33,10 +33,12 @@ import org.springframework.pulsar.config.PulsarClientFactoryBean;
 import org.springframework.pulsar.core.CachingPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.PulsarAdministration;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.function.PulsarFunction;
 import org.springframework.pulsar.function.PulsarFunctionAdministration;
 import org.springframework.pulsar.function.PulsarSink;
@@ -65,26 +67,26 @@ public class PulsarAutoConfiguration {
 	}
 
 	@Bean
-	@ConditionalOnMissingBean(PulsarClientConfiguration.class)
+	@ConditionalOnMissingBean
 	public PulsarClientConfiguration pulsarClientConfiguration() {
 		return new PulsarClientConfiguration(this.properties.buildClientProperties());
 	}
 
 	@Bean
-	@ConditionalOnMissingBean(PulsarClientFactoryBean.class)
+	@ConditionalOnMissingBean
 	public PulsarClientFactoryBean pulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) {
 		return new PulsarClientFactoryBean(pulsarClientConfiguration);
 	}
 
 	@Bean
-	@ConditionalOnMissingBean(PulsarProducerFactory.class)
+	@ConditionalOnMissingBean
 	@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "false")
 	public PulsarProducerFactory<?> pulsarProducerFactory(PulsarClient pulsarClient) {
 		return new DefaultPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties());
 	}
 
 	@Bean
-	@ConditionalOnMissingBean(PulsarProducerFactory.class)
+	@ConditionalOnMissingBean
 	@ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "true", matchIfMissing = true)
 	public PulsarProducerFactory<?> cachingPulsarProducerFactory(PulsarClient pulsarClient) {
 		return new CachingPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties(),
@@ -94,31 +96,37 @@ public class PulsarAutoConfiguration {
 	}
 
 	@Bean
-	@ConditionalOnMissingBean(PulsarTemplate.class)
+	@ConditionalOnMissingBean
 	public PulsarTemplate<?> pulsarTemplate(PulsarProducerFactory<?> pulsarProducerFactory,
-			ObjectProvider<ProducerInterceptor> interceptorsProvider,
+			ObjectProvider<ProducerInterceptor> interceptorsProvider, SchemaResolver schemaResolver,
 			ObjectProvider<ObservationRegistry> observationRegistryProvider,
 			ObjectProvider<PulsarTemplateObservationConvention> observationConventionProvider) {
 		return new PulsarTemplate<>(pulsarProducerFactory, interceptorsProvider.orderedStream().toList(),
-				this.properties.getTemplate().isObservationsEnabled() ? observationRegistryProvider.getIfUnique()
-						: null,
+				schemaResolver, this.properties.getTemplate().isObservationsEnabled()
+						? observationRegistryProvider.getIfUnique() : null,
 				observationConventionProvider.getIfUnique());
 	}
 
 	@Bean
-	@ConditionalOnMissingBean(PulsarConsumerFactory.class)
+	@ConditionalOnMissingBean
+	public SchemaResolver schemaResolver() {
+		return new DefaultSchemaResolver();
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
 	public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient) {
 		return new DefaultPulsarConsumerFactory<>(pulsarClient, this.properties.buildConsumerProperties());
 	}
 
 	@Bean
-	@ConditionalOnMissingBean(PulsarAdministration.class)
+	@ConditionalOnMissingBean
 	public PulsarAdministration pulsarAdministration() {
 		return new PulsarAdministration(this.properties.buildAdminProperties());
 	}
 
 	@Bean
-	@ConditionalOnMissingBean(PulsarFunctionAdministration.class)
+	@ConditionalOnMissingBean
 	@ConditionalOnProperty(name = "spring.pulsar.function.enabled", havingValue = "true", matchIfMissing = true)
 	public PulsarFunctionAdministration pulsarFunctionAdministration(PulsarAdministration pulsarAdministration,
 			ObjectProvider<PulsarFunction> pulsarFunctions, ObjectProvider<PulsarSink> pulsarSinks,

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.pulsar.config.PulsarListenerBeanNames;
+import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
 import org.springframework.pulsar.reactive.config.annotation.EnableReactivePulsar;
 import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory;
@@ -46,9 +47,11 @@ public class PulsarReactiveAnnotationDrivenConfiguration {
 	@Bean
 	@ConditionalOnMissingBean(name = "reactivePulsarListenerContainerFactory")
 	DefaultReactivePulsarListenerContainerFactory<?> reactivePulsarListenerContainerFactory(
-			ObjectProvider<ReactivePulsarConsumerFactory<Object>> consumerFactoryProvider) {
+			ObjectProvider<ReactivePulsarConsumerFactory<Object>> consumerFactoryProvider,
+			SchemaResolver schemaResolver) {
 
 		ReactivePulsarContainerProperties<Object> containerProperties = new ReactivePulsarContainerProperties<>();
+		containerProperties.setSchemaResolver(schemaResolver);
 		containerProperties.setSubscriptionType(this.properties.getConsumer().getSubscriptionType());
 
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -32,6 +32,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
+import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory;
 import org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory;
 import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory;
@@ -113,9 +114,9 @@ public class PulsarReactiveAutoConfiguration {
 
 	@Bean
 	@ConditionalOnMissingBean
-	public ReactivePulsarTemplate<?> pulsarReactiveTemplate(
-			ReactivePulsarSenderFactory<?> reactivePulsarSenderFactory) {
-		return new ReactivePulsarTemplate<>(reactivePulsarSenderFactory);
+	public ReactivePulsarTemplate<?> pulsarReactiveTemplate(ReactivePulsarSenderFactory<?> reactivePulsarSenderFactory,
+			SchemaResolver schemaResolver) {
+		return new ReactivePulsarTemplate<>(reactivePulsarSenderFactory, schemaResolver);
 	}
 
 }

spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -58,6 +58,7 @@ public class ConcurrentPulsarListenerContainerFactory<T>
 	protected ConcurrentPulsarMessageListenerContainer<T> createContainerInstance(PulsarListenerEndpoint endpoint) {
 
 		PulsarContainerProperties properties = new PulsarContainerProperties();
+		properties.setSchemaResolver(this.getContainerProperties().getSchemaResolver());
 
 		if (!CollectionUtils.isEmpty(endpoint.getTopics())) {
 			properties.setTopics(endpoint.getTopics().toArray(new String[0]));

spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -32,6 +32,7 @@ import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.impl.schema.AvroSchema;
 import org.apache.pulsar.client.impl.schema.JSONSchema;
 import org.apache.pulsar.client.impl.schema.ProtobufSchema;
+import org.apache.pulsar.common.schema.KeyValue;
 import org.apache.pulsar.common.schema.KeyValueEncodingType;
 import org.apache.pulsar.common.schema.SchemaType;
 
@@ -44,7 +45,7 @@ import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
-import org.springframework.pulsar.core.SchemaUtils;
+import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.listener.Acknowledgement;
 import org.springframework.pulsar.listener.ConcurrentPulsarMessageListenerContainer;
 import org.springframework.pulsar.listener.PulsarConsumerErrorHandler;
@@ -148,6 +149,12 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
 		ConcurrentPulsarMessageListenerContainer<?> containerInstance = (ConcurrentPulsarMessageListenerContainer<?>) container;
 		PulsarContainerProperties pulsarContainerProperties = containerInstance.getContainerProperties();
 		SchemaType schemaType = pulsarContainerProperties.getSchemaType();
+		SchemaResolver schemaResolver = pulsarContainerProperties.getSchemaResolver();
+
+		// TODO refactor this all out later to SchemaResolver
+		// ResolvableType messageType = resolvableType(messageParameter);
+		// Schema<?> schema = schemaResolver.getSchema(schemaType, messageType);
+
 		if (schemaType != SchemaType.NONE) {
 			switch (schemaType) {
 				case STRING -> pulsarContainerProperties.setSchema(Schema.STRING);
@@ -179,15 +186,23 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
 					pulsarContainerProperties.setSchema(messageSchema);
 				}
 				case KEY_VALUE -> {
-					Schema<?> messageSchema = getMessageKeyValueSchema(messageParameter);
+					Schema<?> messageSchema = getMessageKeyValueSchema(schemaResolver, messageParameter);
 					pulsarContainerProperties.setSchema(messageSchema);
 				}
 			}
 		}
 		else {
 			if (messageParameter != null) {
-				Schema<?> messageSchema = getMessageSchema(messageParameter,
-						(messageClass) -> SchemaUtils.getSchema(messageClass, false));
+
+				Schema<?> messageSchema = null;
+				ResolvableType type = resolvableType(messageParameter);
+				if (KeyValue.class.isAssignableFrom(type.getRawClass())) {
+					messageSchema = getMessageKeyValueSchema(schemaResolver, messageParameter);
+				}
+				else {
+					messageSchema = getMessageSchema(messageParameter,
+							(messageClass) -> schemaResolver.getSchema(messageClass, false));
+				}
 				if (messageSchema != null) {
 					pulsarContainerProperties.setSchema(messageSchema);
 				}
@@ -210,12 +225,12 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
 		return schemaFactory.apply(messageClass);
 	}
 
-	private Schema<?> getMessageKeyValueSchema(MethodParameter messageParameter) {
+	private Schema<?> getMessageKeyValueSchema(SchemaResolver schemaResolver, MethodParameter messageParameter) {
 		ResolvableType messageType = resolvableType(messageParameter);
 		Class<?> keyClass = messageType.resolveGeneric(0);
 		Class<?> valueClass = messageType.resolveGeneric(1);
-		Schema<? extends Class<?>> keySchema = SchemaUtils.getSchema(keyClass);
-		Schema<? extends Class<?>> valueSchema = SchemaUtils.getSchema(valueClass);
+		Schema<? extends Class<?>> keySchema = schemaResolver.getSchema(keyClass);
+		Schema<? extends Class<?>> valueSchema = schemaResolver.getSchema(valueClass);
 		return Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE);
 	}
 

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

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2022-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.core;
+
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import org.apache.pulsar.client.api.Schema;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * Default schema resolver capable of handling basic message types.
+ *
+ * <p>
+ * Additional message types can be configured with the {@link #DefaultSchemaResolver(Map)}
+ * constructor.
+ *
+ * @author Soby Chacko
+ * @author Alexander Preuß
+ * @author Chris Bono
+ */
+public class DefaultSchemaResolver implements SchemaResolver {
+
+	private static final Map<Class<?>, Schema<?>> BASE_SCHEMA_MAPPINGS = new HashMap<>();
+	static {
+		BASE_SCHEMA_MAPPINGS.put(byte[].class, Schema.BYTES);
+		BASE_SCHEMA_MAPPINGS.put(String.class, Schema.STRING);
+		BASE_SCHEMA_MAPPINGS.put(Byte.class, Schema.INT8);
+		BASE_SCHEMA_MAPPINGS.put(byte.class, Schema.INT8);
+		BASE_SCHEMA_MAPPINGS.put(Short.class, Schema.INT16);
+		BASE_SCHEMA_MAPPINGS.put(short.class, Schema.INT16);
+		BASE_SCHEMA_MAPPINGS.put(Integer.class, Schema.INT32);
+		BASE_SCHEMA_MAPPINGS.put(int.class, Schema.INT32);
+		BASE_SCHEMA_MAPPINGS.put(Long.class, Schema.INT64);
+		BASE_SCHEMA_MAPPINGS.put(long.class, Schema.INT64);
+		BASE_SCHEMA_MAPPINGS.put(Boolean.class, Schema.BOOL);
+		BASE_SCHEMA_MAPPINGS.put(boolean.class, Schema.BOOL);
+		BASE_SCHEMA_MAPPINGS.put(ByteBuffer.class, Schema.BYTEBUFFER);
+		BASE_SCHEMA_MAPPINGS.put(ByteBuffer.allocate(0).getClass(), Schema.BYTEBUFFER);
+		BASE_SCHEMA_MAPPINGS.put(ByteBuffer.allocateDirect(0).getClass(), Schema.BYTEBUFFER);
+		BASE_SCHEMA_MAPPINGS.put(Date.class, Schema.DATE);
+		BASE_SCHEMA_MAPPINGS.put(Double.class, Schema.DOUBLE);
+		BASE_SCHEMA_MAPPINGS.put(double.class, Schema.DOUBLE);
+		BASE_SCHEMA_MAPPINGS.put(Float.class, Schema.FLOAT);
+		BASE_SCHEMA_MAPPINGS.put(float.class, Schema.FLOAT);
+		BASE_SCHEMA_MAPPINGS.put(Instant.class, Schema.INSTANT);
+		BASE_SCHEMA_MAPPINGS.put(LocalDate.class, Schema.LOCAL_DATE);
+		BASE_SCHEMA_MAPPINGS.put(LocalDateTime.class, Schema.LOCAL_DATE_TIME);
+		BASE_SCHEMA_MAPPINGS.put(LocalTime.class, Schema.LOCAL_TIME);
+	}
+
+	private final Map<Class<?>, Schema<?>> customSchemaMappings;
+
+	/**
+	 * Constructs a resolver with no custom type mappings.
+	 */
+	public DefaultSchemaResolver() {
+		this(Collections.emptyMap());
+	}
+
+	/**
+	 * Constructs a resolver with custom type mappings.
+	 * @param customTypeSchemaMappings additional type to schema mappings to use
+	 */
+	public DefaultSchemaResolver(Map<Class<?>, Schema<?>> customTypeSchemaMappings) {
+		this.customSchemaMappings = Objects.requireNonNull(customTypeSchemaMappings);
+	}
+
+	@Override
+	public <T> Schema<T> getSchema(Class<?> messageClass, boolean returnDefault) {
+		Schema<?> schema = BASE_SCHEMA_MAPPINGS.get(messageClass);
+		if (schema == null) {
+			schema = getCustomSchemaOrMaybeDefault(messageClass, returnDefault);
+		}
+		return schema != null ? castToType(schema) : null;
+	}
+
+	@Nullable
+	private Schema<?> getCustomSchemaOrMaybeDefault(Class<?> messageClass, boolean returnDefault) {
+		return this.customSchemaMappings.getOrDefault(messageClass, (returnDefault ? Schema.BYTES : null));
+	}
+
+	@SuppressWarnings({ "unchecked" })
+	private <X> Schema<X> castToType(Schema<?> rawSchema) {
+		return (Schema<X>) rawSchema;
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package org.springframework.pulsar.core;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
@@ -58,6 +59,8 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
 	@Nullable
 	private final List<ProducerInterceptor> interceptors;
 
+	private final SchemaResolver schemaResolver;
+
 	@Nullable
 	private final ObservationRegistry observationRegistry;
 
@@ -70,37 +73,40 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
 	private Schema<T> schema;
 
 	/**
-	 * Construct a template instance.
+	 * Construct a template instance without interceptors that uses the default schema
+	 * resolver.
 	 * @param producerFactory the factory used to create the backing Pulsar producers.
 	 */
 	public PulsarTemplate(PulsarProducerFactory<T> producerFactory) {
-		this(producerFactory, null);
+		this(producerFactory, Collections.emptyList());
 	}
 
 	/**
-	 * Construct a template instance with optional interceptors.
+	 * Construct a template instance with interceptors that uses the default schema
+	 * resolver.
 	 * @param producerFactory the factory used to create the backing Pulsar producers.
 	 * @param interceptors the interceptors to add to the producer.
 	 */
-	public PulsarTemplate(PulsarProducerFactory<T> producerFactory, @Nullable List<ProducerInterceptor> interceptors) {
-		this(producerFactory, interceptors, null, null);
+	public PulsarTemplate(PulsarProducerFactory<T> producerFactory, List<ProducerInterceptor> interceptors) {
+		this(producerFactory, interceptors, new DefaultSchemaResolver(), null, null);
 	}
 
 	/**
-	 * Construct a template instance with optional interceptors and observation
-	 * configuration.
+	 * Construct a template instance with optional observation configuration.
 	 * @param producerFactory the factory used to create the backing Pulsar producers
-	 * @param interceptors the optional list of interceptors to add to the producer
+	 * @param interceptors the list of interceptors to add to the producer
+	 * @param schemaResolver the schema resolver to use
 	 * @param observationRegistry the registry to record observations with or {@code null}
 	 * to not record observations
 	 * @param observationConvention the optional custom observation convention to use when
 	 * recording observations
 	 */
-	public PulsarTemplate(PulsarProducerFactory<T> producerFactory, @Nullable List<ProducerInterceptor> interceptors,
-			@Nullable ObservationRegistry observationRegistry,
+	public PulsarTemplate(PulsarProducerFactory<T> producerFactory, List<ProducerInterceptor> interceptors,
+			SchemaResolver schemaResolver, @Nullable ObservationRegistry observationRegistry,
 			@Nullable PulsarTemplateObservationConvention observationConvention) {
 		this.producerFactory = producerFactory;
 		this.interceptors = interceptors;
+		this.schemaResolver = schemaResolver;
 		this.observationRegistry = observationRegistry;
 		this.observationConvention = observationConvention;
 	}
@@ -204,7 +210,7 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
 	private Producer<T> prepareProducerForSend(@Nullable String topic, T message,
 			@Nullable Collection<String> encryptionKeys, @Nullable ProducerBuilderCustomizer<T> producerCustomizer)
 			throws PulsarClientException {
-		Schema<T> schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message);
+		Schema<T> schema = this.schema != null ? this.schema : this.schemaResolver.getSchema(message);
 		List<ProducerBuilderCustomizer<T>> customizers = new ArrayList<>();
 		if (!CollectionUtils.isEmpty(this.interceptors)) {
 			customizers.add(builder -> this.interceptors.forEach(builder::intercept));

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

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2023-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.core;
+
+import org.apache.pulsar.client.api.Schema;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * Resolves schema to use for message types.
+ *
+ * @author Chris Bono
+ */
+public interface SchemaResolver {
+
+	/**
+	 * Get the schema to use for a particular message.
+	 * @param <T> the schema type
+	 * @param message the message instance
+	 * @return the schema to use or {@code null} if no schema could be resolved
+	 */
+	@Nullable
+	default <T> Schema<T> getSchema(T message) {
+		return getSchema(message.getClass());
+	}
+
+	/**
+	 * Get the schema to use for a message type.
+	 * @param <T> the schema type
+	 * @param messageType the message type
+	 * @return the schema to use or {@code null} if no schema could be resolved
+	 */
+	@Nullable
+	default <T> Schema<T> getSchema(Class<?> messageType) {
+		return getSchema(messageType, true);
+	}
+
+	/**
+	 * Get the schema to use for a message type.
+	 * @param <T> the schema type
+	 * @param messageType the message type
+	 * @param returnDefault whether to return default schema if no schema could be
+	 * resolved
+	 * @return the schema to use or the default schema if no schema could be resolved and
+	 * {@code returnDefault} is {@code true} - otherwise {@code null}
+	 */
+	@Nullable
+	<T> Schema<T> getSchema(Class<?> messageType, boolean returnDefault);
+
+}

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

@@ -1,62 +0,0 @@
-/*
- * Copyright 2022 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.pulsar.core;
-
-import org.apache.pulsar.client.api.Schema;
-
-/**
- * Utility class for Pulsar schema inference.
- *
- * @author Soby Chacko
- * @author Alexander Preuß
- */
-public final class SchemaUtils {
-
-	private SchemaUtils() {
-	}
-
-	public static <T> Schema<T> getSchema(T message) {
-		return getSchema(message.getClass());
-	}
-
-	public static <T> Schema<T> getSchema(Class<?> messageClass) {
-		return getSchema(messageClass, true);
-	}
-
-	@SuppressWarnings("unchecked")
-	public static <T> Schema<T> getSchema(Class<?> messageClass, boolean returnDefault) {
-		return switch (messageClass.getName()) {
-			case "java.lang.String" -> (Schema<T>) Schema.STRING;
-			case "[B" -> (Schema<T>) Schema.BYTES;
-			case "java.lang.Byte", "byte" -> (Schema<T>) Schema.INT8;
-			case "java.lang.Short", "short" -> (Schema<T>) Schema.INT16;
-			case "java.lang.Integer", "int" -> (Schema<T>) Schema.INT32;
-			case "java.lang.Long", "long" -> (Schema<T>) Schema.INT64;
-			case "java.lang.Boolean", "boolean" -> (Schema<T>) Schema.BOOL;
-			case "java.nio.ByteBuffer" -> (Schema<T>) Schema.BYTEBUFFER;
-			case "java.util.Date" -> (Schema<T>) Schema.DATE;
-			case "java.lang.Double", "double" -> (Schema<T>) Schema.DOUBLE;
-			case "java.lang.Float", "float" -> (Schema<T>) Schema.FLOAT;
-			case "java.time.Instant" -> (Schema<T>) Schema.INSTANT;
-			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 -> (returnDefault ? (Schema<T>) Schema.BYTES : null);
-		};
-	}
-
-}

spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,6 +24,8 @@ import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.common.schema.SchemaType;
 
 import org.springframework.core.task.AsyncTaskExecutor;
+import org.springframework.pulsar.core.DefaultSchemaResolver;
+import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.observation.PulsarListenerObservationConvention;
 import org.springframework.util.Assert;
 
@@ -32,6 +34,7 @@ import org.springframework.util.Assert;
  *
  * @author Soby Chacko
  * @author Alexander Preuß
+ * @author Chris Bono
  */
 public class PulsarContainerProperties {
 
@@ -51,6 +54,8 @@ public class PulsarContainerProperties {
 
 	private SchemaType schemaType;
 
+	private SchemaResolver schemaResolver;
+
 	private Object messageListener;
 
 	private AsyncTaskExecutor consumerTaskExecutor;
@@ -72,11 +77,13 @@ public class PulsarContainerProperties {
 	public PulsarContainerProperties(String... topics) {
 		this.topics = topics.clone();
 		this.topicsPattern = null;
+		this.schemaResolver = new DefaultSchemaResolver();
 	}
 
 	public PulsarContainerProperties(String topicPattern) {
 		this.topicsPattern = topicPattern;
 		this.topics = null;
+		this.schemaResolver = new DefaultSchemaResolver();
 	}
 
 	public Object getMessageListener() {
@@ -169,14 +176,6 @@ public class PulsarContainerProperties {
 		this.consumerStartTimeout = consumerStartTimeout;
 	}
 
-	public Schema<?> getSchema() {
-		return this.schema;
-	}
-
-	public void setSchema(Schema<?> schema) {
-		this.schema = schema;
-	}
-
 	public String[] getTopics() {
 		return this.topics;
 	}
@@ -201,6 +200,14 @@ public class PulsarContainerProperties {
 		this.subscriptionName = subscriptionName;
 	}
 
+	public Schema<?> getSchema() {
+		return this.schema;
+	}
+
+	public void setSchema(Schema<?> schema) {
+		this.schema = schema;
+	}
+
 	public SchemaType getSchemaType() {
 		return this.schemaType;
 	}
@@ -209,6 +216,14 @@ public class PulsarContainerProperties {
 		this.schemaType = schemaType;
 	}
 
+	public SchemaResolver getSchemaResolver() {
+		return this.schemaResolver;
+	}
+
+	public void setSchemaResolver(SchemaResolver schemaResolver) {
+		this.schemaResolver = schemaResolver;
+	}
+
 	public Properties getPulsarConsumerProperties() {
 		return this.pulsarConsumerProperties;
 	}

spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2023-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.core;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import org.apache.pulsar.client.api.Schema;
+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;
+
+/**
+ * Unit tests for {@link DefaultSchemaResolver}.
+ *
+ * @author Chris Bono
+ */
+class DefaultSchemaResolverTests {
+
+	@ParameterizedTest
+	@MethodSource("schemaByMessageInstanceProvider")
+	<T> void getSchemaByMessageInstance(T message, Schema<T> expectedSchema) {
+		DefaultSchemaResolver resolver = new DefaultSchemaResolver();
+		assertThat(resolver.getSchema(message)).isEqualTo(expectedSchema);
+	}
+
+	static Stream<Arguments> schemaByMessageInstanceProvider() {
+		// @formatter:off
+		return Stream.of(
+				arguments("foo", Schema.STRING),
+				arguments("foo".getBytes(), Schema.BYTES),
+				arguments(Byte.valueOf("0"), Schema.INT8),
+				arguments((byte) 0, Schema.INT8),
+				arguments(Short.valueOf("0"), Schema.INT16),
+				arguments((short) 0, Schema.INT16),
+				arguments(Integer.valueOf("0"), Schema.INT32),
+				arguments(0, Schema.INT32),
+				arguments(Long.valueOf("0"), Schema.INT64),
+				arguments(0L, Schema.INT64),
+				arguments(Boolean.TRUE, Schema.BOOL),
+				arguments(ByteBuffer.wrap("foo".getBytes()), Schema.BYTEBUFFER),
+				arguments(ByteBuffer.allocateDirect(10), Schema.BYTEBUFFER),
+				arguments(new Date(), Schema.DATE),
+				arguments(Double.valueOf("2.2"), Schema.DOUBLE),
+				arguments(2.3d, Schema.DOUBLE),
+				arguments(Float.valueOf("2.4"), Schema.FLOAT),
+				arguments(2.5f, Schema.FLOAT),
+				arguments(Instant.now(), Schema.INSTANT),
+				arguments(LocalDate.now(), Schema.LOCAL_DATE),
+				arguments(LocalDateTime.now(), Schema.LOCAL_DATE_TIME),
+				arguments(LocalTime.NOON, Schema.LOCAL_TIME)
+		);
+		// @formatter:on
+	}
+
+	@Test
+	void getSchemaByMessageInstanceCustomTypes() {
+		Schema<?> fooSchema = Schema.AVRO(Foo.class);
+		Map<Class<?>, Schema<?>> customTypes = new HashMap<>();
+		customTypes.put(Foo.class, fooSchema);
+		customTypes.put(Bar.class, Schema.STRING);
+		DefaultSchemaResolver resolver = new DefaultSchemaResolver(customTypes);
+		assertThat(resolver.getSchema(new Foo("foo1"))).isSameAs(fooSchema);
+		assertThat(resolver.getSchema(new Bar<>("bar1"))).isEqualTo(Schema.STRING);
+		assertThat(resolver.getSchema(new Zaa("zaa1"))).isEqualTo(Schema.BYTES); // default
+	}
+
+	@ParameterizedTest
+	@MethodSource("schemaByMessageTypeProvider")
+	<T> void getSchemaByMessageType(Class<?> messageType, Schema<T> expectedSchema) {
+		DefaultSchemaResolver resolver = new DefaultSchemaResolver();
+		assertThat(resolver.getSchema(messageType)).isEqualTo(expectedSchema);
+	}
+
+	static Stream<Arguments> schemaByMessageTypeProvider() {
+		// @formatter:off
+		return Stream.of(
+				arguments(String.class, Schema.STRING),
+				arguments(byte[].class, Schema.BYTES),
+				arguments(Byte.class, Schema.INT8),
+				arguments(byte.class, Schema.INT8),
+				arguments(Short.class, Schema.INT16),
+				arguments(short.class, Schema.INT16),
+				arguments(Integer.class, Schema.INT32),
+				arguments(int.class, Schema.INT32),
+				arguments(Long.class, Schema.INT64),
+				arguments(long.class, Schema.INT64),
+				arguments(Boolean.class, Schema.BOOL),
+				arguments(boolean.class, Schema.BOOL),
+				arguments(ByteBuffer.class, Schema.BYTEBUFFER),
+				arguments(ByteBuffer.wrap("foo".getBytes()).getClass(), Schema.BYTEBUFFER),
+				arguments(ByteBuffer.allocateDirect(10).getClass(), Schema.BYTEBUFFER),
+				arguments(Date.class, Schema.DATE),
+				arguments(Double.class, Schema.DOUBLE),
+				arguments(double.class, Schema.DOUBLE),
+				arguments(Float.class, Schema.FLOAT),
+				arguments(float.class, Schema.FLOAT),
+				arguments(Instant.class, Schema.INSTANT),
+				arguments(LocalDate.class, Schema.LOCAL_DATE),
+				arguments(LocalDateTime.class, Schema.LOCAL_DATE_TIME),
+				arguments(LocalTime.class, Schema.LOCAL_TIME)
+		);
+		// @formatter:on
+	}
+
+	@Test
+	void getSchemaByMessageTypeCustomTypes() {
+		Schema<?> fooSchema = Schema.AVRO(Foo.class);
+		Map<Class<?>, Schema<?>> customTypes = new HashMap<>();
+		customTypes.put(Foo.class, fooSchema);
+		customTypes.put(Bar.class, Schema.STRING);
+		DefaultSchemaResolver resolver = new DefaultSchemaResolver(customTypes);
+		assertThat(resolver.getSchema(Foo.class)).isSameAs(fooSchema);
+		assertThat(resolver.getSchema(Bar.class)).isEqualTo(Schema.STRING);
+		assertThat(resolver.getSchema(Zaa.class)).isEqualTo(Schema.BYTES); // default
+	}
+
+	@Test
+	void getSchemaByMessageTypeWithoutDefaults() {
+		DefaultSchemaResolver resolver = new DefaultSchemaResolver();
+		assertThat(resolver.getSchema(Foo.class, false)).isNull();
+
+		resolver = new DefaultSchemaResolver(Collections.singletonMap(Foo.class, Schema.STRING));
+		assertThat(resolver.getSchema(Foo.class, false)).isEqualTo(Schema.STRING);
+		assertThat(resolver.getSchema(Bar.class, false)).isNull();
+	}
+
+	record Foo(String value) {
+	}
+
+	record Bar<T> (T value) {
+	}
+
+	record Zaa(String value) {
+	}
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -89,7 +89,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
 		try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
 				.build()) {
 			try (Consumer<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic)
-					.subscriptionName("test-specific-schema-subscription").subscribe()) {
+					.subscriptionName("smt-specific-schema-subscription").subscribe()) {
 				PulsarProducerFactory<Foo> producerFactory = new DefaultPulsarProducerFactory<>(client,
 						Collections.singletonMap("topicName", topic));
 				PulsarTemplate<Foo> pulsarTemplate = new PulsarTemplate<>(producerFactory);
@@ -102,6 +102,29 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
 		}
 	}
 
+	@Test
+	void sendMessageWithSpecificSchemaAndCustomTypeMappings() throws Exception {
+		String topic = "smt-specific-schema-custom-topic";
+		try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
+				.build()) {
+			try (Consumer<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic)
+					.subscriptionName("smt-specific-schema-custom-subscription").subscribe()) {
+				PulsarProducerFactory<Foo> producerFactory = new DefaultPulsarProducerFactory<>(client,
+						Collections.singletonMap("topicName", topic));
+				// Custom schema resolver allows not calling setSchema on template
+				DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(
+						Collections.singletonMap(Foo.class, Schema.JSON(Foo.class)));
+				PulsarTemplate<Foo> pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(),
+						schemaResolver, null, null);
+
+				Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID());
+				pulsarTemplate.send(foo);
+				assertThat(consumer.receiveAsync()).succeedsWithin(Duration.ofSeconds(3)).extracting(Message::getValue)
+						.isEqualTo(foo);
+			}
+		}
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	void sendMessageWithEncryptionKeys() throws Exception {

spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -64,12 +64,14 @@ import org.springframework.pulsar.config.PulsarListenerContainerFactory;
 import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.PulsarAdministration;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
 import org.springframework.pulsar.core.PulsarTestContainerSupport;
 import org.springframework.pulsar.core.PulsarTopic;
+import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.support.PulsarHeaders;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
@@ -560,59 +562,179 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 		}
 
-		static class User {
+	}
 
-			private String name;
+	/**
+	 * Do not convert this to a Record as Avro does not seem to work well w/ records.
+	 */
+	static class User {
 
-			private int age;
+		private String name;
 
-			User() {
+		private int age;
 
+		User() {
+		}
+
+		User(String name, int age) {
+			this.name = name;
+			this.age = age;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		public int getAge() {
+			return age;
+		}
+
+		public void setAge(int age) {
+			this.age = age;
+		}
+
+		@Override
+		public boolean equals(Object o) {
+			if (this == o) {
+				return true;
+			}
+			if (o == null || getClass() != o.getClass()) {
+				return false;
+			}
+			User user = (User) o;
+			return age == user.age && Objects.equals(name, user.name);
+		}
+
+		@Override
+		public int hashCode() {
+			return Objects.hash(name, age);
+		}
+
+		@Override
+		public String toString() {
+			return "User{" + "name='" + name + '\'' + ", age=" + age + '}';
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.class)
+	class SchemaCustomMappingsTestCases {
+
+		static CountDownLatch jsonLatch = new CountDownLatch(3);
+		static CountDownLatch avroLatch = new CountDownLatch(3);
+		static CountDownLatch keyvalueLatch = new CountDownLatch(3);
+		static CountDownLatch protobufLatch = new CountDownLatch(3);
+
+		@Test
+		void jsonSchema() throws Exception {
+			PulsarProducerFactory<User2> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+					Collections.emptyMap());
+			PulsarTemplate<User2> template = new PulsarTemplate<>(pulsarProducerFactory);
+			template.setSchema(Schema.JSON(User2.class));
+
+			for (int i = 0; i < 3; i++) {
+				template.send("json-custom-mappings-topic", new User2("Jason", i));
+			}
+			assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+
+		@Test
+		void avroSchema() throws Exception {
+			PulsarProducerFactory<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+					Collections.emptyMap());
+			PulsarTemplate<User> template = new PulsarTemplate<>(pulsarProducerFactory);
+			template.setSchema(AvroSchema.of(User.class));
+
+			for (int i = 0; i < 3; i++) {
+				template.send("avro-custom-mappings-topic", new User("Avi", i));
+			}
+			assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+
+		@Test
+		void keyvalueSchema() throws Exception {
+			PulsarProducerFactory<KeyValue<String, User2>> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
+					pulsarClient, Collections.emptyMap());
+			PulsarTemplate<KeyValue<String, User2>> template = new PulsarTemplate<>(pulsarProducerFactory);
+			Schema<KeyValue<String, User2>> kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class),
+					KeyValueEncodingType.INLINE);
+			template.setSchema(kvSchema);
+
+			for (int i = 0; i < 3; i++) {
+				template.send("keyvalue-custom-mappings-topic", new KeyValue<>("Kevin", new User2("Kevin", 5150)));
+			}
+			assertThat(keyvalueLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+
+		@Test
+		void protobufSchema() throws Exception {
+			PulsarProducerFactory<Proto.Person> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+					Collections.emptyMap());
+			PulsarTemplate<Proto.Person> template = new PulsarTemplate<>(pulsarProducerFactory);
+			template.setSchema(ProtobufSchema.of(Proto.Person.class));
+
+			for (int i = 0; i < 3; i++) {
+				template.send("protobuf-custom-mappings-topic",
+						Proto.Person.newBuilder().setId(i).setName("Paul").build());
+			}
+			assertThat(protobufLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		}
+
+		@EnablePulsar
+		@Configuration
+		static class SchemaCustomMappingsTestConfig {
+
+			@Bean
+			SchemaResolver customSchemaResolver() {
+				Map<Class<?>, Schema<?>> customMappings = new HashMap<>();
+				customMappings.put(User.class, Schema.AVRO(User.class));
+				customMappings.put(User2.class, Schema.JSON(User2.class));
+				customMappings.put(Proto.Person.class, Schema.PROTOBUF(Proto.Person.class));
+				return new DefaultSchemaResolver(customMappings);
 			}
 
-			User(String name, int age) {
-				this.name = name;
-				this.age = age;
+			@Bean
+			PulsarListenerContainerFactory pulsarListenerContainerFactory(
+					PulsarConsumerFactory<Object> pulsarConsumerFactory, SchemaResolver schemaResolver) {
+				PulsarContainerProperties containerProps = new PulsarContainerProperties();
+				containerProps.setSchemaResolver(schemaResolver);
+				ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>(
+						pulsarConsumerFactory, containerProps, null);
+				return pulsarListenerContainerFactory;
 			}
 
-			public String getName() {
-				return name;
+			@PulsarListener(id = "jsonListener", topics = "json-custom-mappings-topic",
+					subscriptionName = "subscription-4", properties = { "subscriptionInitialPosition=Earliest" })
+			void listenJson(User2 message) {
+				jsonLatch.countDown();
 			}
 
-			public void setName(String name) {
-				this.name = name;
+			@PulsarListener(id = "avroListener", topics = "avro-custom-mappings-topic",
+					subscriptionName = "subscription-6", properties = { "subscriptionInitialPosition=Earliest" })
+			void listenAvro(User message) {
+				avroLatch.countDown();
 			}
 
-			public int getAge() {
-				return age;
+			@PulsarListener(id = "keyvalueListener", topics = "keyvalue-custom-mappings-topic",
+					subscriptionName = "subscription-8", properties = { "subscriptionInitialPosition=Earliest" })
+			void listenKeyvalue(KeyValue<String, User2> message) {
+				keyvalueLatch.countDown();
 			}
 
-			public void setAge(int age) {
-				this.age = age;
+			@PulsarListener(id = "protobufListener", topics = "protobuf-custom-mappings-topic",
+					subscriptionName = "subscription-10", properties = { "subscriptionInitialPosition=Earliest" })
+			void listenProtobuf(Proto.Person message) {
+				protobufLatch.countDown();
 			}
 
-			@Override
-			public boolean equals(Object o) {
-				if (this == o) {
-					return true;
-				}
-				if (o == null || getClass() != o.getClass()) {
-					return false;
-				}
-				User user = (User) o;
-				return age == user.age && Objects.equals(name, user.name);
-			}
-
-			@Override
-			public int hashCode() {
-				return Objects.hash(name, age);
-			}
-
-			@Override
-			public String toString() {
-				return "User{" + "name='" + name + '\'' + ", age=" + age + '}';
-			}
+		}
 
+		record User2(String name, int age) {
 		}
 
 	}

spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -40,6 +40,7 @@ import org.springframework.pulsar.config.PulsarClientFactoryBean;
 import org.springframework.pulsar.config.PulsarListenerContainerFactory;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.PulsarAdministration;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
@@ -146,7 +147,8 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul
 		@Bean
 		public PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory,
 				ObservationRegistry observationRegistry) {
-			return new PulsarTemplate<>(pulsarProducerFactory, null, observationRegistry, null);
+			return new PulsarTemplate<>(pulsarProducerFactory, null, new DefaultSchemaResolver(), observationRegistry,
+					null);
 		}
 
 		@Bean

spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -45,6 +45,7 @@ import org.springframework.pulsar.config.PulsarClientFactoryBean;
 import org.springframework.pulsar.config.PulsarListenerContainerFactory;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.PulsarAdministration;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
@@ -186,7 +187,7 @@ public class ObservationTests implements PulsarTestContainerSupport {
 		@Bean(name = "observationTestsTemplate")
 		PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory,
 				ObservationRegistry observationRegistry) {
-			return new PulsarTemplate<>(pulsarProducerFactory, null, observationRegistry,
+			return new PulsarTemplate<>(pulsarProducerFactory, null, new DefaultSchemaResolver(), observationRegistry,
 					new DefaultPulsarTemplateObservationConvention() {
 						@Override
 						public KeyValues getLowCardinalityKeyValues(PulsarMessageSenderContext context) {

src/checkstyle/checkstyle-suppressions.xml

@@ -7,6 +7,7 @@
 	<suppress files="[\\/]test[\\/]" checks="RequireThis"/>
 	<suppress files="[\\/]test[\\/]" checks="Javadoc*"/>
 	<suppress files="PulsarFunctionAdministrationIntegrationTests" checks="Regexp"/>
+	<suppress files="DefaultSchemaResolverTests" checks="AvoidStaticImport|MethodParamPad" />
 	<suppress files="Proto" checks=".*"/>
 	<suppress files="ReactiveSpringPulsarBootApp" checks="HideUtilityClassConstructor"/>
 	<suppress files="[\\/]spring-pulsar-docs[\\/]" checks="JavadocPackage|JavadocType|JavadocVariable|SpringDeprecatedCheck" />