Make subscriptionType nullable on @Reactive/PulsarListener (#489)

See #488

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc

@@ -149,11 +149,15 @@ NOTE: If you are not using the starter, you will need to configure and register
 When it comes to Pulsar consumers, we recommend that end-user applications use the `PulsarListener` annotation.
 To use `PulsarListener`, you need to use the `@EnablePulsar` annotation.
 When you use Spring Boot support, it automatically enables this annotation and configures all the components necessary for `PulsarListener`, such as the message listener infrastructure (which is responsible for creating the Pulsar consumer).
-`PulsarMessageListenerContainer` uses a `PulsarConsumerFactory` to create and manage the Pulsar consumer.
+`PulsarMessageListenerContainer` uses a `PulsarConsumerFactory` to create and manage the Pulsar consumer the underlying Pulsar consumer that it uses to consume messages.
 
-Spring Boot auto-configuration also provides this consumer factory which you can further configure by specifying **most** of the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties.
+Spring Boot provides this consumer factory which you can further configure by specifying the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties.
+**Most** of the configured properties on the factory will be respected in the listener with the following **exceptions**:
+
+TIP: The `spring.pulsar.consumer.subscription.name` property is ignored and is instead generated when not specified on the annotation.
+
+TIP: The `spring.pulsar.consumer.subscription-type` property is ignored and is instead taken from the value on the annotation. However, you can set the `subscriptionType = {}` on the annotation to instead use the property value as the default.
 
-NOTE: `spring.pulsar.consumer.subscription.name` is ignored and is instead generated when not specified on the annotation.
 
 Let us revisit the `PulsarListener` code snippet we saw in the quick-tour section:
 

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc

@@ -159,8 +159,8 @@ Mono<Void> listen(String message) {
 }
 ----
 
-In this most basic form, when the `subscriptionName` is not provided on the `@ReactivePulsarListener` annotation an auto-generated subscription name will be used.
-Likewise, when the `topics` are not directly provided, a <<topic-resolution-process-reactive,topic resolution process>> is used to determine the destination topic.
+In this most basic form, when the `topics` are not directly provided, a <<topic-resolution-process-reactive,topic resolution process>> is used to determine the destination topic.
+Likewise, when the `subscriptionName` is not provided on the `@ReactivePulsarListener` annotation an auto-generated subscription name will be used.
 
 In the `ReactivePulsarListener` method shown earlier, we receive the data as `String`, but we do not specify any schema types.
 Internally, the framework relies on Pulsar's schema mechanism to convert the data to the required type.
@@ -238,9 +238,14 @@ Flux<MessageResult<Void>> listen2(Flux<org.springframework.messaging.Message<Foo
 ----
 
 ==== Configuration - Application Properties
-The listener ultimately relies on `ReactivePulsarConsumerFactory` to create and manage the underlying Pulsar consumer.
+The listener relies on the `ReactivePulsarConsumerFactory` to create and manage the underlying Pulsar consumer that it uses to consume messages.
+Spring Boot provides this consumer factory which you can further configure by specifying the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties.
+**Most** of the configured properties on the factory will be respected in the listener with the following **exceptions**:
+
+TIP: The `spring.pulsar.consumer.subscription.name` property is ignored and is instead generated when not specified on the annotation.
+
+TIP: The `spring.pulsar.consumer.subscription-type` property is ignored and is instead taken from the value on the annotation. However, you can set the `subscriptionType = {}` on the annotation to instead use the property value as the default.
 
-Spring Boot provides this consumer factory which can be configured with any of the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application-properties.
 
 [[reactive-consumer-customizer]]
 ==== Consumer Customization

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

@@ -72,10 +72,12 @@ public @interface ReactivePulsarListener {
 	String subscriptionName() default "";
 
 	/**
-	 * Pulsar subscription type for this listener.
-	 * @return the {@code subscriptionType} for this listener
+	 * Pulsar subscription type for this listener - expected to be a single element array
+	 * with subscription type or empty array to indicate null type.
+	 * @return single element array with the subscription type or empty array to indicate
+	 * no type chosen by user
 	 */
-	SubscriptionType subscriptionType() default SubscriptionType.Exclusive;
+	SubscriptionType[] subscriptionType() default { SubscriptionType.Exclusive };
 
 	/**
 	 * Pulsar schema type for this listener.

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

@@ -153,7 +153,8 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 					});
 			if (annotatedMethods.isEmpty()) {
 				this.nonAnnotatedClasses.add(bean.getClass());
-				this.logger.trace(() -> "No @PulsarListener annotations found on bean type: " + bean.getClass());
+				this.logger
+					.trace(() -> "No @ReactivePulsarListener annotations found on bean type: " + bean.getClass());
 			}
 			else {
 				// Non-empty set of methods
@@ -236,7 +237,7 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 		endpoint.setId(getEndpointId(reactivePulsarListener));
 		endpoint.setTopics(topics);
 		endpoint.setTopicPattern(topicPattern);
-		endpoint.setSubscriptionType(reactivePulsarListener.subscriptionType());
+		resolveSubscriptionType(endpoint, reactivePulsarListener);
 		endpoint.setSchemaType(reactivePulsarListener.schemaType());
 
 		String concurrency = reactivePulsarListener.concurrency();
@@ -260,6 +261,15 @@ public class ReactivePulsarListenerAnnotationBeanPostProcessor<V> extends Abstra
 		resolveConsumerCustomizer(endpoint, reactivePulsarListener);
 	}
 
+	private void resolveSubscriptionType(MethodReactivePulsarListenerEndpoint<?> endpoint,
+			ReactivePulsarListener reactivePulsarListener) {
+		Assert.state(reactivePulsarListener.subscriptionType().length <= 1,
+				() -> "ReactivePulsarListener.subscriptionType must have 0 or 1 elements");
+		if (reactivePulsarListener.subscriptionType().length == 1) {
+			endpoint.setSubscriptionType(reactivePulsarListener.subscriptionType()[0]);
+		}
+	}
+
 	private void resolveDeadLetterPolicy(MethodReactivePulsarListenerEndpoint<?> endpoint,
 			ReactivePulsarListener reactivePulsarListener) {
 		Object deadLetterPolicy = resolveExpression(reactivePulsarListener.deadLetterPolicy());

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

@@ -43,7 +43,7 @@ public class ReactivePulsarContainerProperties<T> {
 
 	private String subscriptionName;
 
-	private SubscriptionType subscriptionType = SubscriptionType.Exclusive;
+	private SubscriptionType subscriptionType;
 
 	private Schema<T> schema;
 

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

@@ -20,8 +20,9 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
@@ -45,10 +46,13 @@ import org.apache.pulsar.common.schema.KeyValueEncodingType;
 import org.apache.pulsar.common.schema.SchemaType;
 import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
 import org.apache.pulsar.reactive.client.api.MessageResult;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
+import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerSpec;
 import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -71,13 +75,19 @@ import org.springframework.pulsar.reactive.config.annotation.EnableReactivePulsa
 import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener;
 import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListenerMessageConsumerBuilderCustomizer;
 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.PulsarHeadersTest.PulsarListenerWithHeadersConfig;
 import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.User2;
+import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SubscriptionTypeTests.WithDefaultType.WithDefaultTypeConfig;
+import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SubscriptionTypeTests.WithSpecificTypes.WithSpecificTypesConfig;
 import org.springframework.pulsar.support.PulsarHeaders;
 import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.util.ObjectUtils;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -122,9 +132,14 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 			return new PulsarTemplate<>(pulsarProducerFactory);
 		}
 
+		@SuppressWarnings("unchecked")
 		@Bean
-		public ReactivePulsarConsumerFactory<String> pulsarConsumerFactory(ReactivePulsarClient pulsarClient) {
-			return new DefaultReactivePulsarConsumerFactory<>(pulsarClient, Collections.emptyList());
+		public ConsumerTrackingReactivePulsarConsumerFactory<String> pulsarConsumerFactory(
+				ReactivePulsarClient pulsarClient,
+				ObjectProvider<ReactiveMessageConsumerBuilderCustomizer<String>> defaultConsumerCustomizersProvider) {
+			DefaultReactivePulsarConsumerFactory<String> consumerFactory = new DefaultReactivePulsarConsumerFactory<>(
+					pulsarClient, defaultConsumerCustomizersProvider.orderedStream().toList());
+			return new ConsumerTrackingReactivePulsarConsumerFactory<>(consumerFactory);
 		}
 
 		@Bean
@@ -721,7 +736,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 	}
 
 	@Nested
-	@ContextConfiguration(classes = ReactivePulsarListenerTests.PulsarHeadersTest.PulsarListenerWithHeadersConfig.class)
+	@ContextConfiguration(classes = PulsarListenerWithHeadersConfig.class)
 	class PulsarHeadersTest {
 
 		static CountDownLatch simpleListenerLatch = new CountDownLatch(1);
@@ -894,4 +909,160 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 
 	}
 
+	@Nested
+	class SubscriptionTypeTests {
+
+		@Nested
+		@ContextConfiguration(classes = WithDefaultTypeConfig.class)
+		class WithDefaultType {
+
+			static final CountDownLatch latchTypeNotSet = new CountDownLatch(1);
+
+			@Test
+			void whenTypeNotSetAnywhereThenFallbackTypeIsUsed(
+					@Autowired ConsumerTrackingReactivePulsarConsumerFactory<String> consumerFactory) throws Exception {
+				assertThat(consumerFactory.topicNameToConsumerSpec).hasEntrySatisfying("rpl-typeNotSetAnywhere-topic",
+						(consumerSpec) -> assertThat(consumerSpec.getSubscriptionType())
+							.isEqualTo(SubscriptionType.Exclusive));
+				pulsarTemplate.send("rpl-typeNotSetAnywhere-topic", "hello-rpl-typeNotSetAnywhere");
+				assertThat(latchTypeNotSet.await(10, TimeUnit.SECONDS)).isTrue();
+			}
+
+			@Configuration(proxyBeanMethods = false)
+			static class WithDefaultTypeConfig {
+
+				@ReactivePulsarListener(topics = "rpl-typeNotSetAnywhere-topic",
+						subscriptionName = "rpl-typeNotSetAnywhere-sub",
+						consumerCustomizer = "subscriptionInitialPositionEarliest")
+				Mono<Void> listenWithoutTypeSetAnywhere(String ignored) {
+					latchTypeNotSet.countDown();
+					return Mono.empty();
+				}
+
+			}
+
+		}
+
+		@Nested
+		@ContextConfiguration(classes = WithSpecificTypesConfig.class)
+		class WithSpecificTypes {
+
+			static final CountDownLatch latchTypeSetConsumerFactory = new CountDownLatch(1);
+
+			static final CountDownLatch latchTypeSetAnnotation = new CountDownLatch(1);
+
+			static final CountDownLatch latchWithCustomizer = new CountDownLatch(1);
+
+			@Test
+			void whenTypeSetOnlyInConsumerFactoryThenConsumerFactoryTypeIsUsed(
+					@Autowired ConsumerTrackingReactivePulsarConsumerFactory<String> consumerFactory) throws Exception {
+				assertThat(consumerFactory.getSpec("rpl-typeSetConsumerFactory-topic"))
+					.extracting(ReactiveMessageConsumerSpec::getSubscriptionType)
+					.isEqualTo(SubscriptionType.Shared);
+				pulsarTemplate.send("rpl-typeSetConsumerFactory-topic", "hello-rpl-typeSetConsumerFactory");
+				assertThat(latchTypeSetConsumerFactory.await(10, TimeUnit.SECONDS)).isTrue();
+			}
+
+			@Test
+			void whenTypeSetOnAnnotationThenAnnotationTypeIsUsed(
+					@Autowired ConsumerTrackingReactivePulsarConsumerFactory<String> consumerFactory) throws Exception {
+				assertThat(consumerFactory.getSpec("rpl-typeSetAnnotation-topic"))
+					.extracting(ReactiveMessageConsumerSpec::getSubscriptionType)
+					.isEqualTo(SubscriptionType.Key_Shared);
+				pulsarTemplate.send("rpl-typeSetAnnotation-topic", "hello-rpl-typeSetAnnotation");
+				assertThat(latchTypeSetAnnotation.await(10, TimeUnit.SECONDS)).isTrue();
+			}
+
+			@Test
+			void whenTypeSetWithCustomizerThenCustomizerTypeIsUsed(
+					@Autowired ConsumerTrackingReactivePulsarConsumerFactory<String> consumerFactory) throws Exception {
+				assertThat(consumerFactory.getSpec("rpl-typeSetCustomizer-topic"))
+					.extracting(ReactiveMessageConsumerSpec::getSubscriptionType)
+					.isEqualTo(SubscriptionType.Failover);
+				pulsarTemplate.send("rpl-typeSetCustomizer-topic", "hello-rpl-typeSetCustomizer");
+				assertThat(latchWithCustomizer.await(10, TimeUnit.SECONDS)).isTrue();
+			}
+
+			@Configuration(proxyBeanMethods = false)
+			static class WithSpecificTypesConfig {
+
+				@Bean
+				ReactiveMessageConsumerBuilderCustomizer<String> consumerFactoryDefaultSubTypeCustomizer() {
+					return (b) -> b.subscriptionType(SubscriptionType.Shared);
+				}
+
+				@ReactivePulsarListener(topics = "rpl-typeSetConsumerFactory-topic",
+						subscriptionName = "rpl-typeSetConsumerFactory-sub", subscriptionType = {},
+						consumerCustomizer = "subscriptionInitialPositionEarliest")
+				Mono<Void> listenWithTypeSetOnlyOnConsumerFactory(String ignored) {
+					latchTypeSetConsumerFactory.countDown();
+					return Mono.empty();
+				}
+
+				@ReactivePulsarListener(topics = "rpl-typeSetAnnotation-topic",
+						subscriptionName = "rpl-typeSetAnnotation-sub", subscriptionType = SubscriptionType.Key_Shared,
+						consumerCustomizer = "subscriptionInitialPositionEarliest")
+				Mono<Void> listenWithTypeSetOnAnnotation(String ignored) {
+					latchTypeSetAnnotation.countDown();
+					return Mono.empty();
+				}
+
+				@ReactivePulsarListener(topics = "rpl-typeSetCustomizer-topic",
+						subscriptionName = "rpl-typeSetCustomizer-sub", subscriptionType = SubscriptionType.Key_Shared,
+						consumerCustomizer = "myCustomizer")
+				Mono<Void> listenWithTypeSetInCustomizer(String ignored) {
+					latchWithCustomizer.countDown();
+					return Mono.empty();
+				}
+
+				@Bean
+				public ReactivePulsarListenerMessageConsumerBuilderCustomizer<String> myCustomizer() {
+					return cb -> cb.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+						.subscriptionType(SubscriptionType.Failover);
+				}
+
+			}
+
+		}
+
+	}
+
+	static class ConsumerTrackingReactivePulsarConsumerFactory<T> implements ReactivePulsarConsumerFactory<T> {
+
+		private Map<String, ReactiveMessageConsumerSpec> topicNameToConsumerSpec = new HashMap<>();
+
+		private ReactivePulsarConsumerFactory<T> delegate;
+
+		ConsumerTrackingReactivePulsarConsumerFactory(ReactivePulsarConsumerFactory<T> delegate) {
+			this.delegate = delegate;
+		}
+
+		@Override
+		public ReactiveMessageConsumer<T> createConsumer(Schema<T> schema) {
+			var consumer = this.delegate.createConsumer(schema);
+			storeSpec(consumer);
+			return consumer;
+		}
+
+		@Override
+		public ReactiveMessageConsumer<T> createConsumer(Schema<T> schema,
+				List<ReactiveMessageConsumerBuilderCustomizer<T>> reactiveMessageConsumerBuilderCustomizers) {
+			var consumer = this.delegate.createConsumer(schema, reactiveMessageConsumerBuilderCustomizers);
+			storeSpec(consumer);
+			return consumer;
+		}
+
+		private void storeSpec(ReactiveMessageConsumer<T> consumer) {
+			var consumerSpec = (ReactiveMessageConsumerSpec) ReflectionTestUtils.getField(consumer, "consumerSpec");
+			var topicNamesKey = !ObjectUtils.isEmpty(consumerSpec.getTopicNames()) ? consumerSpec.getTopicNames().get(0)
+					: "no-topics-set";
+			this.topicNameToConsumerSpec.put(topicNamesKey, consumerSpec);
+		}
+
+		ReactiveMessageConsumerSpec getSpec(String topic) {
+			return this.topicNameToConsumerSpec.get(topic);
+		}
+
+	}
+
 }

spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java

@@ -74,10 +74,12 @@ public @interface PulsarListener {
 	String subscriptionName() default "";
 
 	/**
-	 * Pulsar subscription type for this listener.
-	 * @return the {@code subscriptionType} for this listener
+	 * Pulsar subscription type for this listener - expected to be a single element array
+	 * with subscription type or empty array to indicate null type.
+	 * @return single element array with the subscription type or empty array to indicate
+	 * no type chosen by user
 	 */
-	SubscriptionType subscriptionType() default SubscriptionType.Exclusive;
+	SubscriptionType[] subscriptionType() default { SubscriptionType.Exclusive };
 
 	/**
 	 * Pulsar schema type for this listener.
@@ -114,7 +116,7 @@ public @interface PulsarListener {
 	 * a {@link String}, in which case the {@link Boolean#parseBoolean(String)} is used to
 	 * obtain the value.
 	 * <p>
-	 * SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
+	 * SpEL {@code #{...}} and property placeholders {@code ${...}} are supported.
 	 * @return true to auto start, false to not auto start.
 	 */
 	String autoStartup() default "";

spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java

@@ -226,7 +226,7 @@ public class PulsarListenerAnnotationBeanPostProcessor<V> extends AbstractPulsar
 		endpoint.setId(getEndpointId(pulsarListener));
 		endpoint.setTopics(topics);
 		endpoint.setTopicPattern(topicPattern);
-		endpoint.setSubscriptionType(pulsarListener.subscriptionType());
+		resolveSubscriptionType(endpoint, pulsarListener);
 		endpoint.setSchemaType(pulsarListener.schemaType());
 		endpoint.setAckMode(pulsarListener.ackMode());
 
@@ -250,6 +250,14 @@ public class PulsarListenerAnnotationBeanPostProcessor<V> extends AbstractPulsar
 		resolveConsumerCustomizer(endpoint, pulsarListener);
 	}
 
+	private void resolveSubscriptionType(MethodPulsarListenerEndpoint<?> endpoint, PulsarListener pulsarListener) {
+		Assert.state(pulsarListener.subscriptionType().length <= 1,
+				() -> "PulsarListener.subscriptionType must have 0 or 1 elements");
+		if (pulsarListener.subscriptionType().length == 1) {
+			endpoint.setSubscriptionType(pulsarListener.subscriptionType()[0]);
+		}
+	}
+
 	@SuppressWarnings({ "rawtypes" })
 	private void resolvePulsarConsumerErrorHandler(MethodPulsarListenerEndpoint<?> endpoint,
 			PulsarListener pulsarListener) {

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

@@ -46,6 +46,8 @@ import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.RedeliveryBackoff;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.ConsumerImpl;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.log.LogAccessor;
@@ -62,6 +64,7 @@ import org.springframework.pulsar.observation.PulsarListenerObservation;
 import org.springframework.pulsar.observation.PulsarMessageReceiverContext;
 import org.springframework.scheduling.SchedulingAwareRunnable;
 import org.springframework.util.Assert;
+import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
 
 import io.micrometer.observation.Observation;
@@ -226,7 +229,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 
 		private final AckMode ackMode;
 
-		private final SubscriptionType subscriptionType;
+		private SubscriptionType subscriptionType;
 
 		@SuppressWarnings({ "unchecked", "rawtypes" })
 		Listener(MessageListener<?> messageListener, PulsarContainerProperties containerProperties) {
@@ -280,12 +283,34 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 				this.consumer = getPulsarConsumerFactory().createConsumer((Schema) containerProperties.getSchema(),
 						topicNames, this.containerProperties.getSubscriptionName(), properties, customizers);
 				Assert.state(this.consumer != null, "Unable to create a consumer");
+
+				// If our subscriptionType is null - update it based on the actual
+				// subscriptionType
+				// of the underlying consumer
+				if (this.subscriptionType == null) {
+					updateSubscriptionTypeFromConsumer(this.consumer);
+				}
 			}
 			catch (PulsarClientException e) {
 				DefaultPulsarMessageListenerContainer.this.logger.error(e, () -> "Pulsar client exceptions.");
 			}
 		}
 
+		private void updateSubscriptionTypeFromConsumer(Consumer<T> consumer) {
+			try {
+				var confField = ReflectionUtils.findField(ConsumerImpl.class, "conf");
+				ReflectionUtils.makeAccessible(confField);
+				var conf = ReflectionUtils.getField(confField, consumer);
+				if (conf instanceof ConsumerConfigurationData<?> confData) {
+					this.subscriptionType = confData.getSubscriptionType();
+				}
+			}
+			catch (Exception ex) {
+				DefaultPulsarMessageListenerContainer.this.logger.error(ex,
+						() -> "Unable to determine default subscription type from consumer due to: " + ex.getMessage());
+			}
+		}
+
 		private Map<String, Object> extractDirectConsumerProperties() {
 			Properties propertyOverrides = this.containerProperties.getPulsarConsumerProperties();
 			return propertyOverrides.entrySet()
@@ -410,6 +435,8 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 									}
 								}
 								catch (PulsarClientException pce) {
+									DefaultPulsarMessageListenerContainer.this.logger.warn(pce,
+											() -> "Batch acknowledgment failed: " + pce.getMessage());
 									this.consumer.negativeAcknowledge(messages);
 								}
 							}
@@ -440,7 +467,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 					}
 					// All the records are processed at this point. Handle acks.
 					if (this.ackMode.equals(AckMode.BATCH)) {
-						handleAcks(messages);
+						handleBatchAcks(messages);
 					}
 				}
 			}
@@ -609,11 +636,11 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 		}
 
 		private boolean isSharedSubscriptionType() {
-			return this.subscriptionType.equals(SubscriptionType.Shared)
-					|| this.subscriptionType.equals(SubscriptionType.Key_Shared);
+			return this.subscriptionType != null && (this.subscriptionType.equals(SubscriptionType.Shared)
+					|| this.subscriptionType.equals(SubscriptionType.Key_Shared));
 		}
 
-		private void handleAcks(Messages<T> messages) {
+		private void handleBatchAcks(Messages<T> messages) {
 			if (this.nackableMessages.isEmpty()) {
 				try {
 					if (messages.size() > 0) {
@@ -628,6 +655,8 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 					}
 				}
 				catch (PulsarClientException pce) {
+					DefaultPulsarMessageListenerContainer.this.logger.warn(pce,
+							() -> "Batch acknowledgments failed: " + pce.getMessage());
 					this.consumer.negativeAcknowledge(messages);
 				}
 			}

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

@@ -58,7 +58,7 @@ public class PulsarContainerProperties {
 
 	private String subscriptionName;
 
-	private SubscriptionType subscriptionType = SubscriptionType.Exclusive;
+	private SubscriptionType subscriptionType;
 
 	private Schema<?> schema;
 

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

@@ -40,16 +40,20 @@ import org.apache.pulsar.client.api.RedeliveryBackoff;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 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;
+import org.assertj.core.api.AbstractObjectAssert;
+import org.assertj.core.api.InstanceOfAssertFactories;
 import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
@@ -61,6 +65,7 @@ import org.springframework.pulsar.annotation.PulsarListenerConsumerBuilderCustom
 import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
 import org.springframework.pulsar.config.PulsarListenerContainerFactory;
 import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
+import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
 import org.springframework.pulsar.core.DefaultPulsarClientFactory;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
@@ -73,6 +78,8 @@ import org.springframework.pulsar.core.PulsarTemplate;
 import org.springframework.pulsar.core.PulsarTopic;
 import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.core.TopicResolver;
+import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithDefaultType.WithDefaultTypeConfig;
+import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithSpecificTypes.WithSpecificTypesConfig;
 import org.springframework.pulsar.support.PulsarHeaders;
 import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
 import org.springframework.test.annotation.DirtiesContext;
@@ -115,8 +122,10 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 		}
 
 		@Bean
-		public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient) {
-			return new DefaultPulsarConsumerFactory<>(pulsarClient, null);
+		public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient,
+				ObjectProvider<ConsumerBuilderCustomizer<String>> defaultConsumerCustomizersProvider) {
+			return new DefaultPulsarConsumerFactory<>(pulsarClient,
+					defaultConsumerCustomizersProvider.orderedStream().toList());
 		}
 
 		@Bean
@@ -1103,4 +1112,107 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 
 	}
 
+	@Nested
+	class SubscriptionTypeTests {
+
+		@SuppressWarnings("rawtypes")
+		private static AbstractObjectAssert<?, ?> assertSubscriptionType(Consumer<?> consumer) {
+			return assertThat(consumer)
+				.extracting("conf", InstanceOfAssertFactories.type(ConsumerConfigurationData.class))
+				.extracting(ConsumerConfigurationData::getSubscriptionType);
+		}
+
+		@Nested
+		@ContextConfiguration(classes = WithDefaultTypeConfig.class)
+		class WithDefaultType {
+
+			static final CountDownLatch latchTypeNotSet = new CountDownLatch(1);
+
+			@Test
+			void whenTypeNotSetAnywhereThenFallbackTypeIsUsed() throws Exception {
+				pulsarTemplate.send("typeNotSetAnywhere-topic", "hello-typeNotSetAnywhere");
+				assertThat(latchTypeNotSet.await(5, TimeUnit.SECONDS)).isTrue();
+			}
+
+			@Configuration(proxyBeanMethods = false)
+			static class WithDefaultTypeConfig {
+
+				@PulsarListener(topics = "typeNotSetAnywhere-topic", subscriptionName = "typeNotSetAnywhere-sub")
+				void listenWithoutTypeSetAnywhere(String ignored, Consumer<String> consumer) {
+					assertSubscriptionType(consumer).isEqualTo(SubscriptionType.Exclusive);
+					latchTypeNotSet.countDown();
+				}
+
+			}
+
+		}
+
+		@Nested
+		@ContextConfiguration(classes = WithSpecificTypesConfig.class)
+		class WithSpecificTypes {
+
+			static final CountDownLatch latchTypeSetConsumerFactory = new CountDownLatch(1);
+
+			static final CountDownLatch latchTypeSetAnnotation = new CountDownLatch(1);
+
+			static final CountDownLatch latchWithCustomizer = new CountDownLatch(1);
+
+			@Test
+			void whenTypeSetOnlyInConsumerFactoryThenConsumerFactoryTypeIsUsed() throws Exception {
+				pulsarTemplate.send("typeSetConsumerFactory-topic", "hello-typeSetConsumerFactory");
+				assertThat(latchTypeSetConsumerFactory.await(5, TimeUnit.SECONDS)).isTrue();
+			}
+
+			@Test
+			void whenTypeSetOnAnnotationThenAnnotationTypeIsUsed() throws Exception {
+				pulsarTemplate.send("typeSetAnnotation-topic", "hello-typeSetAnnotation");
+				assertThat(latchTypeSetAnnotation.await(5, TimeUnit.SECONDS)).isTrue();
+			}
+
+			@Test
+			void whenTypeSetWithCustomizerThenCustomizerTypeIsUsed() throws Exception {
+				pulsarTemplate.send("typeSetCustomizer-topic", "hello-typeSetCustomizer");
+				assertThat(latchWithCustomizer.await(5, TimeUnit.SECONDS)).isTrue();
+			}
+
+			@Configuration(proxyBeanMethods = false)
+			static class WithSpecificTypesConfig {
+
+				@Bean
+				ConsumerBuilderCustomizer<String> consumerFactoryDefaultSubTypeCustomizer() {
+					return (b) -> b.subscriptionType(SubscriptionType.Shared);
+				}
+
+				@PulsarListener(topics = "typeSetConsumerFactory-topic",
+						subscriptionName = "typeSetConsumerFactory-sub", subscriptionType = {})
+				void listenWithTypeSetOnlyOnConsumerFactory(String ignored, Consumer<String> consumer) {
+					assertSubscriptionType(consumer).isEqualTo(SubscriptionType.Shared);
+					latchTypeSetConsumerFactory.countDown();
+				}
+
+				@PulsarListener(topics = "typeSetAnnotation-topic", subscriptionName = "typeSetAnnotation-sub",
+						subscriptionType = SubscriptionType.Key_Shared)
+				void listenWithTypeSetOnAnnotation(String ignored, Consumer<String> consumer) {
+					assertSubscriptionType(consumer).isEqualTo(SubscriptionType.Key_Shared);
+					latchTypeSetAnnotation.countDown();
+				}
+
+				@PulsarListener(topics = "typeSetCustomizer-topic", subscriptionName = "typeSetCustomizer-sub",
+						subscriptionType = SubscriptionType.Key_Shared, consumerCustomizer = "myCustomizer")
+				void listenWithTypeSetInCustomizer(String ignored, Consumer<String> consumer) {
+					assertSubscriptionType(consumer).isEqualTo(SubscriptionType.Failover);
+					latchWithCustomizer.countDown();
+				}
+
+				@Bean
+				public PulsarListenerConsumerBuilderCustomizer<String> myCustomizer() {
+					return cb -> cb.subscriptionType(SubscriptionType.Failover);
+				}
+
+			}
+
+		}
+
+	}
+
 }