feat: Support dead letter topic.

Author: shibd

pulsar/__init__.py

@@ -48,7 +48,7 @@
 import _pulsar
 
 from _pulsar import Result, CompressionType, ConsumerType, InitialPosition, PartitionsRoutingMode, BatchingType, \
-    LoggerLevel, BatchReceivePolicy, KeySharedPolicy, KeySharedMode  # noqa: F401
+    LoggerLevel, BatchReceivePolicy, KeySharedPolicy, KeySharedMode, DeadLetterPolicy, DeadLetterPolicyBuilder  # noqa: F401
 
 from pulsar.__about__ import __version__
 
@@ -374,6 +374,64 @@ def __init__(self, username=None, password=None, method='basic', auth_params_str
             _check_type(str, method, 'method')
             self.auth = _pulsar.AuthenticationBasic.create(username, password, method)
 
+class ConsumerDeadLetterPolicy:
+    """
+    Configuration for the "dead letter queue" feature in consumer.
+    """
+    def __init__(self, dead_letter_topic: str = None,
+                 max_redeliver_count: int = None,
+                 initial_subscription_name: str = None):
+        """
+        Wrapper DeadLetterPolicy.
+
+        Parameters
+        ----------
+        dead_letter_topic: Name of the dead topic where the failing messages are sent.
+            The default value is: sourceTopicName + "-" + subscriptionName + "-DLQ"
+        max_redeliver_count: Maximum number of times that a message is redelivered before being sent to the dead letter queue.
+            - The maxRedeliverCount must be greater than 0.
+            - The default value is None (DLQ is not enabled)
+        initial_subscription_name: Name of the initial subscription name of the dead letter topic.
+            If this field is not set, the initial subscription for the dead letter topic is not created.
+            If this field is set but the broker's `allowAutoSubscriptionCreation` is disabled, the DLQ producer
+            fails to be created.
+        """
+        builder = DeadLetterPolicyBuilder()
+        if dead_letter_topic is not None:
+            builder.deadLetterTopic(dead_letter_topic)
+        if max_redeliver_count is not None:
+            builder.maxRedeliverCount(max_redeliver_count)
+        if initial_subscription_name is not None:
+            builder.initialSubscriptionName(initial_subscription_name)
+        self._policy = builder.build()
+
+    @property
+    def dead_letter_topic(self) -> str:
+        """
+        Return the dead letter topic for dead letter policy.
+        """
+        return self._policy.getDeadLetterTopic()
+
+    @property
+    def max_redeliver_count(self) -> int:
+        """
+        Return the max redeliver count for dead letter policy.
+        """
+        return self._policy.getMaxRedeliverCount()
+
+    @property
+    def initial_subscription_name(self) -> str:
+        """
+        Return the initial subscription name for dead letter policy.
+        """
+        return self._policy.getInitialSubscriptionName()
+
+    def policy(self):
+        """
+        Returns the actual one DeadLetterPolicy.
+        """
+        return self._policy
+
 class Client:
     """
     The Pulsar client. A single client instance can be used to create producers
@@ -694,6 +752,7 @@ def subscribe(self, topic, subscription_name,
                   batch_receive_policy=None,
                   key_shared_policy=None,
                   batch_index_ack_enabled=False,
+                  dead_letter_policy: ConsumerDeadLetterPolicy = None,
                   ):
         """
         Subscribe to the given topic and subscription combination.
@@ -783,6 +842,12 @@ def my_listener(consumer, message):
         batch_index_ack_enabled: Enable the batch index acknowledgement.
             It should be noted that this option can only work when the broker side also enables the batch index
             acknowledgement. See the `acknowledgmentAtBatchIndexLevelEnabled` config in `broker.conf`.
+        dead_letter_policy: class ConsumerDeadLetterPolicy
+          Set dead letter policy for consumer.
+          By default, some messages are redelivered many times, even to the extent that they can never be
+          stopped. By using the dead letter mechanism, messages have the max redelivery count, when they're
+          exceeding the maximum number of redeliveries. Messages are sent to dead letter topics and acknowledged
+          automatically.
         """
         _check_type(str, subscription_name, 'subscription_name')
         _check_type(ConsumerType, consumer_type, 'consumer_type')
@@ -840,6 +905,8 @@ def my_listener(consumer, message):
         if key_shared_policy:
             conf.key_shared_policy(key_shared_policy.policy())
         conf.batch_index_ack_enabled(batch_index_ack_enabled)
+        if dead_letter_policy:
+            conf.dead_letter_policy(dead_letter_policy.policy())
 
         c = Consumer()
         if isinstance(topic, str):

src/config.cc

@@ -22,6 +22,7 @@
 #include <pulsar/ConsumerConfiguration.h>
 #include <pulsar/ProducerConfiguration.h>
 #include <pulsar/KeySharedPolicy.h>
+#include <pulsar/DeadLetterPolicyBuilder.h>
 #include <pybind11/functional.h>
 #include <pybind11/pybind11.h>
 #include <pybind11/stl.h>
@@ -231,6 +232,20 @@ void export_config(py::module_& m) {
         .def("getMaxNumMessages", &BatchReceivePolicy::getMaxNumMessages)
         .def("getMaxNumBytes", &BatchReceivePolicy::getMaxNumBytes);
 
+    class_<DeadLetterPolicy>(m, "DeadLetterPolicy")
+        .def(init<>())
+        .def("getDeadLetterTopic", &DeadLetterPolicy::getDeadLetterTopic)
+        .def("getMaxRedeliverCount", &DeadLetterPolicy::getMaxRedeliverCount)
+        .def("getInitialSubscriptionName", &DeadLetterPolicy::getInitialSubscriptionName);
+
+    class_<DeadLetterPolicyBuilder>(m, "DeadLetterPolicyBuilder")
+        .def(init<>())
+        .def("deadLetterTopic", &DeadLetterPolicyBuilder::deadLetterTopic, return_value_policy::reference)
+        .def("maxRedeliverCount", &DeadLetterPolicyBuilder::maxRedeliverCount, return_value_policy::reference)
+        .def("initialSubscriptionName", &DeadLetterPolicyBuilder::initialSubscriptionName, return_value_policy::reference)
+        .def("build", &DeadLetterPolicyBuilder::build, return_value_policy::reference)
+        .def("build", &DeadLetterPolicyBuilder::build, return_value_policy::reference);
+
     class_<ConsumerConfiguration, std::shared_ptr<ConsumerConfiguration>>(m, "ConsumerConfiguration")
         .def(init<>())
         .def("key_shared_policy", &ConsumerConfiguration::getKeySharedPolicy)
@@ -283,7 +298,9 @@ void export_config(py::module_& m) {
              return_value_policy::reference)
         .def("batch_index_ack_enabled", &ConsumerConfiguration::isBatchIndexAckEnabled)
         .def("batch_index_ack_enabled", &ConsumerConfiguration::setBatchIndexAckEnabled,
-             return_value_policy::reference);
+             return_value_policy::reference)
+        .def("dead_letter_policy", &ConsumerConfiguration::setDeadLetterPolicy)
+        .def("dead_letter_policy", &ConsumerConfiguration::getDeadLetterPolicy, return_value_policy::copy);
 
     class_<ReaderConfiguration, std::shared_ptr<ReaderConfiguration>>(m, "ReaderConfiguration")
         .def(init<>())

tests/pulsar_test.py

@@ -42,6 +42,7 @@
     InitialPosition,
     CryptoKeyReader,
     ConsumerBatchReceivePolicy,
+    ConsumerDeadLetterPolicy,
 )
 from pulsar.schema import JsonSchema, Record, Integer
 
@@ -1657,6 +1658,40 @@ def test_batch_index_ack(self):
         # assert no more msgs.
         with self.assertRaises(pulsar.Timeout):
             consumer.receive(timeout_millis=1000)
+        client.close()
+
+    def test_dead_letter_policy(self):
+        client = Client(self.serviceUrl)
+        topic = "my-python-topic-test-dlq" + str(time.time())
+        dlq_topic = 'dlq-' + topic
+        max_redeliver_count = 5
+        consumer = client.subscribe(topic, "my-sub", consumer_type=ConsumerType.Shared,
+                                    dead_letter_policy=ConsumerDeadLetterPolicy(dlq_topic, max_redeliver_count, 'init-sub'))
+        dlq_consumer = client.subscribe(dlq_topic, "my-sub", consumer_type=ConsumerType.Shared)
+
+        # Sen num msgs.
+        producer = client.create_producer(topic)
+        num = 10
+        for i in range(num):
+            producer.send(b"hello-%d" % i)
+        producer.flush()
+
+        # Redelivery all messages maxRedeliverCountNum time.
+        for i in range(1, num * max_redeliver_count + num + 1):
+            msg = consumer.receive()
+            if i % num == 0:
+                consumer.redeliver_unacknowledged_messages()
+                print(f"Start redeliver msgs '{i}'")
+        with self.assertRaises(pulsar.Timeout):
+            consumer.receive(100)
+
+        for i in range(num):
+            msg = dlq_consumer.receive()
+            self.assertTrue(msg)
+            self.assertEqual(msg.data(), b"hello-%d" % i)
+            dlq_consumer.acknowledge(msg)
+        with self.assertRaises(pulsar.Timeout):
+            dlq_consumer.receive(100)
 
         client.close()