From 8e33ac0b122bc0e75df299919c956cacabcc9809 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Thu, 11 Aug 2022 22:06:26 -0400 Subject: [PATCH] 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" }) --- ...bstractPulsarListenerContainerFactory.java | 4 +- .../AbstractPulsarListenerEndpoint.java | 4 + .../pulsar/config/PulsarListenerEndpoint.java | 3 + .../config/PulsarListenerEndpointAdapter.java | 6 + ...bstractPulsarMessageListenerContainer.java | 4 + .../pulsar/listener/Acknowledgement.java | 9 +- ...DefaultPulsarMessageListenerContainer.java | 43 ++++-- .../listener/PulsarContainerProperties.java | 11 ++ .../PulsarMessageListenerContainer.java | 4 + .../listener/PulsarRecordMessageListener.java | 4 +- .../core/AbstractContainerBaseTests.java | 2 +- .../pulsar/listener/PulsarListenerTests.java | 123 ++++++++++++++++++ 12 files changed, 197 insertions(+), 20 deletions(-) create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java index 75ca563b..4e71bd3f 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java @@ -156,7 +156,9 @@ public abstract class AbstractPulsarListenerContainerFactory this.consumerProperties = consumerProperties; } + public Properties getConsumerProperties() { + return this.consumerProperties; + } + @Nullable public Boolean getBatchListener() { return this.batchListener; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java index dcff1531..0bd4e346 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java +++ b/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(); + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java index 8f5038f5..e3b59e4d 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpointAdapter.java +++ b/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; + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java index 4b67a2cc..0885393b 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java @@ -131,4 +131,8 @@ public abstract class AbstractPulsarMessageListenerContainer implements Pulsa return this.phase; } + public PulsarContainerProperties getContainerProperties() { + return this.pulsarContainerProperties; + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/Acknowledgement.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/Acknowledgement.java index dfef9da5..73753b6c 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/Acknowledgement.java +++ b/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 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); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java index 2a1d912e..5fcada54 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java +++ b/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 extends AbstractPulsarMess } private Map extractPropertiesToOverride(PulsarContainerProperties pulsarContainerProperties) { - final SubscriptionType subscriptionType = pulsarContainerProperties.getSubscriptionType(); - final Map propertiesToOverride = new HashMap<>(); - if (subscriptionType != null) { - propertiesToOverride.put("subscriptionType", subscriptionType); + + Properties propertyOverrides = this.containerProperties.getPulsarConsumerProperties(); + + final Map propOverridesAsMap = propertyOverrides.entrySet().stream().collect(Collectors + .toMap(e -> String.valueOf(e.getKey()), Map.Entry::getValue, (prev, next) -> next, HashMap::new)); + + final Map propertiesToOverride = new HashMap<>(propOverridesAsMap); + if (propertiesToOverride.containsKey("topicNames")) { + final String topicsFromMap = (String) propertiesToOverride.get("topicNames"); + final String[] topicNames = topicsFromMap.split(","); + final Set propertiesDefinedTopics = new HashSet<>(Arrays.stream(topicNames).toList()); + if (!propertiesDefinedTopics.isEmpty()) { + propertiesToOverride.put("topicNames", propertiesDefinedTopics); + } } - final String[] topics = pulsarContainerProperties.getTopics(); - final Set 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 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; } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java index 30458d3b..d48d3484 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java +++ b/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; + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java index 7ba4d48a..c386eb17 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarMessageListenerContainer.java +++ b/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"); + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarRecordMessageListener.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarRecordMessageListener.java index ddbd11ad..149cdf83 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarRecordMessageListener.java +++ b/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 message payload type * @author Soby Chacko diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/AbstractContainerBaseTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/AbstractContainerBaseTests.java index ca45c3bf..34122cb5 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/AbstractContainerBaseTests.java +++ b/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"); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java new file mode 100644 index 00000000..28d54f87 --- /dev/null +++ b/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 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 pulsarProducerFactory(PulsarClient pulsarClient) { + Map 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 pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { + return new PulsarTemplate<>(pulsarProducerFactory); + } + + @Bean + public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { + Map config = new HashMap<>(); + return new DefaultPulsarConsumerFactory<>(pulsarClient, config); + } + + @Bean + PulsarListenerContainerFactory pulsarListenerContainerFactory( + PulsarConsumerFactory pulsarConsumerFactory) { + final DefaultPulsarListenerContainerFactory pulsarListenerContainerFactory = new DefaultPulsarListenerContainerFactory<>(); + pulsarListenerContainerFactory.setPulsarConsumerFactory(pulsarConsumerFactory); + return pulsarListenerContainerFactory; + } + + } + +}