Merge pull request #81 from majusko/feature/topic-name-fix-tenancy-namespaces

Implemented the proper way of topic name creation with a support for …

Author: majusko

README.md

@@ -90,15 +90,18 @@ pulsar.operation-timeout-sec=15
 pulsar.starting-backoff-interval-ms=100
 pulsar.max-backoff-interval-sec=10
 pulsar.consumer-name-delimiter=
+pulsar.namespace=default
+pulsar.tenant=public
 
 #Consumer
 pulsar.consumer.default.dead-letter-policy-max-redeliver-count=-1
 pulsar.consumer.default.ack-timeout-ms=30
 
 ```
 
-Properties explained:
+###Properties explained:
 
+####PulsarClient
 - `pulsar.service-url` - URL used to connect to pulsar cluster.
 - `pulsar.io-threads` - Number of threads to be used for handling connections to brokers.
 - `pulsar.listener-threads` - Set the number of threads to be used for message listeners/subscribers.
@@ -109,6 +112,10 @@ Properties explained:
 - `pulsar.starting-backoff-interval-ms` - Duration of time for a backoff interval (Retry algorithm).
 - `pulsar.max-backoff-interval-sec` - The maximum duration of time for a backoff interval (Retry algorithm).
 - `pulsar.consumer-name-delimiter` - Consumer names are connection of bean name and method with a delimiter. By default, there is no delimiter and words are connected together.
+- `pulsar.namespace` - Namespace separation. For example: app1/app2 OR dev/staging/prod. More in [Namespaces docs](https://pulsar.apache.org/docs/en/concepts-messaging/#namespaces).
+- `pulsar.tenant` - Pulsar multi-tenancy support. More in [Multi Tenancy docs](https://pulsar.apache.org/docs/en/concepts-multi-tenancy/).
+
+####Consumer
 - `pulsar.consumer.default.dead-letter-policy-max-redeliver-count` - How many times should pulsar try to retry sending the message to consumer.
 - `pulsar.consumer.default.ack-timeout-ms` - How soon should be the message acked and how soon will dead letter mechanism try to retry to send the message.
 

src/main/java/io/github/majusko/pulsar/PulsarAutoConfiguration.java

@@ -1,5 +1,7 @@
 package io.github.majusko.pulsar;
 
+import io.github.majusko.pulsar.properties.ConsumerProperties;
+import io.github.majusko.pulsar.properties.PulsarProperties;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;

src/main/java/io/github/majusko/pulsar/collector/ConsumerCollector.java

@@ -13,7 +13,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
-import static io.github.majusko.pulsar.PulsarSpringStarterUtils.getParameterType;
+import static io.github.majusko.pulsar.utils.SchemaUtils.getParameterType;
 
 @Configuration
 public class ConsumerCollector implements BeanPostProcessor {

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

@@ -1,14 +1,13 @@
 package io.github.majusko.pulsar.consumer;
 
-import com.google.protobuf.GeneratedMessageV3;
-import io.github.majusko.pulsar.ConsumerProperties;
 import io.github.majusko.pulsar.PulsarMessage;
-import io.github.majusko.pulsar.PulsarSpringStarterUtils;
 import io.github.majusko.pulsar.collector.ConsumerCollector;
 import io.github.majusko.pulsar.collector.ConsumerHolder;
-import io.github.majusko.pulsar.constant.Serialization;
 import io.github.majusko.pulsar.error.FailedMessage;
 import io.github.majusko.pulsar.error.exception.ConsumerInitException;
+import io.github.majusko.pulsar.properties.ConsumerProperties;
+import io.github.majusko.pulsar.utils.SchemaUtils;
+import io.github.majusko.pulsar.utils.TopicUrlService;
 import org.apache.pulsar.client.api.*;
 import org.springframework.context.EmbeddedValueResolverAware;
 import org.springframework.context.annotation.DependsOn;
@@ -31,15 +30,17 @@ public class ConsumerAggregator implements EmbeddedValueResolverAware {
     private final ConsumerCollector consumerCollector;
     private final PulsarClient pulsarClient;
     private final ConsumerProperties consumerProperties;
+    private final TopicUrlService topicUrlService;
 
     private StringValueResolver stringValueResolver;
     private List<Consumer> consumers;
 
     public ConsumerAggregator(ConsumerCollector consumerCollector, PulsarClient pulsarClient,
-                              ConsumerProperties consumerProperties) {
+                              ConsumerProperties consumerProperties, TopicUrlService topicUrlService) {
         this.consumerCollector = consumerCollector;
         this.pulsarClient = pulsarClient;
         this.consumerProperties = consumerProperties;
+        this.topicUrlService = topicUrlService;
     }
 
     @PostConstruct
@@ -52,11 +53,13 @@ private void init() {
     private Consumer<?> subscribe(String name, ConsumerHolder holder) {
         try {
             final ConsumerBuilder<?> clientBuilder = pulsarClient
-                .newConsumer(PulsarSpringStarterUtils.getSchema(holder.getAnnotation().serialization(),
+                .newConsumer(SchemaUtils.getSchema(holder.getAnnotation().serialization(),
                     holder.getAnnotation().clazz()))
                 .consumerName("consumer-" + name)
                 .subscriptionName("subscription-" + name)
-                .topic(stringValueResolver.resolveStringValue(holder.getAnnotation().topic()))
+                .topic(topicUrlService
+                    .buildTopicUrl(stringValueResolver
+                        .resolveStringValue(holder.getAnnotation().topic())))
                 .subscriptionType(holder.getAnnotation().subscriptionType())
                 .messageListener((consumer, msg) -> {
                     try {

src/main/java/io/github/majusko/pulsar/producer/ProducerCollector.java

@@ -1,9 +1,10 @@
 package io.github.majusko.pulsar.producer;
 
-import io.github.majusko.pulsar.PulsarSpringStarterUtils;
+import io.github.majusko.pulsar.utils.SchemaUtils;
 import io.github.majusko.pulsar.annotation.PulsarProducer;
 import io.github.majusko.pulsar.collector.ProducerHolder;
 import io.github.majusko.pulsar.error.exception.ProducerInitException;
+import io.github.majusko.pulsar.utils.TopicUrlService;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
@@ -19,11 +20,13 @@
 public class ProducerCollector implements BeanPostProcessor {
 
     private final PulsarClient pulsarClient;
+    private final TopicUrlService topicUrlService;
 
     private final Map<String, Producer> producers = new ConcurrentHashMap<>();
 
-    public ProducerCollector(PulsarClient pulsarClient) {
+    public ProducerCollector(PulsarClient pulsarClient, TopicUrlService topicUrlService) {
         this.pulsarClient = pulsarClient;
+        this.topicUrlService = topicUrlService;
     }
 
     @Override
@@ -47,15 +50,15 @@ public Object postProcessAfterInitialization(Object bean, String beanName) {
     private Producer<?> buildProducer(ProducerHolder holder) {
         try {
             return pulsarClient.newProducer(getSchema(holder))
-                .topic(holder.getTopic())
+                .topic(topicUrlService.buildTopicUrl(holder.getTopic()))
                 .create();
         } catch (PulsarClientException e) {
             throw new ProducerInitException("Failed to init producer.", e);
         }
     }
 
     private <T> Schema<?> getSchema(ProducerHolder holder) throws RuntimeException {
-        return PulsarSpringStarterUtils.getSchema(holder.getSerialization(), holder.getClazz());
+        return SchemaUtils.getSchema(holder.getSerialization(), holder.getClazz());
     }
 
     Map<String, Producer> getProducers() {

src/main/java/io/github/majusko/pulsar/ConsumerProperties.java renamed to src/main/java/io/github/majusko/pulsar/properties/ConsumerProperties.java

@@ -1,4 +1,4 @@
-package io.github.majusko.pulsar;
+package io.github.majusko.pulsar.properties;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 

src/main/java/io/github/majusko/pulsar/PulsarProperties.java renamed to src/main/java/io/github/majusko/pulsar/properties/PulsarProperties.java

@@ -1,4 +1,4 @@
-package io.github.majusko.pulsar;
+package io.github.majusko.pulsar.properties;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
@@ -14,6 +14,8 @@ public class PulsarProperties {
     private Integer startingBackoffIntervalMs = 100;
     private Integer maxBackoffIntervalSec = 10;
     private String consumerNameDelimiter = "";
+    private String namespace = "default";
+    private String tenant = "public";
 
     public String getServiceUrl() {
         return serviceUrl;
@@ -94,4 +96,20 @@ public String getConsumerNameDelimiter() {
     public void setConsumerNameDelimiter(String consumerNameDelimiter) {
         this.consumerNameDelimiter = consumerNameDelimiter;
     }
+
+    public String getNamespace() {
+        return namespace;
+    }
+
+    public void setNamespace(String namespace) {
+        this.namespace = namespace;
+    }
+
+    public String getTenant() {
+        return tenant;
+    }
+
+    public void setTenant(String tenant) {
+        this.tenant = tenant;
+    }
 }

src/main/java/io/github/majusko/pulsar/PulsarSpringStarterUtils.java renamed to src/main/java/io/github/majusko/pulsar/utils/SchemaUtils.java

@@ -1,4 +1,4 @@
-package io.github.majusko.pulsar;
+package io.github.majusko.pulsar.utils;
 
 import com.google.protobuf.GeneratedMessageV3;
 import io.github.majusko.pulsar.constant.Serialization;
@@ -7,9 +7,9 @@
 
 import java.lang.reflect.Method;
 
-public class PulsarSpringStarterUtils {
+public class SchemaUtils {
 
-    private PulsarSpringStarterUtils() {
+    private SchemaUtils() {
     }
 
     private static <T> Schema<?> getGenericSchema(Serialization serialization, Class<T> clazz) throws RuntimeException {

src/main/java/io/github/majusko/pulsar/utils/TopicUrlService.java

@@ -0,0 +1,23 @@
+package io.github.majusko.pulsar.utils;
+
+import io.github.majusko.pulsar.properties.PulsarProperties;
+import org.springframework.stereotype.Service;
+
+@Service
+public class TopicUrlService {
+
+    private static final String PERSISTENT_PREFIX = "persistent";
+    private static final String NON_PERSISTENT_PREFIX = "non-persistent";
+    private static final String DEFAULT_PERSISTENCE = PERSISTENT_PREFIX;
+
+    private final PulsarProperties pulsarProperties;
+
+    private TopicUrlService(PulsarProperties pulsarProperties) {
+        this.pulsarProperties = pulsarProperties;
+    }
+
+    public String buildTopicUrl(String topic) {
+        return DEFAULT_PERSISTENCE + "://" + pulsarProperties.getTenant() + "/" + pulsarProperties.getNamespace() +
+            "/" + topic;
+    }
+}

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

@@ -9,6 +9,7 @@
 import io.github.majusko.pulsar.msg.ProtoMsg;
 import io.github.majusko.pulsar.producer.ProducerFactory;
 import io.github.majusko.pulsar.producer.PulsarTemplate;
+import io.github.majusko.pulsar.utils.TopicUrlService;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.PulsarClientException;
@@ -66,12 +67,15 @@ class PulsarJavaSpringBootStarterApplicationTests {
     @Autowired
     private PulsarTemplate<String> producerForStringTopic;
 
-    @Container
-    static PulsarContainer pulsarContainer = new PulsarContainer();
-
     @Autowired
     private TestConsumers testConsumers;
 
+    @Autowired
+    private TopicUrlService topicUrlService;
+
+    @Container
+    static PulsarContainer pulsarContainer = new PulsarContainer();
+
     public static final String VALIDATION_STRING = "validation-string";
 
     @DynamicPropertySource
@@ -124,7 +128,7 @@ void testConsumerRegistration1() throws Exception {
         Assertions.assertEquals(9, consumers.size());
 
         final Consumer<?> consumer =
-            consumers.stream().filter($ -> $.getTopic().equals("topic-one")).findFirst().orElseThrow(Exception::new);
+            consumers.stream().filter($ -> $.getTopic().equals(topicUrlService.buildTopicUrl("topic-one"))).findFirst().orElseThrow(Exception::new);
 
         Assertions.assertNotNull(consumer);
     }
@@ -165,7 +169,7 @@ void testMessageErrorHandling() throws PulsarClientException {
         final AtomicBoolean receivedError = new AtomicBoolean(false);
         final String messageToSend = "This message will never arrive.";
         final Disposable disposable = consumerAggregator.onError(($) -> {
-            Assertions.assertEquals($.getConsumer().getTopic(), "topic-for-error");
+            Assertions.assertEquals($.getConsumer().getTopic(), topicUrlService.buildTopicUrl("topic-for-error"));
             Assertions.assertEquals($.getMessage().getValue(), messageToSend);
             Assertions.assertNotNull($.getException());