[fix][client] Fix race-condition causing doReconsumeLater to hang when creating retryLetterProducer has failed (apache#23560)

(cherry picked from commit bf1f677)
(cherry picked from commit 5f25367)

Author: hanmz

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

@@ -30,6 +30,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import lombok.Cleanup;
 import lombok.Data;
@@ -45,6 +46,7 @@
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
+import org.testng.collections.Lists;
 
 @Test(groups = "broker-api")
 public class RetryTopicTest extends ProducerConsumerBase {
@@ -713,4 +715,70 @@ public void testRetryProducerWillCloseByConsumer() throws Exception {
         admin.topics().delete(topicDLQ, false);
     }
 
+
+    @Test(timeOut = 30000L)
+    public void testRetryTopicExceptionWithConcurrent() throws Exception {
+        final String topic = "persistent://my-property/my-ns/retry-topic";
+        final int maxRedeliveryCount = 2;
+        final int sendMessages = 10;
+        // subscribe before publish
+        Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
+                .topic(topic)
+                .subscriptionName("my-subscription")
+                .subscriptionType(SubscriptionType.Shared)
+                .enableRetry(true)
+                .receiverQueueSize(100)
+                .deadLetterPolicy(DeadLetterPolicy.builder()
+                        .maxRedeliverCount(maxRedeliveryCount)
+                        .retryLetterTopic("persistent://my-property/my-ns/my-subscription-custom-Retry")
+                        .build())
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscribe();
+
+        Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
+                .topic(topic)
+                .create();
+        for (int i = 0; i < sendMessages; i++) {
+            producer.newMessage().key("1").value(String.format("Hello Pulsar [%d]", i).getBytes()).send();
+        }
+        producer.close();
+
+        // mock a retry producer exception when reconsumelater is called
+        MultiTopicsConsumerImpl<byte[]> multiTopicsConsumer = (MultiTopicsConsumerImpl<byte[]>) consumer;
+        List<ConsumerImpl<byte[]>> consumers = multiTopicsConsumer.getConsumers();
+        for (ConsumerImpl<byte[]> c : consumers) {
+            Set<Field> deadLetterPolicyField =
+                    ReflectionUtils.getAllFields(c.getClass(), ReflectionUtils.withName("deadLetterPolicy"));
+
+            if (deadLetterPolicyField.size() != 0) {
+                Field field = deadLetterPolicyField.iterator().next();
+                field.setAccessible(true);
+                DeadLetterPolicy deadLetterPolicy = (DeadLetterPolicy) field.get(c);
+                deadLetterPolicy.setRetryLetterTopic("#persistent://invalid-topic#");
+            }
+        }
+
+        List<Message<byte[]>> messages = Lists.newArrayList();
+        for (int i = 0; i < sendMessages; i++) {
+            messages.add(consumer.receive());
+        }
+
+        // mock call the reconsumeLater method concurrently
+        CountDownLatch latch = new CountDownLatch(messages.size());
+        for (Message<byte[]> message : messages) {
+            new Thread(() -> {
+                try {
+                    consumer.reconsumeLater(message, 1, TimeUnit.SECONDS);
+                } catch (Exception ignore) {
+
+                } finally {
+                    latch.countDown();
+                }
+            }).start();
+        }
+
+        latch.await();
+        consumer.close();
+    }
+
 }

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

@@ -708,6 +708,8 @@ protected CompletableFuture<Void> doReconsumeLater(Message<?> message, AckType a
             } catch (Exception e) {
                 result.completeExceptionally(e);
             }
+        } else {
+            result.completeExceptionally(new PulsarClientException("Retry letter producer is null."));
         }
         MessageId finalMessageId = messageId;
         result.exceptionally(ex -> {