[fix][client] Create the retry producer async (apache#23157)

Co-authored-by: Ómar Yasin <omarkj@apple.com> (cherry picked from commit a025938) (cherry picked from commit f2e1920)
datastax · Aug 16, 2024 · 93f0546 · 93f0546
1 parent a469492
commit 93f0546
Showing 1 changed file with 19 additions and 0 deletions.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -2218,6 +2218,25 @@ private void initRetryLetterProducerIfNeeded() {
         }
     }
 
+    private void initRetryLetterProducerIfNeeded() {
+        if (retryLetterProducer == null) {
+            createProducerLock.writeLock().lock();
+            try {
+                if (retryLetterProducer == null) {
+                    retryLetterProducer = client
+                            .newProducer(Schema.AUTO_PRODUCE_BYTES(schema))
+                            .topic(this.deadLetterPolicy.getRetryLetterTopic())
+                            .enableBatching(false)
+                            .enableChunking(true)
+                            .blockIfQueueFull(false)
+                            .createAsync();
+                }
+            } finally {
+                createProducerLock.writeLock().unlock();
+            }
+        }
+    }
+
     @Override
     public void seek(MessageId messageId) throws PulsarClientException {
         try {