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

See #488

Author: onobc

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 @@
 	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 Object postProcessAfterInitialization(Object bean, String beanName) throw
 					});
 			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 @@ private void processReactivePulsarListenerAnnotation(MethodReactivePulsarListene
 		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 @@ private void processReactivePulsarListenerAnnotation(MethodReactivePulsarListene
 		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 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.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.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 PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> pulsa
 			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 @@ Mono<Void> listenString(String ignored) {
 	}
 
 	@Nested
-	@ContextConfiguration(classes = ReactivePulsarListenerTests.PulsarHeadersTest.PulsarListenerWithHeadersConfig.class)
+	@ContextConfiguration(classes = PulsarListenerWithHeadersConfig.class)
 	class PulsarHeadersTest {
 
 		static CountDownLatch simpleListenerLatch = new CountDownLatch(1);
@@ -894,4 +909,160 @@ Mono<Void> listen2(String message) {
 
 	}
 
+	@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 @@
 	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 @@
 	 * 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 @@ private void processPulsarListenerAnnotation(MethodPulsarListenerEndpoint<?> end
 		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 @@ private void processPulsarListenerAnnotation(MethodPulsarListenerEndpoint<?> end
 		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) {