From 59cfddf41d4c01d05f26c51217860ef6ae27fab8 Mon Sep 17 00:00:00 2001
From: thetumbled <843221020@qq.com>
Date: Tue, 25 Feb 2025 12:10:56 +0800
Subject: [PATCH 1/5] add pip-409.

---
 pip/pip-409.md | 132 +++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 132 insertions(+)
 create mode 100644 pip/pip-409.md

diff --git a/pip/pip-409.md b/pip/pip-409.md
new file mode 100644
index 0000000000000..ebf671f3a8f23
--- /dev/null
+++ b/pip/pip-409.md
@@ -0,0 +1,132 @@
+
+# PIP-409: support producer configuration for retry/dead letter topic producer
+
+# Background knowledge
+
+Retry topic is a topic that stores messages that have failed to be processed by the consumer, or the consumer can't process the message at the moment. 
+The consumer can retry the message after a certain period of time. After several retries, the message will be moved to the dead letter topic.
+
+There is a retry topic producer in consumer instance. Each time the consumer can call `consumer.reconsumeLater(msg, 3, TimeUnit.SECONDS);` 
+to retry the message after 3 seconds. The hidden producer will send the corresponding message to retry topic, and ack the message in original topic.
+
+After several retries, the message will be sent to dead letter topic instead of retry topic.
+
+
+# Motivation
+
+Currently, we don't support configure the producer of retry/dead letter topic.
+But enable the chunk message feature and disable the batch configuration in hard code, which can't handle many 
+situations. For example, when the throughput of message of retry topic become considerable, the resource consumed
+by the un-batched messages is pretty large. There is no reason that we disable the batch message feature.
+
+For better control for the retry/dead letter topic feature, we can support configuration for the producer of 
+retry/dead letter topic.
+
+# Goals
+
+## In Scope
+
+- Support configuration for the producer of retry/dead letter topic.
+
+
+# Detailed Design
+
+## Design & Implementation Details
+
+<!--
+This is the section where you dive into the details. It can be:
+* Concrete class names and their roles and responsibility, including methods.
+* Code snippets of existing code.
+* Interface names and its methods.
+* ...
+-->
+
+## Public-facing Changes
+
+<!--
+Describe the additions you plan to make for each public facing component. 
+Remove the sections you are not changing.
+Clearly mark any changes which are BREAKING backward compatability.
+-->
+
+### Public API
+<!--
+When adding a new endpoint to the REST API, please make sure to document the following:
+
+* path
+* query parameters
+* HTTP body parameters, usually as JSON.
+* Response codes, and for each what they mean.
+  For each response code, please include a detailed description of the response body JSON, specifying each field and what it means.
+  This is the place to document the errors.
+-->
+
+### Binary protocol
+
+### Configuration
+
+### CLI
+
+### Metrics
+
+<!--
+For each metric provide:
+* Full name
+* Description
+* Attributes (labels)
+* Unit
+-->
+
+
+# Monitoring
+
+<!-- 
+Describe how the changes you make in this proposal should be monitored. 
+Don't describe the detailed metrics - they should be at "Public-facing Changes" / "Metrics" section.
+Describe how the user will use the metrics to monitor the feature: Which alerts they should set up, which thresholds, ...
+-->
+
+# Security Considerations
+<!--
+A detailed description of the security details that ought to be considered for the PIP. This is most relevant for any new HTTP endpoints, new Pulsar Protocol Commands, and new security features. The goal is to describe details like which role will have permission to perform an action.
+
+An important aspect to consider is also multi-tenancy: Does the feature I'm adding have the permissions / roles set in such a way that prevent one tenant accessing another tenant's data/configuration? For example, the Admin API to read a specific message for a topic only allows a client to read messages for the target topic. However, that was not always the case. CVE-2021-41571 (https://github.com/apache/pulsar/wiki/CVE-2021-41571) resulted because the API was incorrectly written and did not properly prevent a client from reading another topic's messages even though authorization was in place. The problem was missing input validation that verified the requested message was actually a message for that topic. The fix to CVE-2021-41571 was input validation. 
+
+If there is uncertainty for this section, please submit the PIP and request for feedback on the mailing list.
+-->
+
+# Backward & Forward Compatibility
+
+## Upgrade
+
+<!--
+Specify the list of instructions, if there are such, needed to perform before/after upgrading to Pulsar version containing this feature.
+-->
+
+## Downgrade / Rollback
+
+<!--
+Describe a cookbook detailing the steps required to rollback Pulsar to previous version *without* this feature.
+-->
+
+## Pulsar Geo-Replication Upgrade & Downgrade/Rollback Considerations
+
+<!--
+Describe what needs to be considered in Pulsar Geo-Replication in the upgrade and possible downgrade/rollback of this feature.
+-->
+
+# Alternatives
+
+<!--
+If there are alternatives that were already considered by the authors or, after the discussion, by the community, and were rejected, please list them here along with the reason why they were rejected.
+-->
+
+# General Notes
+
+# Links
+
+<!--
+Updated afterwards
+-->
+* Mailing List discussion thread:
+* Mailing List voting thread:

From 107e021094735e6513d33a998b3dc1687930f7b8 Mon Sep 17 00:00:00 2001
From: thetumbled <843221020@qq.com>
Date: Tue, 25 Feb 2025 15:26:24 +0800
Subject: [PATCH 2/5] add doc.

---
 pip/pip-409.md | 99 +++++++++++---------------------------------------
 1 file changed, 21 insertions(+), 78 deletions(-)

diff --git a/pip/pip-409.md b/pip/pip-409.md
index ebf671f3a8f23..4d5952b25f13e 100644
--- a/pip/pip-409.md
+++ b/pip/pip-409.md
@@ -33,93 +33,36 @@ retry/dead letter topic.
 
 ## Design & Implementation Details
 
-<!--
-This is the section where you dive into the details. It can be:
-* Concrete class names and their roles and responsibility, including methods.
-* Code snippets of existing code.
-* Interface names and its methods.
-* ...
--->
-
-## Public-facing Changes
-
-<!--
-Describe the additions you plan to make for each public facing component. 
-Remove the sections you are not changing.
-Clearly mark any changes which are BREAKING backward compatability.
--->
-
-### Public API
-<!--
-When adding a new endpoint to the REST API, please make sure to document the following:
-
-* path
-* query parameters
-* HTTP body parameters, usually as JSON.
-* Response codes, and for each what they mean.
-  For each response code, please include a detailed description of the response body JSON, specifying each field and what it means.
-  This is the place to document the errors.
--->
+- add a new class `DeadLetterProducerConfig`.
+Instead of reusing `ProducerConfigurationData`, we introduce a new config class. First of all, this is because `ProducerConfigurationData`
+lies in the `pulsar-client-original` module, and we don't want to introduce it as a new dependency in the `pulsar-client-api` module.
+Secondly, we actually don't need all the configurations in `ProducerConfigurationData` for the producer of retry/dead letter topic.
 
-### Binary protocol
+- add two new configurations in `DeadLetterPolicy`, both of them are of type `DeadLetterProducerConfig`.
+  - `retryTopicProducerConfig`: the configuration for the producer of retry topic.
+  - `deadLetterProducerConfig`: the configuration for the producer of dead letter topic.
 
-### Configuration
+```java
+public class DeadLetterPolicy implements Serializable {
+    /**
+     * Configuration used to create a producer that will send messages to the dead letter topic.
+     */
+    private DeadLetterProducerConfig deadLetterProducerConfig;
 
-### CLI
-
-### Metrics
-
-<!--
-For each metric provide:
-* Full name
-* Description
-* Attributes (labels)
-* Unit
--->
-
-
-# Monitoring
-
-<!-- 
-Describe how the changes you make in this proposal should be monitored. 
-Don't describe the detailed metrics - they should be at "Public-facing Changes" / "Metrics" section.
-Describe how the user will use the metrics to monitor the feature: Which alerts they should set up, which thresholds, ...
--->
+    /**
+     * Configuration used to create a producer that will send messages to the retry letter topic.
+     */
+    private DeadLetterProducerConfig retryLetterProducerConfig;
+}
+```
 
-# Security Considerations
-<!--
-A detailed description of the security details that ought to be considered for the PIP. This is most relevant for any new HTTP endpoints, new Pulsar Protocol Commands, and new security features. The goal is to describe details like which role will have permission to perform an action.
+- use the configuration in `DeadLetterPolicy` to create the producer of retry/dead letter topic.
 
-An important aspect to consider is also multi-tenancy: Does the feature I'm adding have the permissions / roles set in such a way that prevent one tenant accessing another tenant's data/configuration? For example, the Admin API to read a specific message for a topic only allows a client to read messages for the target topic. However, that was not always the case. CVE-2021-41571 (https://github.com/apache/pulsar/wiki/CVE-2021-41571) resulted because the API was incorrectly written and did not properly prevent a client from reading another topic's messages even though authorization was in place. The problem was missing input validation that verified the requested message was actually a message for that topic. The fix to CVE-2021-41571 was input validation. 
 
-If there is uncertainty for this section, please submit the PIP and request for feedback on the mailing list.
--->
 
 # Backward & Forward Compatibility
 
-## Upgrade
-
-<!--
-Specify the list of instructions, if there are such, needed to perform before/after upgrading to Pulsar version containing this feature.
--->
-
-## Downgrade / Rollback
-
-<!--
-Describe a cookbook detailing the steps required to rollback Pulsar to previous version *without* this feature.
--->
-
-## Pulsar Geo-Replication Upgrade & Downgrade/Rollback Considerations
-
-<!--
-Describe what needs to be considered in Pulsar Geo-Replication in the upgrade and possible downgrade/rollback of this feature.
--->
-
-# Alternatives
-
-<!--
-If there are alternatives that were already considered by the authors or, after the discussion, by the community, and were rejected, please list them here along with the reason why they were rejected.
--->
+No backward or forward compatibility issues.
 
 # General Notes
 

From 5d5ff795edeb2df3d863bc92f71a4a49ac29aa91 Mon Sep 17 00:00:00 2001
From: thetumbled <843221020@qq.com>
Date: Tue, 25 Feb 2025 15:34:44 +0800
Subject: [PATCH 3/5] add link.

---
 pip/pip-409.md | 16 +++++++---------
 1 file changed, 7 insertions(+), 9 deletions(-)

diff --git a/pip/pip-409.md b/pip/pip-409.md
index 4d5952b25f13e..76742cbfe3ace 100644
--- a/pip/pip-409.md
+++ b/pip/pip-409.md
@@ -6,18 +6,18 @@
 Retry topic is a topic that stores messages that have failed to be processed by the consumer, or the consumer can't process the message at the moment. 
 The consumer can retry the message after a certain period of time. After several retries, the message will be moved to the dead letter topic.
 
-There is a retry topic producer in consumer instance. Each time the consumer can call `consumer.reconsumeLater(msg, 3, TimeUnit.SECONDS);` 
-to retry the message after 3 seconds. The hidden producer will send the corresponding message to retry topic, and ack the message in original topic.
+There is a retry topic producer in consumer instance. Each time the consumer call `consumer.reconsumeLater(msg, 3, TimeUnit.SECONDS);` 
+to retry the message after 3 seconds, the hidden producer will send the corresponding message to retry topic, and ack the message in original topic.
 
 After several retries, the message will be sent to dead letter topic instead of retry topic.
 
 
 # Motivation
 
-Currently, we don't support configure the producer of retry/dead letter topic.
-But enable the chunk message feature and disable the batch configuration in hard code, which can't handle many 
-situations. For example, when the throughput of message of retry topic become considerable, the resource consumed
-by the un-batched messages is pretty large. There is no reason that we disable the batch message feature.
+Currently, we don't support configure the producer of retry/dead letter topic. But enable the chunk message feature
+and disable the batch configuration in hard code, which can't handle many situations. For example, when the throughput 
+of message of retry topic become considerable, the resource consumed by the un-batched messages is pretty large. 
+There is no reason that we disable the batch message feature.
 
 For better control for the retry/dead letter topic feature, we can support configuration for the producer of 
 retry/dead letter topic.
@@ -64,12 +64,10 @@ public class DeadLetterPolicy implements Serializable {
 
 No backward or forward compatibility issues.
 
-# General Notes
-
 # Links
 
 <!--
 Updated afterwards
 -->
-* Mailing List discussion thread:
+* Mailing List discussion thread: https://lists.apache.org/thread/h6jjf9wn2h4zmpjw5zjtnl5ds1r4nknq
 * Mailing List voting thread:

From 212d0da6e684a568c2d54f33558f9f57627c2d04 Mon Sep 17 00:00:00 2001
From: thetumbled <843221020@qq.com>
Date: Wed, 26 Feb 2025 17:26:36 +0800
Subject: [PATCH 4/5] update doc.

---
 pip/pip-409.md | 39 ++++++++++++++++++---------------------
 1 file changed, 18 insertions(+), 21 deletions(-)

diff --git a/pip/pip-409.md b/pip/pip-409.md
index 76742cbfe3ace..3c43d7a1298a3 100644
--- a/pip/pip-409.md
+++ b/pip/pip-409.md
@@ -33,36 +33,33 @@ retry/dead letter topic.
 
 ## Design & Implementation Details
 
-- add a new class `DeadLetterProducerConfig`.
-Instead of reusing `ProducerConfigurationData`, we introduce a new config class. First of all, this is because `ProducerConfigurationData`
-lies in the `pulsar-client-original` module, and we don't want to introduce it as a new dependency in the `pulsar-client-api` module.
-Secondly, we actually don't need all the configurations in `ProducerConfigurationData` for the producer of retry/dead letter topic.
-
-- add two new configurations in `DeadLetterPolicy`, both of them are of type `DeadLetterProducerConfig`.
-  - `retryTopicProducerConfig`: the configuration for the producer of retry topic.
-  - `deadLetterProducerConfig`: the configuration for the producer of dead letter topic.
-
+- Add two new configurations in `DeadLetterPolicy`:
 ```java
 public class DeadLetterPolicy implements Serializable {
-    /**
-     * Configuration used to create a producer that will send messages to the dead letter topic.
-     */
-    private DeadLetterProducerConfig deadLetterProducerConfig;
-
-    /**
-     * Configuration used to create a producer that will send messages to the retry letter topic.
-     */
-    private DeadLetterProducerConfig retryLetterProducerConfig;
+  /**
+   * Function to build the producer for the retry letter topic.
+   * The input parameter is the topic name.
+   */
+  private Function<String, ProducerBuilder<byte[]>> retryLetterProducerBuilder;
+
+  /**
+   * Function to build the producer for the dead letter topic.
+   * The input parameter is the topic name.
+   */
+  private Function<String, ProducerBuilder<byte[]>> deadLetterProducerBuilder;
 }
 ```
 
-- use the configuration in `DeadLetterPolicy` to create the producer of retry/dead letter topic.
-
+- use the `retryLetterProducerBuilder` to build the producer for retry topic, and use the
+`deadLetterProducerBuilder` to build the producer for dead letter topic.
 
 
 # Backward & Forward Compatibility
 
-No backward or forward compatibility issues.
+Default behavior changes:
+- Before this PIP, the batch message feature is disabled and the chunk message feature is enabled in the producer of retry/dead letter topic.
+- After this PIP, the batch message feature is enabled and the chunk message feature is disabled in the producer of retry/dead letter topic,
+which is consistent with the default value of `ProducerConfigurationData`.
 
 # Links
 

From 189c977b4d7ba4fabbb55a6b351dfebb6b94ae8c Mon Sep 17 00:00:00 2001
From: thetumbled <843221020@qq.com>
Date: Thu, 27 Feb 2025 15:15:39 +0800
Subject: [PATCH 5/5] update compatible description and add vote thread.

---
 pip/pip-409.md | 7 ++-----
 1 file changed, 2 insertions(+), 5 deletions(-)

diff --git a/pip/pip-409.md b/pip/pip-409.md
index 3c43d7a1298a3..bc91c27610b85 100644
--- a/pip/pip-409.md
+++ b/pip/pip-409.md
@@ -56,10 +56,7 @@ public class DeadLetterPolicy implements Serializable {
 
 # Backward & Forward Compatibility
 
-Default behavior changes:
-- Before this PIP, the batch message feature is disabled and the chunk message feature is enabled in the producer of retry/dead letter topic.
-- After this PIP, the batch message feature is enabled and the chunk message feature is disabled in the producer of retry/dead letter topic,
-which is consistent with the default value of `ProducerConfigurationData`.
+Fully compatible.
 
 # Links
 
@@ -67,4 +64,4 @@ which is consistent with the default value of `ProducerConfigurationData`.
 Updated afterwards
 -->
 * Mailing List discussion thread: https://lists.apache.org/thread/h6jjf9wn2h4zmpjw5zjtnl5ds1r4nknq
-* Mailing List voting thread:
+* Mailing List voting thread: https://lists.apache.org/thread/6wgxovk0f72d10zdgsnto6bkh6pwfzj1