[GitHub] merlimat closed pull request #1272: Add prefix setting for Java reader

[GitBox <gi...@apache.org>]

merlimat closed pull request #1272: Add prefix setting for Java reader
URL: https://github.com/apache/incubator-pulsar/pull/1272
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java
index 196af2057..8c378993a 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java
@@ -137,4 +137,11 @@
      * @param readerName
      */
     ReaderBuilder readerName(String readerName);
+
+    /**
+     * Set the subscription role prefix. The default prefix is "reader".
+     *
+     * @param subscriptionRolePrefix
+     */
+    ReaderBuilder subscriptionRolePrefix(String subscriptionRolePrefix);
 }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderConfiguration.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderConfiguration.java
index cdd51a514..bca04e598 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderConfiguration.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderConfiguration.java
@@ -18,6 +18,8 @@
  */
 package org.apache.pulsar.client.api;
 
+import org.apache.commons.lang3.StringUtils;
+
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
@@ -130,11 +132,29 @@ public String getReaderName() {
      * @param readerName
      */
     public ReaderConfiguration setReaderName(String readerName) {
-        checkArgument(readerName != null && !readerName.equals(""));
+        checkArgument(StringUtils.isNotBlank(readerName));
         conf.setReaderName(readerName);
         return this;
     }
 
+    /**
+     * @return the subscription role prefix for subscription auth
+     */
+    public String getSubscriptionRolePrefix() {
+        return conf.getSubscriptionRolePrefix();
+    }
+
+    /**
+     * Set the subscription role prefix for subscription auth. The default prefix is "reader".
+     *
+     * @param subscriptionRolePrefix
+     */
+    public ReaderConfiguration setSubscriptionRolePrefix(String subscriptionRolePrefix) {
+        checkArgument(StringUtils.isNotBlank(subscriptionRolePrefix));
+        conf.setSubscriptionRolePrefix(subscriptionRolePrefix);
+        return this;
+    }
+
     public ReaderConfigurationData getReaderConfigurationData() {
         return conf;
     }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java
index eb1aeb8c9..9a537b15e 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java
@@ -130,4 +130,10 @@ public ReaderBuilder readerName(String readerName) {
         conf.setReaderName(readerName);
         return this;
     }
+
+    @Override
+    public ReaderBuilder subscriptionRolePrefix(String subscriptionRolePrefix) {
+        conf.setSubscriptionRolePrefix(subscriptionRolePrefix);
+        return this;
+    }
 }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
index aaa3902b8..aaa1e4f61 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
@@ -25,6 +25,7 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageListener;
@@ -44,6 +45,9 @@ public ReaderImpl(PulsarClientImpl client, ReaderConfigurationData readerConfigu
             ExecutorService listenerExecutor, CompletableFuture<Consumer> consumerFuture) {
 
         String subscription = "reader-" + DigestUtils.sha1Hex(UUID.randomUUID().toString()).substring(0, 10);
+        if (StringUtils.isNotBlank(readerConfiguration.getSubscriptionRolePrefix())) {
+            subscription = readerConfiguration.getSubscriptionRolePrefix() + "-" + subscription;
+        }
 
         ConsumerConfigurationData consumerConfiguration = new ConsumerConfigurationData();
         consumerConfiguration.getTopicNames().add(readerConfiguration.getTopicName());
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java
index d1323ee9e..d369eec33 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java
@@ -38,6 +38,7 @@
     private ReaderListener readerListener;
 
     private String readerName = null;
+    private String subscriptionRolePrefix = null;
 
     private CryptoKeyReader cryptoKeyReader = null;
     private ConsumerCryptoFailureAction cryptoFailureAction = ConsumerCryptoFailureAction.FAIL;


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services