Add support for AUTO_CONSUME schema type

See #380

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 the original author or authors.
+ * Copyright 2022-2024 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,11 +26,11 @@ import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.DeadLetterPolicy;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.Messages;
-import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.common.schema.SchemaType;
 
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.expression.BeanResolver;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.converter.SmartMessageConverter;
@@ -65,6 +65,8 @@ import reactor.core.publisher.Flux;
  */
 public class MethodReactivePulsarListenerEndpoint<V> extends AbstractReactivePulsarListenerEndpoint<V> {
 
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
 	private Object bean;
 
 	private Method method;
@@ -134,18 +136,29 @@ public class MethodReactivePulsarListenerEndpoint<V> extends AbstractReactivePul
 			messageParameter = parameter.get();
 		}
 
-		DefaultReactivePulsarMessageListenerContainer<?> containerInstance = (DefaultReactivePulsarMessageListenerContainer<?>) container;
-		ReactivePulsarContainerProperties<?> pulsarContainerProperties = containerInstance.getContainerProperties();
+		DefaultReactivePulsarMessageListenerContainer<Object> containerInstance = (DefaultReactivePulsarMessageListenerContainer<Object>) container;
+		ReactivePulsarContainerProperties<Object> pulsarContainerProperties = containerInstance
+			.getContainerProperties();
+
+		// Resolve the schema using the reader schema type
 		SchemaResolver schemaResolver = pulsarContainerProperties.getSchemaResolver();
 		SchemaType schemaType = pulsarContainerProperties.getSchemaType();
 		ResolvableType messageType = resolvableType(messageParameter);
 		schemaResolver.resolveSchema(schemaType, messageType)
-			.ifResolved(schema -> pulsarContainerProperties.setSchema((Schema) schema));
+			.ifResolvedOrElse(pulsarContainerProperties::setSchema,
+					(ex) -> this.logger
+						.warn(() -> "Failed to resolve schema for type %s - will default to BYTES (due to: %s)"
+							.formatted(schemaType, ex.getMessage())));
 
-		// Make sure the schemaType is updated to match the current schema
+		// Attempt to make sure the schemaType is updated to match the resolved schema.
+		// This can occur when the resolver returns a schema that is not necessarily of
+		// the same type as the input scheme type (e.g. SchemaType.NONE uses the message
+		// type to determine the schema.
 		if (pulsarContainerProperties.getSchema() != null) {
-			SchemaType type = pulsarContainerProperties.getSchema().getSchemaInfo().getType();
-			pulsarContainerProperties.setSchemaType(type);
+			var schemaInfo = pulsarContainerProperties.getSchema().getSchemaInfo();
+			if (schemaInfo != null) {
+				pulsarContainerProperties.setSchemaType(schemaInfo.getType());
+			}
 		}
 
 		// If no topic info is set on endpoint attempt to resolve via message type

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

@@ -0,0 +1,205 @@
+/*
+ * Copyright 2022-2024 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.reactive.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.schema.Field;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.impl.schema.AvroSchema;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroRecord;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonRecord;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.KeyValueEncodingType;
+import org.apache.pulsar.common.schema.SchemaType;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.annotation.EnablePulsar;
+import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener;
+import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListenerMessageConsumerBuilderCustomizer;
+import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerAutoConsumeSchemaTests.ReactivePulsarListenerAutoConsumeSchemaTestsConfig;
+import org.springframework.pulsar.test.support.model.UserPojo;
+import org.springframework.pulsar.test.support.model.UserRecord;
+import org.springframework.test.context.ContextConfiguration;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * Tests for {@link ReactivePulsarListener @ReactivePulsarListener} using
+ * {@code schemaType} of {@link SchemaType#AUTO_CONSUME}.
+ *
+ * @author Chris Bono
+ */
+@ContextConfiguration(classes = ReactivePulsarListenerAutoConsumeSchemaTestsConfig.class)
+class ReactivePulsarListenerAutoConsumeSchemaTests extends ReactivePulsarListenerTestsBase {
+
+	static final String STRING_TOPIC = "placst-str-topic";
+	static CountDownLatch stringLatch = new CountDownLatch(3);
+	static List<String> stringMessages = new ArrayList<>();
+
+	static final String JSON_TOPIC = "placst-json-topic";
+	static CountDownLatch jsonLatch = new CountDownLatch(3);
+	static List<Map<String, Object>> jsonMessages = new ArrayList<>();
+
+	static final String AVRO_TOPIC = "placst-avro-topic";
+	static CountDownLatch avroLatch = new CountDownLatch(3);
+	static List<Map<String, Object>> avroMessages = new ArrayList<>();
+
+	static final String KEYVALUE_TOPIC = "placst-kv-topic";
+	static CountDownLatch keyValueLatch = new CountDownLatch(3);
+	static List<Map<String, Object>> keyValueMessages = new ArrayList<>();
+
+	@Test
+	void stringSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<String>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var expectedMessages = new ArrayList<String>();
+		for (int i = 0; i < 3; i++) {
+			var msg = "str-" + i;
+			template.send(STRING_TOPIC, msg, Schema.STRING);
+			expectedMessages.add(msg);
+		}
+		assertThat(stringLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(stringMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@Test
+	void jsonSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserRecord>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var schema = JSONSchema.of(UserRecord.class);
+		var expectedMessages = new ArrayList<Map<String, Object>>();
+		for (int i = 0; i < 3; i++) {
+			var user = new UserRecord("Jason", i);
+			template.send(JSON_TOPIC, user, schema);
+			expectedMessages.add(Map.of("name", user.name(), "age", user.age()));
+		}
+		assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(jsonMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@Test
+	void avroSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserPojo>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var schema = AvroSchema.of(UserPojo.class);
+		var expectedMessages = new ArrayList<Map<String, Object>>();
+		for (int i = 0; i < 3; i++) {
+			var user = new UserPojo("Avi", i);
+			template.send(AVRO_TOPIC, user, schema);
+			expectedMessages.add(Map.of("name", user.getName(), "age", user.getAge()));
+		}
+		assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(avroMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@Test
+	void keyValueSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<KeyValue<String, Integer>>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32, KeyValueEncodingType.INLINE);
+		var expectedMessages = new ArrayList<Map<String, Object>>();
+		for (int i = 0; i < 3; i++) {
+			var kv = new KeyValue<>("Kevin", i);
+			template.send(KEYVALUE_TOPIC, kv, kvSchema);
+			expectedMessages.add(Map.of(kv.getKey(), kv.getValue()));
+		}
+		assertThat(keyValueLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(keyValueMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@EnablePulsar
+	@Configuration
+	static class ReactivePulsarListenerAutoConsumeSchemaTestsConfig {
+
+		@ReactivePulsarListener(id = "stringAcListener", topics = STRING_TOPIC, schemaType = SchemaType.AUTO_CONSUME,
+				consumerCustomizer = "earliestCustomizer")
+		Mono<Void> listenString(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue().getNativeObject()).isInstanceOf(String.class);
+			stringMessages.add(genericMessage.getValue().getNativeObject().toString());
+			stringLatch.countDown();
+			return Mono.empty();
+		}
+
+		@ReactivePulsarListener(id = "jsonAcListener", topics = JSON_TOPIC, schemaType = SchemaType.AUTO_CONSUME,
+				consumerCustomizer = "earliestCustomizer")
+		Mono<Void> listenJson(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue()).isInstanceOf(GenericJsonRecord.class);
+			GenericJsonRecord record = (GenericJsonRecord) genericMessage.getValue();
+			assertThat(record.getSchemaType()).isEqualTo(SchemaType.JSON);
+			assertThat(record).extracting("schemaInfo")
+				.satisfies((obj) -> assertThat(obj.toString()).contains("\"name\": \"UserRecord\""));
+			jsonMessages.add(record.getFields()
+				.stream()
+				.map(Field::getName)
+				.collect(Collectors.toMap(Function.identity(), record::getField)));
+			jsonLatch.countDown();
+			return Mono.empty();
+		}
+
+		@ReactivePulsarListener(id = "avroAcListener", topics = AVRO_TOPIC, schemaType = SchemaType.AUTO_CONSUME,
+				consumerCustomizer = "earliestCustomizer")
+		Mono<Void> listenAvro(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue()).isInstanceOf(GenericAvroRecord.class);
+			GenericAvroRecord record = (GenericAvroRecord) genericMessage.getValue();
+			assertThat(record.getSchemaType()).isEqualTo(SchemaType.AVRO);
+			assertThat(record).extracting("schema")
+				.satisfies((obj) -> assertThat(obj.toString()).contains("\"name\":\"UserPojo\""));
+			avroMessages.add(record.getFields()
+				.stream()
+				.map(Field::getName)
+				.collect(Collectors.toMap(Function.identity(), record::getField)));
+			avroLatch.countDown();
+			return Mono.empty();
+		}
+
+		@SuppressWarnings("unchecked")
+		@ReactivePulsarListener(id = "keyvalueAcListener", topics = KEYVALUE_TOPIC,
+				schemaType = SchemaType.AUTO_CONSUME, consumerCustomizer = "earliestCustomizer")
+		Mono<Void> listenKeyvalue(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue().getSchemaType()).isEqualTo(SchemaType.KEY_VALUE);
+			assertThat(genericMessage.getValue().getNativeObject()).isInstanceOf(KeyValue.class);
+			var record = (KeyValue<String, Object>) genericMessage.getValue().getNativeObject();
+			keyValueMessages.add(Map.of(record.getKey(), record.getValue()));
+			keyValueLatch.countDown();
+			return Mono.empty();
+		}
+
+		@Bean
+		ReactivePulsarListenerMessageConsumerBuilderCustomizer<?> earliestCustomizer() {
+			return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+		}
+
+	}
+
+}

spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/model/UserPojo.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2022-2024 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.test.support.model;
+
+import java.util.Objects;
+
+/**
+ * Test object (user) defined via standard Java beans get/set methods.
+ * <p>
+ * <b>WARN</b> Do not convert this to a Record as this is used for Avro tests and Avro
+ * does not work well w/ records yet.
+ */
+public class UserPojo {
+
+	private String name;
+
+	private int age;
+
+	UserPojo() {
+	}
+
+	public UserPojo(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;
+		}
+		UserPojo user = (UserPojo) 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 + '}';
+	}
+
+}

spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/model/UserRecord.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2022-2024 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.test.support.model;
+
+/**
+ * Test object (user) defined via a Java record.
+ *
+ * @param name the user's name
+ * @param age the user's age
+ */
+public record UserRecord(String name, int age) {
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 the original author or authors.
+ * Copyright 2022-2024 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.common.schema.SchemaType;
 
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.expression.BeanResolver;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.converter.SmartMessageConverter;
@@ -66,6 +67,8 @@ import org.springframework.util.StringUtils;
  */
 public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpoint<V> {
 
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
 	private Object bean;
 
 	private Method method;
@@ -140,15 +143,26 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
 
 		ConcurrentPulsarMessageListenerContainer<?> containerInstance = (ConcurrentPulsarMessageListenerContainer<?>) container;
 		PulsarContainerProperties pulsarContainerProperties = containerInstance.getContainerProperties();
+
+		// Resolve the schema using the listener schema type
 		SchemaResolver schemaResolver = pulsarContainerProperties.getSchemaResolver();
 		SchemaType schemaType = pulsarContainerProperties.getSchemaType();
 		ResolvableType messageType = resolvableType(messageParameter);
-		schemaResolver.resolveSchema(schemaType, messageType).ifResolved(pulsarContainerProperties::setSchema);
+		schemaResolver.resolveSchema(schemaType, messageType)
+			.ifResolvedOrElse(pulsarContainerProperties::setSchema,
+					(ex) -> this.logger
+						.warn(() -> "Failed to resolve schema for type %s - will default to BYTES (due to: %s)"
+							.formatted(schemaType, ex.getMessage())));
 
-		// Make sure the schemaType is updated to match the current schema
+		// Attempt to make sure the schemaType is updated to match the resolved schema.
+		// This can occur when the resolver returns a schema that is not necessarily of
+		// the same type as the input scheme type (e.g. SchemaType.NONE uses the message
+		// type to determine the schema.
 		if (pulsarContainerProperties.getSchema() != null) {
-			SchemaType type = pulsarContainerProperties.getSchema().getSchemaInfo().getType();
-			pulsarContainerProperties.setSchemaType(type);
+			var schemaInfo = pulsarContainerProperties.getSchema().getSchemaInfo();
+			if (schemaInfo != null) {
+				pulsarContainerProperties.setSchemaType(schemaInfo.getType());
+			}
 		}
 
 		// If no topic info is set on endpoint attempt to resolve via message type

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 the original author or authors.
+ * Copyright 2024 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.apache.pulsar.common.schema.SchemaType;
 
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.expression.BeanResolver;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.converter.SmartMessageConverter;
@@ -56,6 +57,8 @@ import org.springframework.util.Assert;
  */
 public class MethodPulsarReaderEndpoint<V> extends AbstractPulsarReaderEndpoint<V> {
 
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
 	private Object bean;
 
 	private Method method;
@@ -114,21 +117,30 @@ public class MethodPulsarReaderEndpoint<V> extends AbstractPulsarReaderEndpoint<
 
 		DefaultPulsarMessageReaderContainer<?> containerInstance = (DefaultPulsarMessageReaderContainer<?>) container;
 		PulsarReaderContainerProperties pulsarContainerProperties = containerInstance.getContainerProperties();
+
+		// Resolve the schema using the reader schema type
 		SchemaResolver schemaResolver = pulsarContainerProperties.getSchemaResolver();
 		SchemaType schemaType = pulsarContainerProperties.getSchemaType();
 		ResolvableType messageType = resolvableType(messageParameter);
-		schemaResolver.resolveSchema(schemaType, messageType).ifResolved(pulsarContainerProperties::setSchema);
+		schemaResolver.resolveSchema(schemaType, messageType)
+			.ifResolvedOrElse(pulsarContainerProperties::setSchema,
+					(ex) -> this.logger
+						.warn(() -> "Failed to resolve schema for type %s - will default to BYTES (due to: %s)"
+							.formatted(schemaType, ex.getMessage())));
 
-		// Make sure the schemaType is updated to match the current schema
+		// Attempt to make sure the schemaType is updated to match the resolved schema.
+		// This can occur when the resolver returns a schema that is not necessarily of
+		// the same type as the input scheme type (e.g. SchemaType.NONE uses the message
+		// type to determine the schema.
 		if (pulsarContainerProperties.getSchema() != null) {
-			SchemaType type = pulsarContainerProperties.getSchema().getSchemaInfo().getType();
-			pulsarContainerProperties.setSchemaType(type);
+			var schemaInfo = pulsarContainerProperties.getSchema().getSchemaInfo();
+			if (schemaInfo != null) {
+				pulsarContainerProperties.setSchemaType(schemaInfo.getType());
+			}
 		}
 
 		// TODO: If no topic info is set on endpoint attempt to resolve via message type
-
 		container.setReaderCustomizer(this.readerBuilderCustomizer);
-
 		return readerListener;
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 the original author or authors.
+ * Copyright 2022-2024 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.
@@ -185,6 +185,7 @@ public class DefaultSchemaResolver implements SchemaResolver {
 					requireNonNullMessageType(schemaType, messageType);
 					yield getMessageKeyValueSchema(messageType);
 				}
+				case AUTO_CONSUME -> Schema.AUTO_CONSUME();
 				case NONE -> {
 					if (messageType == null || messageType.getRawClass() == null) {
 						yield Schema.BYTES;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2023 the original author or authors.
+ * Copyright 2023-2024 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.
@@ -33,6 +33,7 @@ import java.util.stream.Stream;
 
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.schema.KeyValueSchema;
+import org.apache.pulsar.client.impl.schema.AutoConsumeSchema;
 import org.apache.pulsar.client.impl.schema.AvroSchema;
 import org.apache.pulsar.client.impl.schema.JSONSchema;
 import org.apache.pulsar.client.impl.schema.ProtobufSchema;
@@ -255,6 +256,17 @@ class DefaultSchemaResolverTests {
 				}));
 		}
 
+		@Test
+		void autoConsumeSchema() {
+			assertThat(resolver.resolveSchema(SchemaType.AUTO_CONSUME, ResolvableType.forType(Foo.class)).orElseThrow())
+				.isInstanceOf(AutoConsumeSchema.class);
+			assertThat(
+					resolver.resolveSchema(SchemaType.AUTO_CONSUME, ResolvableType.forType(String.class)).orElseThrow())
+				.isInstanceOf(AutoConsumeSchema.class);
+			assertThat(resolver.resolveSchema(SchemaType.AUTO_CONSUME, null).orElseThrow())
+				.isInstanceOf(AutoConsumeSchema.class);
+		}
+
 		@ParameterizedTest
 		@EnumSource(value = SchemaType.class, names = { "JSON", "AVRO", "PROTOBUF", "KEY_VALUE" })
 		void structSchemasRequireMessageType(SchemaType schemaType) {
@@ -264,7 +276,7 @@ class DefaultSchemaResolverTests {
 		}
 
 		@ParameterizedTest
-		@EnumSource(value = SchemaType.class, names = { "PROTOBUF_NATIVE", "AUTO", "AUTO_CONSUME", "AUTO_PUBLISH" })
+		@EnumSource(value = SchemaType.class, names = { "PROTOBUF_NATIVE", "AUTO", "AUTO_PUBLISH" })
 		void unsupportedSchemaTypes(SchemaType unsupportedType) {
 			assertThatExceptionOfType(IllegalArgumentException.class)
 				.isThrownBy(() -> resolver.resolveSchema(unsupportedType, null).orElseThrow())

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

@@ -0,0 +1,191 @@
+/*
+ * Copyright 2022-2024 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.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.Field;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.impl.schema.AvroSchema;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroRecord;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonRecord;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.KeyValueEncodingType;
+import org.apache.pulsar.common.schema.SchemaType;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.annotation.EnablePulsar;
+import org.springframework.pulsar.annotation.PulsarListener;
+import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.listener.PulsarListenerAutoConsumeSchemaTests.PulsarListenerAutoConsumeSchemaTestsConfig;
+import org.springframework.pulsar.test.support.model.UserPojo;
+import org.springframework.pulsar.test.support.model.UserRecord;
+import org.springframework.test.context.ContextConfiguration;
+
+/**
+ * Tests for {@link PulsarListener @PulsarListener} using {@code schemaType} of
+ * {@link SchemaType#AUTO_CONSUME}.
+ *
+ * @author Chris Bono
+ */
+@ContextConfiguration(classes = PulsarListenerAutoConsumeSchemaTestsConfig.class)
+class PulsarListenerAutoConsumeSchemaTests extends PulsarListenerTestsBase {
+
+	static final String STRING_TOPIC = "placst-str-topic";
+	static CountDownLatch stringLatch = new CountDownLatch(3);
+	static List<String> stringMessages = new ArrayList<>();
+
+	static final String JSON_TOPIC = "placst-json-topic";
+	static CountDownLatch jsonLatch = new CountDownLatch(3);
+	static List<Map<String, Object>> jsonMessages = new ArrayList<>();
+
+	static final String AVRO_TOPIC = "placst-avro-topic";
+	static CountDownLatch avroLatch = new CountDownLatch(3);
+	static List<Map<String, Object>> avroMessages = new ArrayList<>();
+
+	static final String KEYVALUE_TOPIC = "placst-kv-topic";
+	static CountDownLatch keyValueLatch = new CountDownLatch(3);
+	static List<Map<String, Object>> keyValueMessages = new ArrayList<>();
+
+	@Test
+	void stringSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<String>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var expectedMessages = new ArrayList<String>();
+		for (int i = 0; i < 3; i++) {
+			var msg = "str-" + i;
+			template.send(STRING_TOPIC, msg, Schema.STRING);
+			expectedMessages.add(msg);
+		}
+		assertThat(stringLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(stringMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@Test
+	void jsonSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserRecord>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var schema = JSONSchema.of(UserRecord.class);
+		var expectedMessages = new ArrayList<Map<String, Object>>();
+		for (int i = 0; i < 3; i++) {
+			var user = new UserRecord("Jason", i);
+			template.send(JSON_TOPIC, user, schema);
+			expectedMessages.add(Map.of("name", user.name(), "age", user.age()));
+		}
+		assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(jsonMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@Test
+	void avroSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserPojo>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var schema = AvroSchema.of(UserPojo.class);
+		var expectedMessages = new ArrayList<Map<String, Object>>();
+		for (int i = 0; i < 3; i++) {
+			var user = new UserPojo("Avi", i);
+			template.send(AVRO_TOPIC, user, schema);
+			expectedMessages.add(Map.of("name", user.getName(), "age", user.getAge()));
+		}
+		assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(avroMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@Test
+	void keyValueSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<KeyValue<String, Integer>>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32, KeyValueEncodingType.INLINE);
+		var expectedMessages = new ArrayList<Map<String, Object>>();
+		for (int i = 0; i < 3; i++) {
+			var kv = new KeyValue<>("Kevin", i);
+			template.send(KEYVALUE_TOPIC, kv, kvSchema);
+			expectedMessages.add(Map.of(kv.getKey(), kv.getValue()));
+		}
+		assertThat(keyValueLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(keyValueMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@EnablePulsar
+	@Configuration
+	static class PulsarListenerAutoConsumeSchemaTestsConfig {
+
+		@PulsarListener(id = "stringAcListener", topics = STRING_TOPIC, schemaType = SchemaType.AUTO_CONSUME,
+				properties = { "subscriptionInitialPosition=Earliest" })
+		void listenString(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue().getNativeObject()).isInstanceOf(String.class);
+			stringMessages.add(genericMessage.getValue().getNativeObject().toString());
+			stringLatch.countDown();
+		}
+
+		@PulsarListener(id = "jsonAcListener", topics = JSON_TOPIC, schemaType = SchemaType.AUTO_CONSUME,
+				properties = { "subscriptionInitialPosition=Earliest" })
+		void listenJson(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue()).isInstanceOf(GenericJsonRecord.class);
+			GenericJsonRecord record = (GenericJsonRecord) genericMessage.getValue();
+			assertThat(record.getSchemaType()).isEqualTo(SchemaType.JSON);
+			assertThat(record).extracting("schemaInfo")
+				.satisfies((obj) -> assertThat(obj.toString()).contains("\"name\": \"UserRecord\""));
+			jsonMessages.add(record.getFields()
+				.stream()
+				.map(Field::getName)
+				.collect(Collectors.toMap(Function.identity(), record::getField)));
+			jsonLatch.countDown();
+		}
+
+		@PulsarListener(id = "avroAcListener", topics = AVRO_TOPIC, schemaType = SchemaType.AUTO_CONSUME,
+				properties = { "subscriptionInitialPosition=Earliest" })
+		void listenAvro(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue()).isInstanceOf(GenericAvroRecord.class);
+			GenericAvroRecord record = (GenericAvroRecord) genericMessage.getValue();
+			assertThat(record.getSchemaType()).isEqualTo(SchemaType.AVRO);
+			assertThat(record).extracting("schema")
+				.satisfies((obj) -> assertThat(obj.toString()).contains("\"name\":\"UserPojo\""));
+			avroMessages.add(record.getFields()
+				.stream()
+				.map(Field::getName)
+				.collect(Collectors.toMap(Function.identity(), record::getField)));
+			avroLatch.countDown();
+		}
+
+		@SuppressWarnings("unchecked")
+		@PulsarListener(id = "keyvalueAcListener", topics = KEYVALUE_TOPIC, schemaType = SchemaType.AUTO_CONSUME,
+				properties = { "subscriptionInitialPosition=Earliest" })
+		void listenKeyvalue(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue().getSchemaType()).isEqualTo(SchemaType.KEY_VALUE);
+			assertThat(genericMessage.getValue().getNativeObject()).isInstanceOf(KeyValue.class);
+			var record = (KeyValue<String, Object>) genericMessage.getValue().getNativeObject();
+			keyValueMessages.add(Map.of(record.getKey(), record.getValue()));
+			keyValueLatch.countDown();
+		}
+
+	}
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderAutoConsumeSchemaTests.java

@@ -0,0 +1,191 @@
+/*
+ * Copyright 2022-2024 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.reader;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.Field;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.impl.schema.AvroSchema;
+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroRecord;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonRecord;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.KeyValueEncodingType;
+import org.apache.pulsar.common.schema.SchemaType;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.annotation.EnablePulsar;
+import org.springframework.pulsar.annotation.PulsarReader;
+import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.reader.PulsarReaderAutoConsumeSchemaTests.PulsarReaderAutoConsumeSchemaTestsConfig;
+import org.springframework.pulsar.test.support.model.UserPojo;
+import org.springframework.pulsar.test.support.model.UserRecord;
+import org.springframework.test.context.ContextConfiguration;
+
+/**
+ * Tests for {@link PulsarReader @PulsarReader} using {@code schemaType} of
+ * {@link SchemaType#AUTO_CONSUME}.
+ *
+ * @author Chris Bono
+ */
+@ContextConfiguration(classes = PulsarReaderAutoConsumeSchemaTestsConfig.class)
+class PulsarReaderAutoConsumeSchemaTests extends PulsarReaderTestsBase {
+
+	static final String STRING_TOPIC = "pracst-str-topic";
+	static CountDownLatch stringLatch = new CountDownLatch(3);
+	static List<String> stringMessages = new ArrayList<>();
+
+	static final String JSON_TOPIC = "pracst-json-topic";
+	static CountDownLatch jsonLatch = new CountDownLatch(3);
+	static List<Map<String, Object>> jsonMessages = new ArrayList<>();
+
+	static final String AVRO_TOPIC = "pracst-avro-topic";
+	static CountDownLatch avroLatch = new CountDownLatch(3);
+	static List<Map<String, Object>> avroMessages = new ArrayList<>();
+
+	static final String KEYVALUE_TOPIC = "pracst-kv-topic";
+	static CountDownLatch keyValueLatch = new CountDownLatch(3);
+	static List<Map<String, Object>> keyValueMessages = new ArrayList<>();
+
+	@Test
+	void stringSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<String>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var expectedMessages = new ArrayList<String>();
+		for (int i = 0; i < 3; i++) {
+			var msg = "str-" + i;
+			template.send(STRING_TOPIC, msg, Schema.STRING);
+			expectedMessages.add(msg);
+		}
+		assertThat(stringLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(stringMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@Test
+	void jsonSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserRecord>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var schema = JSONSchema.of(UserRecord.class);
+		var expectedMessages = new ArrayList<Map<String, Object>>();
+		for (int i = 0; i < 3; i++) {
+			var user = new UserRecord("Jason", i);
+			template.send(JSON_TOPIC, user, schema);
+			expectedMessages.add(Map.of("name", user.name(), "age", user.age()));
+		}
+		assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(jsonMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@Test
+	void avroSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserPojo>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var schema = AvroSchema.of(UserPojo.class);
+		var expectedMessages = new ArrayList<Map<String, Object>>();
+		for (int i = 0; i < 3; i++) {
+			var user = new UserPojo("Avi", i);
+			template.send(AVRO_TOPIC, user, schema);
+			expectedMessages.add(Map.of("name", user.getName(), "age", user.getAge()));
+		}
+		assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(avroMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@Test
+	void keyValueSchema() throws Exception {
+		var pulsarProducerFactory = new DefaultPulsarProducerFactory<KeyValue<String, Integer>>(pulsarClient);
+		var template = new PulsarTemplate<>(pulsarProducerFactory);
+		var kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32, KeyValueEncodingType.INLINE);
+		var expectedMessages = new ArrayList<Map<String, Object>>();
+		for (int i = 0; i < 3; i++) {
+			var kv = new KeyValue<>("Kevin", i);
+			template.send(KEYVALUE_TOPIC, kv, kvSchema);
+			expectedMessages.add(Map.of(kv.getKey(), kv.getValue()));
+		}
+		assertThat(keyValueLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(keyValueMessages).containsExactlyInAnyOrderElementsOf(expectedMessages);
+	}
+
+	@EnablePulsar
+	@Configuration
+	static class PulsarReaderAutoConsumeSchemaTestsConfig {
+
+		@PulsarReader(id = "stringAcListener", topics = STRING_TOPIC, schemaType = SchemaType.AUTO_CONSUME,
+				startMessageId = "earliest")
+		void listenString(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue().getNativeObject()).isInstanceOf(String.class);
+			stringMessages.add(genericMessage.getValue().getNativeObject().toString());
+			stringLatch.countDown();
+		}
+
+		@PulsarReader(id = "jsonAcListener", topics = JSON_TOPIC, schemaType = SchemaType.AUTO_CONSUME,
+				startMessageId = "earliest")
+		void listenJson(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue()).isInstanceOf(GenericJsonRecord.class);
+			GenericJsonRecord record = (GenericJsonRecord) genericMessage.getValue();
+			assertThat(record.getSchemaType()).isEqualTo(SchemaType.JSON);
+			assertThat(record).extracting("schemaInfo")
+				.satisfies((obj) -> assertThat(obj.toString()).contains("\"name\": \"UserRecord\""));
+			jsonMessages.add(record.getFields()
+				.stream()
+				.map(Field::getName)
+				.collect(Collectors.toMap(Function.identity(), record::getField)));
+			jsonLatch.countDown();
+		}
+
+		@PulsarReader(id = "avroAcListener", topics = AVRO_TOPIC, schemaType = SchemaType.AUTO_CONSUME,
+				startMessageId = "earliest")
+		void listenAvro(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue()).isInstanceOf(GenericAvroRecord.class);
+			GenericAvroRecord record = (GenericAvroRecord) genericMessage.getValue();
+			assertThat(record.getSchemaType()).isEqualTo(SchemaType.AVRO);
+			assertThat(record).extracting("schema")
+				.satisfies((obj) -> assertThat(obj.toString()).contains("\"name\":\"UserPojo\""));
+			avroMessages.add(record.getFields()
+				.stream()
+				.map(Field::getName)
+				.collect(Collectors.toMap(Function.identity(), record::getField)));
+			avroLatch.countDown();
+		}
+
+		@SuppressWarnings("unchecked")
+		@PulsarReader(id = "keyvalueAcListener", topics = KEYVALUE_TOPIC, schemaType = SchemaType.AUTO_CONSUME,
+				startMessageId = "earliest")
+		void listenKeyvalue(Message<GenericRecord> genericMessage) {
+			assertThat(genericMessage.getValue().getSchemaType()).isEqualTo(SchemaType.KEY_VALUE);
+			assertThat(genericMessage.getValue().getNativeObject()).isInstanceOf(KeyValue.class);
+			var record = (KeyValue<String, Object>) genericMessage.getValue().getNativeObject();
+			keyValueMessages.add(Map.of(record.getKey(), record.getValue()));
+			keyValueLatch.countDown();
+		}
+
+	}
+
+}