Move default subscription type to factory

This commit moves the default subscription type from the
`@PulsarListener` annotation to the container factory (props) which
allows the `spring.pulsar.consumer.subscription-type` config prop to
be respected.

See spring-projects/spring-boot#42053

Author: onobc

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

@@ -81,7 +81,7 @@
 	 * @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 {};
 
 	/**
 	 * Pulsar schema type for this listener.

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

@@ -20,6 +20,8 @@
 import java.util.Collection;
 import java.util.HashSet;
 
+import org.apache.pulsar.client.api.SubscriptionType;
+
 import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.listener.ConcurrentPulsarMessageListenerContainer;
 import org.springframework.pulsar.listener.PulsarContainerProperties;
@@ -74,6 +76,7 @@ protected ConcurrentPulsarMessageListenerContainer<T> createContainerInstance(Pu
 		PulsarContainerProperties properties = new PulsarContainerProperties();
 		properties.setSchemaResolver(this.getContainerProperties().getSchemaResolver());
 		properties.setTopicResolver(this.getContainerProperties().getTopicResolver());
+		properties.setSubscriptionType(this.getContainerProperties().getSubscriptionType());
 
 		var parentTxnProps = this.getContainerProperties().transactions();
 		var childTxnProps = properties.transactions();
@@ -102,6 +105,10 @@ protected ConcurrentPulsarMessageListenerContainer<T> createContainerInstance(Pu
 		if (endpoint.getSubscriptionType() != null) {
 			properties.setSubscriptionType(endpoint.getSubscriptionType());
 		}
+		// Default to Exclusive if not set on container props or endpoint
+		if (properties.getSubscriptionType() == null) {
+			properties.setSubscriptionType(SubscriptionType.Exclusive);
+		}
 
 		properties.setSchemaType(endpoint.getSchemaType());
 

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

@@ -212,6 +212,51 @@ private record PulsarListenerMockComponents(PulsarConsumerFactory<String> consum
 			Consumer<String> consumer, ConcurrentPulsarMessageListenerContainer<String> concurrentContainer) {
 	}
 
+	@SuppressWarnings("unchecked")
+	@Nested
+	class SubscriptionTypeFrom {
+
+		@Test
+		void factoryPropsUsedWhenNotSetOnEndpoint() {
+			var factoryProps = new PulsarContainerProperties();
+			factoryProps.setSubscriptionType(SubscriptionType.Shared);
+			var containerFactory = new ConcurrentPulsarListenerContainerFactory<String>(
+					mock(PulsarConsumerFactory.class), factoryProps);
+			var endpoint = mock(PulsarListenerEndpoint.class);
+			when(endpoint.getConcurrency()).thenReturn(1);
+			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			assertThat(createdContainer.getContainerProperties().getSubscriptionType())
+				.isEqualTo(SubscriptionType.Shared);
+		}
+
+		@Test
+		void endpointTakesPrecedenceOverFactoryProps() {
+			var factoryProps = new PulsarContainerProperties();
+			factoryProps.setSubscriptionType(SubscriptionType.Shared);
+			var containerFactory = new ConcurrentPulsarListenerContainerFactory<String>(
+					mock(PulsarConsumerFactory.class), factoryProps);
+			var endpoint = mock(PulsarListenerEndpoint.class);
+			when(endpoint.getConcurrency()).thenReturn(1);
+			when(endpoint.getSubscriptionType()).thenReturn(SubscriptionType.Failover);
+			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			assertThat(createdContainer.getContainerProperties().getSubscriptionType())
+				.isEqualTo(SubscriptionType.Failover);
+		}
+
+		@Test
+		void defaultUsedWhenNotSetOnEndpointNorFactoryProps() {
+			var factoryProps = new PulsarContainerProperties();
+			var containerFactory = new ConcurrentPulsarListenerContainerFactory<String>(
+					mock(PulsarConsumerFactory.class), factoryProps);
+			var endpoint = mock(PulsarListenerEndpoint.class);
+			when(endpoint.getConcurrency()).thenReturn(1);
+			var createdContainer = containerFactory.createListenerContainer(endpoint);
+			assertThat(createdContainer.getContainerProperties().getSubscriptionType())
+				.isEqualTo(SubscriptionType.Exclusive);
+		}
+
+	}
+
 	@Nested
 	class ObservationConfigurationTests {