[fix][client] PIP-409 retry/dead letter topic producer config don't take effect. (#24071)

Author: thetumbled

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

@@ -24,6 +24,7 @@
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
+
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
@@ -45,6 +46,9 @@
 import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.client.api.schema.GenericRecord;
 import org.apache.pulsar.client.impl.ConsumerBuilderImpl;
+import org.apache.pulsar.client.impl.ConsumerImpl;
+import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
+import org.apache.pulsar.client.impl.ProducerImpl;
 import org.apache.pulsar.client.util.RetryMessageUtil;
 import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
 import org.awaitility.Awaitility;
@@ -1123,6 +1127,7 @@ public void testDeadLetterTopicWithProducerBuilder() throws Exception {
         // enable batch
         DeadLetterProducerBuilderCustomizer producerBuilderCustomizer = (context, producerBuilder) -> {
             producerBuilder.enableBatching(true);
+            producerBuilder.enableChunking(false);
         };
         String subscriptionName = "my-subscription";
         String subscriptionNameDLQ = "my-subscription-DLQ";
@@ -1183,6 +1188,23 @@ public void testDeadLetterTopicWithProducerBuilder() throws Exception {
         } while (totalInDeadLetter < sendMessages);
         assertTrue(messageContent.isEmpty());
 
+        // check the retry topic producer enable batch
+        List<ConsumerImpl<byte[]>> consumers = ((MultiTopicsConsumerImpl<byte[]>) consumer).getConsumers();
+        for (ConsumerImpl<byte[]> consumerImpl : consumers) {
+            Producer<byte[]> retryProducer = consumerImpl.getRetryLetterProducer();
+            Producer<byte[]> deadLetterProducer = consumerImpl.getDeadLetterProducer();
+            // there are two type of consumers in MultiTopicsConsumerImpl when enableRetry is true
+            // 1. consumer subscribe to original topic
+            // 2. consumer subscribe to retry topic
+            if (consumerImpl.getTopic().endsWith(RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX)) {
+                assertTrue(((ProducerImpl<byte[]>) retryProducer).isBatchMessagingEnabled());
+                assertTrue(((ProducerImpl<byte[]>) deadLetterProducer).isBatchMessagingEnabled());
+            } else {
+                assertTrue(((ProducerImpl<byte[]>) retryProducer).isBatchMessagingEnabled());
+                assertNull(deadLetterProducer);
+            }
+        }
+
         deadLetterConsumer.close();
         consumer.close();
 

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

@@ -146,6 +146,7 @@ public void testRetryTopicWithProducerBuilder() throws Exception {
         // enable batch
         DeadLetterProducerBuilderCustomizer producerBuilderCustomizer = (context, producerBuilder) -> {
             producerBuilder.enableBatching(true);
+            producerBuilder.enableChunking(false);
         };
         String subscriptionName = "my-subscription";
         String subscriptionNameDLQ = "my-subscription-DLQ";

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

@@ -400,20 +400,21 @@ protected ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurat
                                 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));
             }
-
             if (StringUtils.isNotBlank(conf.getDeadLetterPolicy().getInitialSubscriptionName())) {
                 this.deadLetterPolicy.setInitialSubscriptionName(
                         conf.getDeadLetterPolicy().getInitialSubscriptionName());
             }
-
+            this.deadLetterPolicy.setRetryLetterProducerBuilderCustomizer(
+                    conf.getDeadLetterPolicy().getRetryLetterProducerBuilderCustomizer());
+            this.deadLetterPolicy.setDeadLetterProducerBuilderCustomizer(
+                    conf.getDeadLetterPolicy().getDeadLetterProducerBuilderCustomizer());
         } else {
             deadLetterPolicy = null;
             possibleSendToDeadLetterTopicMessages = null;
@@ -3229,4 +3230,14 @@ enum SeekStatus {
         IN_PROGRESS,
         COMPLETED
     }
+
+    @VisibleForTesting
+    public Producer<byte[]> getRetryLetterProducer() {
+        return (retryLetterProducer == null || !retryLetterProducer.isDone()) ? null : retryLetterProducer.join();
+    }
+
+    @VisibleForTesting
+    public Producer<byte[]> getDeadLetterProducer() throws ExecutionException, InterruptedException {
+        return (deadLetterProducer == null || !deadLetterProducer.isDone()) ? null : deadLetterProducer.get();
+    }
 }

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

@@ -350,7 +350,7 @@ public ConnectionHandler getConnectionHandler() {
         return connectionHandler;
     }
 
-    private boolean isBatchMessagingEnabled() {
+    public boolean isBatchMessagingEnabled() {
         return conf.isBatchingEnabled();
     }