Added test coverage for subscription type overriding.

Author: majusko

src/main/java/io/github/majusko/pulsar/consumer/ConsumerAggregator.java

@@ -102,6 +102,8 @@ private Consumer<?> subscribe(String generatedConsumerName, ConsumerHolder holde
     private SubscriptionType getSubscriptionType(ConsumerHolder holder) throws ClientInitException {
         SubscriptionType subscriptionType = Arrays.stream(holder.getAnnotation().subscriptionType()).findFirst().orElse(null);
 
+        String aa = consumerProperties.getSubscriptionType();
+
         if (subscriptionType == null && Strings.isNullOrEmpty(consumerProperties.getSubscriptionType())) {
             subscriptionType = DEFAULT_SUBSCRIPTION_TYPE;
         } else if (subscriptionType == null && !Strings.isNullOrEmpty(consumerProperties.getSubscriptionType())) {

src/main/resources/application-test.properties

@@ -1,6 +1,6 @@
 pulsar.consumer.default.ack-timeout-ms=1000
 pulsar.consumer.default.dead-letter-policy-max-redeliver-count=1
-pulsar.consmer.default.subscription-type=Shared
+pulsar.consumer.default.subscription-type=Shared
 
 #Custom Topics
 my.custom.topic.name=custom-topic-name-from-app-prop

src/test/java/io/github/majusko/pulsar/PulsarJavaSpringBootStarterApplicationTests.java

@@ -138,7 +138,7 @@ void testProducerCreateMessageMethod() throws PulsarClientException {
     void testConsumerRegistration1() throws Exception {
         final List<Consumer> consumers = consumerAggregator.getConsumers();
 
-        Assertions.assertEquals(14, consumers.size());
+        Assertions.assertEquals(16, consumers.size());
 
         final Consumer<?> consumer =
             consumers.stream().filter($ -> $.getTopic().equals(urlBuildService.buildTopicUrl("topic-one"))).findFirst().orElseThrow(Exception::new);
@@ -169,7 +169,7 @@ void testProducerRegistration() {
 
         final Map<String, ImmutablePair<Class<?>, Serialization>> topics = producerFactory.getTopics();
 
-        Assertions.assertEquals(14, topics.size());
+        Assertions.assertEquals(16, topics.size());
 
         final Set<String> topicNames = new HashSet<>(topics.keySet());
 
@@ -284,4 +284,24 @@ void sharedSubscriptionOverride() throws Exception {
         producer.send(TestConsumers.SHARED_SUB_TEST, new MyMsg(VALIDATION_STRING));
         await().atMost(Duration.ofSeconds(10)).until(() -> testConsumers.subscribeToSharedTopicSubscription.get());
     }
+
+    @Test
+    void exclusiveSubscriptionOverride() throws Exception {
+        final ConsumerBase<?> consumer = (ConsumerBase<?>) consumerAggregator.getConsumers().stream()
+            .filter($ -> $.getTopic().equals(urlBuildService.buildTopicUrl(TestConsumers.EXCLUSIVE_SUB_TEST)))
+            .findFirst()
+            .orElseThrow(() -> new Exception("Missing tested consumer."));
+
+        final Field f = ConsumerBase.class.getDeclaredField("conf");
+
+        f.setAccessible(true);
+
+        final ConsumerConfigurationData<?> conf = (ConsumerConfigurationData<?>) f.get(consumer);
+
+        Assertions.assertEquals(urlBuildService.buildTopicUrl(TestConsumers.EXCLUSIVE_SUB_TEST), consumer.getTopic());
+        Assertions.assertEquals(SubscriptionType.Exclusive, conf.getSubscriptionType());
+
+        producer.send(TestConsumers.EXCLUSIVE_SUB_TEST, new MyMsg(VALIDATION_STRING));
+        await().atMost(Duration.ofSeconds(10)).until(() -> testConsumers.subscribeToSharedTopicSubscription.get());
+    }
 }

src/test/java/io/github/majusko/pulsar/TestConsumers.java

@@ -37,6 +37,7 @@ public class TestConsumers {
     public static final String CUSTOM_CONSUMER_TOPIC = "custom-consumer-topic";
     public static final String CUSTOM_SUB_AND_CONSUMER_TOPIC = "custom-sub-and-consumer";
     public static final String SHARED_SUB_TEST = "shared-sub-consumer";
+    public static final String EXCLUSIVE_SUB_TEST = "exclusive-sub-consumer";
 
     @PulsarConsumer(topic = "topic-one", clazz = MyMsg.class, serialization = Serialization.JSON)
     public void topicOneListener(MyMsg myMsg) {
@@ -167,4 +168,14 @@ public void sharedTopicSubscription(MyMsg myMsg) {
         Assertions.assertEquals(PulsarJavaSpringBootStarterApplicationTests.VALIDATION_STRING, myMsg.getData());
         subscribeToSharedTopicSubscription.set(true);
     }
+
+    @PulsarConsumer(
+        topic = EXCLUSIVE_SUB_TEST,
+        clazz = MyMsg.class,
+        subscriptionType = SubscriptionType.Exclusive)
+    public void exclusiveTopicSubscription(MyMsg myMsg) {
+        Assertions.assertNotNull(myMsg);
+        Assertions.assertEquals(PulsarJavaSpringBootStarterApplicationTests.VALIDATION_STRING, myMsg.getData());
+        subscribeToSharedTopicSubscription.set(true);
+    }
 }

src/test/java/io/github/majusko/pulsar/TestProducerConfiguration.java

@@ -29,6 +29,7 @@ public ProducerFactory producerFactory() {
             .addProducer("${my.custom.topic.name}", MyMsg.class)
             .addProducer(TestConsumers.CUSTOM_SUB_AND_CONSUMER_TOPIC, MyMsg.class)
             .addProducer(TestConsumers.SHARED_SUB_TEST, MyMsg.class)
+            .addProducer(TestConsumers.EXCLUSIVE_SUB_TEST, MyMsg.class)
             .addProducer(TestConsumers.CUSTOM_CONSUMER_TOPIC, MyMsg.class);
     }
 }