[clean][client] Clean code for the construction of retry/dead letter topic name (apache#24082)

(cherry picked from commit a608810)
(cherry picked from commit cf10ac9)

Author: thetumbled

pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerDisallowAutoCreateTopicTest.java

@@ -18,11 +18,11 @@
  */
 package org.apache.pulsar.client.api;
 
-import static org.apache.pulsar.client.util.RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.util.RetryMessageUtil;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
@@ -54,7 +54,7 @@ protected void doInitConf() throws Exception {
     public void testClearErrorIfRetryTopicNotExists() throws Exception {
         final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp_");
         final String subName = "sub";
-        final String retryTopicName = topicName + "-" + subName + RETRY_GROUP_TOPIC_SUFFIX;
+        final String retryTopicName = RetryMessageUtil.getRetryTopic(topicName, subName);
         admin.topics().createNonPartitionedTopic(topicName);
         Consumer consumer = null;
         try {

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java

@@ -1500,8 +1500,7 @@ public void testSendTxnAckMessageToDLQ() throws Exception {
 
         @Cleanup
         Consumer<byte[]> deadLetterConsumer = pulsarClient.newConsumer()
-                .topic(String.format("%s-%s" + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX,
-                        topic, subName))
+                .topic(RetryMessageUtil.getDLQTopic(topic, subName))
                 .subscriptionType(SubscriptionType.Shared)
                 .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(1).build())
                 .subscriptionName("test")
@@ -1536,8 +1535,7 @@ public void testSendTxnAckMessageToDLQ() throws Exception {
         consumer.close();
         deadLetterConsumer.close();
         producer.close();
-        admin.topics().delete(String.format("%s-%s" + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX,
-                topic, subName), true);
+        admin.topics().delete(RetryMessageUtil.getDLQTopic(topic, subName), true);
         admin.topics().delete(topic, true);
     }
 
@@ -1564,8 +1562,7 @@ public void testSendTxnAckBatchMessageToDLQ() throws Exception {
 
         @Cleanup
         Consumer<byte[]> deadLetterConsumer = pulsarClient.newConsumer()
-                .topic(String.format("%s-%s" + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX,
-                        topic, subName))
+                .topic(RetryMessageUtil.getDLQTopic(topic, subName))
                 .subscriptionType(SubscriptionType.Shared)
                 .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(1).build())
                 .subscriptionName("test")
@@ -1611,8 +1608,7 @@ public void testSendTxnAckBatchMessageToDLQ() throws Exception {
         consumer.close();
         deadLetterConsumer.close();
         producer.close();
-        admin.topics().delete(String.format("%s-%s" + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX,
-                topic, subName), true);
+        admin.topics().delete(RetryMessageUtil.getDLQTopic(topic, subName), true);
         admin.topics().delete(topic, true);
     }
 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java

@@ -165,10 +165,10 @@ public CompletableFuture<Consumer<T>> subscribeAsync() {
                 CompletableFuture<Boolean> deadLetterTopicMetadata = checkDlqAlreadyExists(oldDeadLetterTopic);
                 applyDLQConfig = CompletableFuture.allOf(retryLetterTopicMetadata, deadLetterTopicMetadata)
                         .thenAccept(__ -> {
-                            String retryLetterTopic = topicFirst + "-" + conf.getSubscriptionName()
-                                    + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
-                            String deadLetterTopic = topicFirst + "-" + conf.getSubscriptionName()
-                                    + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX;
+                            String retryLetterTopic = RetryMessageUtil.getRetryTopic(topicFirst.toString(),
+                                    conf.getSubscriptionName());
+                            String deadLetterTopic = RetryMessageUtil.getDLQTopic(topicFirst.toString(),
+                                    conf.getSubscriptionName());
                             if (retryLetterTopicMetadata.join()) {
                                 retryLetterTopic = oldRetryLetterTopic;
                             }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -395,16 +395,13 @@ protected ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurat
             } else {
                 this.deadLetterPolicy = DeadLetterPolicy.builder()
                         .maxRedeliverCount(conf.getDeadLetterPolicy().getMaxRedeliverCount())
-                        .deadLetterTopic(String.format("%s-%s" + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX,
-                                topic, subscription))
+                        .deadLetterTopic(RetryMessageUtil.getDLQTopic(topic, subscription))
                         .build();
             }
             if (StringUtils.isNotBlank(conf.getDeadLetterPolicy().getRetryLetterTopic())) {
                 this.deadLetterPolicy.setRetryLetterTopic(conf.getDeadLetterPolicy().getRetryLetterTopic());
             } else {
-                this.deadLetterPolicy.setRetryLetterTopic(String.format(
-                        "%s-%s" + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX,
-                        topic, subscription));
+                this.deadLetterPolicy.setRetryLetterTopic(RetryMessageUtil.getRetryTopic(topic, subscription));
             }
             if (StringUtils.isNotBlank(conf.getDeadLetterPolicy().getInitialSubscriptionName())) {
                 this.deadLetterPolicy.setInitialSubscriptionName(

pulsar-client/src/main/java/org/apache/pulsar/client/util/RetryMessageUtil.java

@@ -31,4 +31,12 @@ public class RetryMessageUtil {
     public static final int MAX_RECONSUMETIMES = 16;
     public static final String RETRY_GROUP_TOPIC_SUFFIX = "-RETRY";
     public static final String DLQ_GROUP_TOPIC_SUFFIX = "-DLQ";
+
+    public static String getRetryTopic(String topic, String subscription) {
+        return topic + "-" + subscription + RETRY_GROUP_TOPIC_SUFFIX;
+    }
+
+    public static String getDLQTopic(String topic, String subscription) {
+        return topic + "-" + subscription + DLQ_GROUP_TOPIC_SUFFIX;
+    }
 }