Concurrent message listener container improvements

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java

@@ -54,6 +54,7 @@ public class PulsarAnnotationDrivenConfiguration {
 		factory.setPulsarConsumerFactory(pulsarConsumerFactory1);
 
 		final PulsarContainerProperties containerProperties = factory.getContainerProperties();
+		containerProperties.setSubscriptionType(this.pulsarProperties.getConsumer().getSubscriptionType());
 
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
 		PulsarProperties.Listener properties = this.pulsarProperties.getListener();

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

@@ -142,10 +142,23 @@ public abstract class AbstractPulsarListenerContainerFactory<C extends AbstractP
 		// BeanUtils.copyProperties(this.containerProperties, properties, "topics",
 		// "messageListener",
 		// "batchListener", "subscriptionName", "subscriptionType", "schema");
+
+		if (properties.getSchemaType() == null) {
+			if (this.containerProperties.getSchemaType() != null) {
+				properties.setSchemaType(this.containerProperties.getSchemaType());
+			}
+		}
+
 		if (properties.getSchema() == null) {
 			properties.setSchema(Schema.BYTES);
 		}
 
+		if (properties.getSubscriptionType() == null) {
+			properties.setSubscriptionType(this.containerProperties.getSubscriptionType());
+		}
+
+		properties.setAckMode(this.containerProperties.getAckMode());
+
 		Boolean autoStart = endpoint.getAutoStartup();
 		if (autoStart != null) {
 			instance.setAutoStartup(autoStart);

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

@@ -225,7 +225,6 @@ public abstract class AbstractPulsarListenerEndpoint<K>
 	/**
 	 * Set the concurrency for this endpoint's container.
 	 * @param concurrency the concurrency.
-	 * @since 2.2
 	 */
 	public void setConcurrency(Integer concurrency) {
 		this.concurrency = concurrency;

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

@@ -19,6 +19,8 @@ package org.springframework.pulsar.listener;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.pulsar.client.api.SubscriptionType;
+
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.task.AsyncTaskExecutor;
@@ -67,6 +69,10 @@ public class ConcurrentPulsarMessageListenerContainer<T> extends AbstractPulsarM
 		if (!isRunning()) {
 			PulsarContainerProperties containerProperties = getContainerProperties();
 
+			if (containerProperties.getSubscriptionType() == SubscriptionType.Exclusive && this.concurrency > 1) {
+				throw new IllegalStateException("concurrency > 1 is not allowed on Exclusive subscription type");
+			}
+
 			setRunning(true);
 
 			for (int i = 0; i < this.concurrency; i++) {

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

@@ -16,6 +16,7 @@
 
 package org.springframework.pulsar.listener;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -28,6 +29,7 @@ import org.apache.pulsar.client.api.BatchReceivePolicy;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Messages;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.pulsar.core.AbstractContainerBaseTests;
@@ -40,7 +42,36 @@ public class ConcurrentPulsarMessageListenerContainerTests extends AbstractConta
 
 	@Test
 	@SuppressWarnings("unchecked")
-	void basicConcurrency() throws Exception {
+	void basicConcurrencyTesting() throws Exception {
+		PulsarConsumerFactory<String> pulsarConsumerFactory = mock(PulsarConsumerFactory.class);
+		Consumer<String> consumer = mock(Consumer.class);
+
+		when(pulsarConsumerFactory.createConsumer(any(Schema.class), any(BatchReceivePolicy.class), any(Map.class)))
+				.thenReturn(consumer);
+
+		when(consumer.batchReceive()).thenReturn(mock(Messages.class));
+
+		PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
+		pulsarContainerProperties.setSchema(Schema.STRING);
+		pulsarContainerProperties.setSubscriptionType(SubscriptionType.Failover);
+		pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (cons, msg) -> {
+		});
+
+		ConcurrentPulsarMessageListenerContainer<String> container = new ConcurrentPulsarMessageListenerContainer<>(
+				pulsarConsumerFactory, pulsarContainerProperties);
+
+		container.setConcurrency(3);
+
+		container.start();
+
+		verify(pulsarConsumerFactory, times(3)).createConsumer(any(Schema.class), any(BatchReceivePolicy.class),
+				any(Map.class));
+		verify(consumer, times(3)).batchReceive();
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	void exclusiveSubscriptionMustUseSingleThread() throws Exception {
 		PulsarConsumerFactory<String> pulsarConsumerFactory = mock(PulsarConsumerFactory.class);
 		Consumer<String> consumer = mock(Consumer.class);
 
@@ -59,12 +90,8 @@ public class ConcurrentPulsarMessageListenerContainerTests extends AbstractConta
 
 		container.setConcurrency(3);
 
-		container.start();
-
-		verify(pulsarConsumerFactory, times(3)).createConsumer(any(Schema.class), any(BatchReceivePolicy.class),
-				any(Map.class));
-
-		verify(consumer, times(3)).batchReceive();
+		assertThatThrownBy(container::start).isInstanceOf(IllegalStateException.class)
+				.hasMessage("concurrency > 1 is not allowed on Exclusive subscription type");
 	}
 
 }