[ServiceBus] Track2 - Dead Letter Queue Receiver Implementation and Dead Letter Fix (Azure#11535)

* dead letter fix

* fix mypy and update dependency

* draft implementation of dead letter queue receiver, re-enable dead letter test

* helpify entity name generation in dead letter queue

* tweak test

* merge kieran commit

* complete dlq feature

* fix pylint

* add dlq samples

* fix pylint

* update according to review comment

* move DLQ path generation outside

* tweak change log

* fix pylint error

* fix pylint

* move idletimeout and prefetch out of config

* fix typo

Co-authored-by: Kieran Brantner-Magee <[email protected]>

Author: yunhaoling

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -2,6 +2,14 @@
 
 ## 7.0.0b3 (Unreleased)
 
+**New Features**
+
+* Added methods `get_queue_deadletter_receiver` and `get_subscription_deadletter_receiver` in `ServiceBusClient` to get a `ServiceBusReceiver` for the dead-letter sub-queue of the target entity.
+
+**BugFixes**
+
+* Updated uAMQP dependency to 1.2.8.
+    * Fixed bug where reason and description were not being set when dead-lettering messages.
 
 ## 7.0.0b2 (2020-05-04)
 
@@ -23,7 +31,7 @@
 
 **BugFixes**
 
-* Fig bug where http_proxy and transport_type in ServiceBusClient are not propagated into Sender/Receiver creation properly.
+* Fixed bug where http_proxy and transport_type in ServiceBusClient are not propagated into Sender/Receiver creation properly.
 * Updated uAMQP dependency to 1.2.7.
     * Fixed bug in setting certificate of tlsio on MacOS. #7201
     * Fixed bug that caused segmentation fault in network tracing on MacOS when setting `logging_enable` to `True` in `ServiceBusClient`.

sdk/servicebus/azure-servicebus/azure/servicebus/_common/_configuration.py

@@ -23,8 +23,3 @@ def __init__(self, **kwargs):
         self.auth_timeout = kwargs.get("auth_timeout", 60)  # type: int
         self.encoding = kwargs.get("encoding", "UTF-8")
         self.auto_reconnect = kwargs.get("auto_reconnect", True)
-        self.idle_timeout = kwargs.get("idle_timeout", None)
-        prefetch = kwargs.get("prefetch", 0)
-        if int(prefetch) < 0 or int(prefetch) > 50000:
-            raise ValueError("Prefetch must be an integer between 0 and 50000 inclusive.")
-        self.prefetch = prefetch + 1

sdk/servicebus/azure-servicebus/azure/servicebus/_common/constants.py

@@ -71,6 +71,8 @@
 MGMT_REQUEST_VIA_PARTITION_KEY = 'via-partition-key'
 MGMT_REQUEST_DEAD_LETTER_REASON = 'deadletter-reason'
 MGMT_REQUEST_DEAD_LETTER_DESCRIPTION = 'deadletter-description'
+RECEIVER_LINK_DEAD_LETTER_REASON = 'DeadLetterReason'
+RECEIVER_LINK_DEAD_LETTER_DESCRIPTION = 'DeadLetterErrorDescription'
 MGMT_REQUEST_OP_TYPE_ENTITY_MGMT = b"entity-mgmt"
 
 MESSAGE_COMPLETE = 'complete'
@@ -97,6 +99,10 @@
 _X_OPT_SCHEDULED_ENQUEUE_TIME = b'x-opt-scheduled-enqueue-time'
 
 
+DEAD_LETTER_QUEUE_SUFFIX = '/$DeadLetterQueue'
+TRANSFER_DEAD_LETTER_QUEUE_SUFFIX = '/$Transfer' + DEAD_LETTER_QUEUE_SUFFIX
+
+
 class ReceiveSettleMode(Enum):
     PeekLock = constants.ReceiverSettleMode.PeekLock
     ReceiveAndDelete = constants.ReceiverSettleMode.ReceiveAndDelete

sdk/servicebus/azure-servicebus/azure/servicebus/_common/message.py

@@ -32,6 +32,8 @@
     MGMT_RESPONSE_MESSAGE_EXPIRATION,
     MGMT_REQUEST_DEAD_LETTER_REASON,
     MGMT_REQUEST_DEAD_LETTER_DESCRIPTION,
+    RECEIVER_LINK_DEAD_LETTER_REASON,
+    RECEIVER_LINK_DEAD_LETTER_DESCRIPTION,
     MESSAGE_COMPLETE,
     MESSAGE_DEAD_LETTER,
     MESSAGE_ABANDON,
@@ -539,8 +541,8 @@ def _check_live(self, action):
         except AttributeError:
             pass
 
-    def _settle_via_mgmt_link(self, settle_operation, dead_letter_details=None):
-        # type: (str, Dict[str, Any]) -> Callable
+    def _settle_via_mgmt_link(self, settle_operation, dead_letter_reason=None, dead_letter_description=None):
+        # type: (str, Optional[str], Optional[str]) -> Callable
         # pylint: disable=protected-access
         if settle_operation == MESSAGE_COMPLETE:
             return functools.partial(
@@ -559,7 +561,10 @@ def _settle_via_mgmt_link(self, settle_operation, dead_letter_details=None):
                 self._receiver._settle_message,
                 SETTLEMENT_DEADLETTER,
                 [self.lock_token],
-                dead_letter_details=dead_letter_details
+                dead_letter_details={
+                    MGMT_REQUEST_DEAD_LETTER_REASON: dead_letter_reason or "",
+                    MGMT_REQUEST_DEAD_LETTER_DESCRIPTION: dead_letter_description or ""
+                }
             )
         if settle_operation == MESSAGE_DEFER:
             return functools.partial(
@@ -569,34 +574,39 @@ def _settle_via_mgmt_link(self, settle_operation, dead_letter_details=None):
             )
         raise ValueError("Unsupported settle operation type: {}".format(settle_operation))
 
-    def _settle_via_receiver_link(self, settle_operation, dead_letter_details=None):  # pylint: disable=unused-argument
-        # type: (str, Dict[str, Any]) -> Callable
-        # dead_letter_detail is not used because of uamqp receiver link doesn't accept it while it
-        # should be accepted. Will revisit this later.
-        # uamqp management link accepts dead_letter_details. Refer to method _settle_via_mgmt_link
-        # TODO: to make dead_letter_details useful
+    def _settle_via_receiver_link(self, settle_operation, dead_letter_reason=None, dead_letter_description=None):
+        # type: (str, Optional[str], Optional[str]) -> Callable
         if settle_operation == MESSAGE_COMPLETE:
             return functools.partial(self.message.accept)
         if settle_operation == MESSAGE_ABANDON:
             return functools.partial(self.message.modify, True, False)
         if settle_operation == MESSAGE_DEAD_LETTER:
-            # note: message.reject() can not set reason and description properly due to the issue
-            # https://github.com/Azure/azure-uamqp-python/issues/155
-            return functools.partial(self.message.reject, condition=DEADLETTERNAME)
+            return functools.partial(
+                self.message.reject,
+                condition=DEADLETTERNAME,
+                description=dead_letter_description,
+                info={
+                    RECEIVER_LINK_DEAD_LETTER_REASON: dead_letter_reason,
+                    RECEIVER_LINK_DEAD_LETTER_DESCRIPTION: dead_letter_description
+                }
+            )
         if settle_operation == MESSAGE_DEFER:
             return functools.partial(self.message.modify, True, True)
         raise ValueError("Unsupported settle operation type: {}".format(settle_operation))
 
     def _settle_message(
             self,
             settle_operation,
-            dead_letter_details=None
+            dead_letter_reason=None,
+            dead_letter_description=None,
     ):
-        # type: (str, Dict[str, Any]) -> None
+        # type: (str, Optional[str], Optional[str]) -> None
         try:
             if not self._is_deferred_message:
                 try:
-                    self._settle_via_receiver_link(settle_operation, dead_letter_details)()
+                    self._settle_via_receiver_link(settle_operation,
+                                                   dead_letter_reason=dead_letter_reason,
+                                                   dead_letter_description=dead_letter_description)()
                     return
                 except RuntimeError as exception:
                     _LOGGER.info(
@@ -605,7 +615,9 @@ def _settle_message(
                         settle_operation,
                         exception
                     )
-            self._settle_via_mgmt_link(settle_operation, dead_letter_details)()
+            self._settle_via_mgmt_link(settle_operation,
+                                       dead_letter_reason=dead_letter_reason,
+                                       dead_letter_description=dead_letter_description)()
         except Exception as e:
             raise MessageSettleFailed(settle_operation, e)
 
@@ -644,12 +656,7 @@ def dead_letter(self, reason=None, description=None):
         """
         # pylint: disable=protected-access
         self._check_live(MESSAGE_DEAD_LETTER)
-
-        details = {
-            MGMT_REQUEST_DEAD_LETTER_REASON: str(reason) if reason else "",
-            MGMT_REQUEST_DEAD_LETTER_DESCRIPTION: str(description) if description else ""}
-
-        self._settle_message(MESSAGE_DEAD_LETTER, dead_letter_details=details)
+        self._settle_message(MESSAGE_DEAD_LETTER, dead_letter_reason=reason, dead_letter_description=description)
         self._settled = True
 
     def abandon(self):

sdk/servicebus/azure-servicebus/azure/servicebus/_common/receiver_mixins.py

@@ -22,7 +22,7 @@
 
 
 class ReceiverMixin(object):  # pylint: disable=too-many-instance-attributes
-    def _create_attribute(self, **kwargs):
+    def _populate_attributes(self, **kwargs):
         if kwargs.get("subscription_name"):
             self._subscription_name = kwargs.get("subscription_name")
             self._is_subscription = True
@@ -38,6 +38,13 @@ def _create_attribute(self, **kwargs):
         )
         self._name = "SBReceiver-{}".format(uuid.uuid4())
         self._last_received_sequenced_number = None
+        self._message_iter = None
+        self._connection = kwargs.get("connection")
+        prefetch = kwargs.get("prefetch", 0)
+        if int(prefetch) < 0 or int(prefetch) > 50000:
+            raise ValueError("Prefetch must be an integer between 0 and 50000 inclusive.")
+        self._prefetch = prefetch + 1
+        self._idle_timeout = kwargs.get("idle_timeout", None)
 
     def _build_message(self, received, message_type=ReceivedMessage):
         message = message_type(message=received, mode=self._mode)
@@ -82,7 +89,7 @@ def _check_live(self):
         if self._session and self._session.expired:
             raise SessionLockExpired(inner_exception=self._session.auto_renew_error)
 
-    def _create_session_attributes(self, **kwargs):
+    def _populate_session_attributes(self, **kwargs):
         self._session_id = kwargs.get("session_id") or NEXT_AVAILABLE
         self._error_policy = _ServiceBusErrorPolicy(
             max_retries=self._config.retry_total,

sdk/servicebus/azure-servicebus/azure/servicebus/_common/utils.py

@@ -23,7 +23,9 @@
 from .constants import (
     JWT_TOKEN_SCOPE,
     TOKEN_TYPE_JWT,
-    TOKEN_TYPE_SASTOKEN
+    TOKEN_TYPE_SASTOKEN,
+    DEAD_LETTER_QUEUE_SUFFIX,
+    TRANSFER_DEAD_LETTER_QUEUE_SUFFIX
 )
 
 _log = logging.getLogger(__name__)
@@ -141,6 +143,21 @@ def create_authentication(client):
     )
 
 
+def generate_dead_letter_entity_name(
+        queue_name=None,
+        topic_name=None,
+        subscription_name=None,
+        transfer_deadletter=False
+):
+    entity_name = queue_name if queue_name else (topic_name + "/Subscriptions/" + subscription_name)
+    entity_name = "{}{}".format(
+        entity_name,
+        TRANSFER_DEAD_LETTER_QUEUE_SUFFIX if transfer_deadletter else DEAD_LETTER_QUEUE_SUFFIX
+    )
+
+    return entity_name
+
+
 class AutoLockRenew(object):
     """Auto renew locks for messages and sessions using a background thread pool.