Allow Pulsar consumer properties on PulsarListener

It is convenient to provide arbitrary Pulsar consumer properties
directly on the PulsarListener annotation. This commit enables that.

Here is an examle usage:

@PulsarListener(properties = { "receiverQueueSize=5000" })

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

@@ -156,7 +156,9 @@ public abstract class AbstractPulsarListenerContainerFactory<C extends AbstractP
 
 		JavaUtils.INSTANCE.acceptIfNotNull(this.phase, instance::setPhase)
 				.acceptIfNotNull(this.applicationContext, instance::setApplicationContext)
-				.acceptIfNotNull(this.applicationEventPublisher, instance::setApplicationEventPublisher);
+				.acceptIfNotNull(this.applicationEventPublisher, instance::setApplicationEventPublisher)
+				.acceptIfNotNull(endpoint.getConsumerProperties(),
+						instance.getContainerProperties()::setPulsarConsumerProperties);
 	}
 
 }

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

@@ -181,6 +181,10 @@ public abstract class AbstractPulsarListenerEndpoint<K>
 		this.consumerProperties = consumerProperties;
 	}
 
+	public Properties getConsumerProperties() {
+		return this.consumerProperties;
+	}
+
 	@Nullable
 	public Boolean getBatchListener() {
 		return this.batchListener;

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

@@ -17,6 +17,7 @@
 package org.springframework.pulsar.config;
 
 import java.util.Collection;
+import java.util.Properties;
 
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.common.schema.SchemaType;
@@ -55,4 +56,6 @@ public interface PulsarListenerEndpoint {
 
 	SchemaType getSchemaType();
 
+	Properties getConsumerProperties();
+
 }

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

@@ -18,6 +18,7 @@ package org.springframework.pulsar.config;
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Properties;
 
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.common.schema.SchemaType;
@@ -73,4 +74,9 @@ public class PulsarListenerEndpointAdapter implements PulsarListenerEndpoint {
 		return null;
 	}
 
+	@Override
+	public Properties getConsumerProperties() {
+		return null;
+	}
+
 }

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

@@ -131,4 +131,8 @@ public abstract class AbstractPulsarMessageListenerContainer<T> implements Pulsa
 		return this.phase;
 	}
 
+	public PulsarContainerProperties getContainerProperties() {
+		return this.pulsarContainerProperties;
+	}
+
 }

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

@@ -23,8 +23,8 @@ import org.apache.pulsar.client.api.MessageId;
 /**
  * Contract for manual acknowledgment.
  *
- * When manual acknowledgment is used, applications can inject an Acknowledgment object
- * in the listener and then invoke manual acknowledgment.
+ * When manual acknowledgment is used, applications can inject an Acknowledgment object in
+ * the listener and then invoke manual acknowledgment.
  *
  * @author Soby Chacko
  */
@@ -37,14 +37,12 @@ public interface Acknowledgement {
 
 	/**
 	 * Manually acknowledges by the message id.
-	 *
 	 * @param messageId message id.
 	 */
 	void acknowledge(MessageId messageId);
 
 	/**
 	 * Manually acknowledges a list of messages based on their message id's.
-	 *
 	 * @param messageIds collection of message id's.
 	 */
 	void acknowledge(List<MessageId> messageIds);
@@ -56,8 +54,7 @@ public interface Acknowledgement {
 
 	/**
 	 * Negative acknowledges the current message based on the message id.
-	 *
-	 * @param messageId message id. 
+	 * @param messageId message id.
 	 */
 	void nack(MessageId messageId);
 

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

@@ -21,10 +21,12 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
@@ -206,18 +208,39 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 		}
 
 		private Map<String, Object> extractPropertiesToOverride(PulsarContainerProperties pulsarContainerProperties) {
-			final SubscriptionType subscriptionType = pulsarContainerProperties.getSubscriptionType();
-			final Map<String, Object> propertiesToOverride = new HashMap<>();
-			if (subscriptionType != null) {
-				propertiesToOverride.put("subscriptionType", subscriptionType);
+
+			Properties propertyOverrides = this.containerProperties.getPulsarConsumerProperties();
+
+			final Map<String, Object> propOverridesAsMap = propertyOverrides.entrySet().stream().collect(Collectors
+					.toMap(e -> String.valueOf(e.getKey()), Map.Entry::getValue, (prev, next) -> next, HashMap::new));
+
+			final Map<String, Object> propertiesToOverride = new HashMap<>(propOverridesAsMap);
+			if (propertiesToOverride.containsKey("topicNames")) {
+				final String topicsFromMap = (String) propertiesToOverride.get("topicNames");
+				final String[] topicNames = topicsFromMap.split(",");
+				final Set<String> propertiesDefinedTopics = new HashSet<>(Arrays.stream(topicNames).toList());
+				if (!propertiesDefinedTopics.isEmpty()) {
+					propertiesToOverride.put("topicNames", propertiesDefinedTopics);
+				}
 			}
-			final String[] topics = pulsarContainerProperties.getTopics();
-			final Set<String> strings = new HashSet<>(Arrays.stream(topics).toList());
-			if (!strings.isEmpty()) {
-				propertiesToOverride.put("topicNames", strings);
+
+			if (!propertiesToOverride.containsKey("subscriptionType")) {
+				final SubscriptionType subscriptionType = pulsarContainerProperties.getSubscriptionType();
+				if (subscriptionType != null) {
+					propertiesToOverride.put("subscriptionType", subscriptionType);
+				}
 			}
-			if (StringUtils.hasText(pulsarContainerProperties.getSubscriptionName())) {
-				propertiesToOverride.put("subscriptionName", pulsarContainerProperties.getSubscriptionName());
+			if (!propertiesToOverride.containsKey("topicNames")) {
+				final String[] topics = pulsarContainerProperties.getTopics();
+				final Set<String> listenerDefinedTopics = new HashSet<>(Arrays.stream(topics).toList());
+				if (!listenerDefinedTopics.isEmpty()) {
+					propertiesToOverride.put("topicNames", listenerDefinedTopics);
+				}
+			}
+			if (!propertiesToOverride.containsKey("subscriptionName")) {
+				if (StringUtils.hasText(pulsarContainerProperties.getSubscriptionName())) {
+					propertiesToOverride.put("subscriptionName", pulsarContainerProperties.getSubscriptionName());
+				}
 			}
 			return propertiesToOverride;
 		}

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

@@ -17,6 +17,7 @@
 package org.springframework.pulsar.listener;
 
 import java.time.Duration;
+import java.util.Properties;
 import java.util.regex.Pattern;
 
 import org.apache.pulsar.client.api.Schema;
@@ -83,6 +84,8 @@ public class PulsarContainerProperties {
 
 	private AckMode ackMode = AckMode.BATCH;
 
+	private Properties pulsarConsumerProperties = new Properties();
+
 	public PulsarContainerProperties(String... topics) {
 		this.topics = topics.clone();
 		this.topicsPattern = null;
@@ -211,4 +214,12 @@ public class PulsarContainerProperties {
 		this.schemaType = schemaType;
 	}
 
+	public Properties getPulsarConsumerProperties() {
+		return this.pulsarConsumerProperties;
+	}
+
+	public void setPulsarConsumerProperties(Properties pulsarConsumerProperties) {
+		this.pulsarConsumerProperties = pulsarConsumerProperties;
+	}
+
 }

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

@@ -38,4 +38,8 @@ public interface PulsarMessageListenerContainer extends SmartLifecycle, Disposab
 		// empty
 	}
 
+	default PulsarContainerProperties getContainerProperties() {
+		throw new UnsupportedOperationException("This container doesn't support retrieving its properties");
+	}
+
 }

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

@@ -21,8 +21,8 @@ import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageListener;
 
 /**
- * Base record MessageListener that simply extends from {@link MessageListener}.
- * This extension is needed as a base class to deal with acknowledgments in the framework.
+ * Base record MessageListener that simply extends from {@link MessageListener}. This
+ * extension is needed as a base class to deal with acknowledgments in the framework.
  *
  * @param <T> message payload type
  * @author Soby Chacko

spring-pulsar/src/test/java/org/springframework/pulsar/core/AbstractContainerBaseTests.java

@@ -19,7 +19,7 @@ package org.springframework.pulsar.core;
 import org.testcontainers.containers.PulsarContainer;
 import org.testcontainers.utility.DockerImageName;
 
-abstract class AbstractContainerBaseTests {
+public abstract class AbstractContainerBaseTests {
 
 	static final DockerImageName PULSAR_IMAGE = DockerImageName.parse("apachepulsar/pulsar:2.10.1");
 

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

@@ -0,0 +1,123 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.annotation.EnablePulsar;
+import org.springframework.pulsar.annotation.PulsarListener;
+import org.springframework.pulsar.config.DefaultPulsarListenerContainerFactory;
+import org.springframework.pulsar.config.PulsarClientConfiguration;
+import org.springframework.pulsar.config.PulsarClientFactoryBean;
+import org.springframework.pulsar.config.PulsarListenerContainerFactory;
+import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
+import org.springframework.pulsar.core.AbstractContainerBaseTests;
+import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
+import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarConsumerFactory;
+import org.springframework.pulsar.core.PulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Soby Chacko
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class PulsarListenerTests extends AbstractContainerBaseTests {
+
+	static CountDownLatch latch = new CountDownLatch(1);
+
+	@Autowired
+	PulsarTemplate<String> pulsarTemplate;
+
+	@Autowired
+	private PulsarListenerEndpointRegistry registry;
+
+	@Test
+	void testPulsarListenerProvidedConsumerProperties() throws Exception {
+		final PulsarContainerProperties pulsarContainerProperties = this.registry.getListenerContainer("foo")
+				.getContainerProperties();
+		final Properties pulsarConsumerProperties = pulsarContainerProperties.getPulsarConsumerProperties();
+		assertThat(pulsarConsumerProperties.size()).isEqualTo(2);
+		assertThat(pulsarConsumerProperties.get("topicNames")).isEqualTo("foo-1");
+		assertThat(pulsarConsumerProperties.get("subscriptionName")).isEqualTo("subscription-1");
+		pulsarTemplate.send("hello foo");
+		assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+	}
+
+	@Configuration
+	@EnablePulsar
+	public static class Config {
+
+		@PulsarListener(id = "foo", properties = { "subscriptionName=subscription-1", "topicNames=foo-1" })
+		void listen1(String message) {
+			latch.countDown();
+		}
+
+		@Bean
+		public PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarClient) {
+			Map<String, Object> config = new HashMap<>();
+			config.put("topicName", "foo-1");
+			return new DefaultPulsarProducerFactory<>(pulsarClient, config);
+		}
+
+		@Bean
+		public PulsarClientFactoryBean pulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) {
+			return new PulsarClientFactoryBean(pulsarClientConfiguration);
+		}
+
+		@Bean
+		public PulsarClientConfiguration pulsarClientConfiguration() {
+			return new PulsarClientConfiguration(Map.of("serviceUrl", getPulsarBrokerUrl()));
+		}
+
+		@Bean
+		public PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
+			return new PulsarTemplate<>(pulsarProducerFactory);
+		}
+
+		@Bean
+		public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient) {
+			Map<String, Object> config = new HashMap<>();
+			return new DefaultPulsarConsumerFactory<>(pulsarClient, config);
+		}
+
+		@Bean
+		PulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
+				PulsarConsumerFactory<Object> pulsarConsumerFactory) {
+			final DefaultPulsarListenerContainerFactory<?, ?> pulsarListenerContainerFactory = new DefaultPulsarListenerContainerFactory<>();
+			pulsarListenerContainerFactory.setPulsarConsumerFactory(pulsarConsumerFactory);
+			return pulsarListenerContainerFactory;
+		}
+
+	}
+
+}