[ServiceBus] Improve AMQP Error handling (#16427)

* improve error handling

* Update sdk/servicebus/azure-servicebus/CHANGELOG.md

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## 7.0.2 (Unreleased)
 
+**BugFixes**
+
+* Operations failing due to `uamqp.errors.LinkForceDetach` caused by no activity on the connection for 10 minutes will now be retried internally except for the session receiver case.
+* `uamqp.errors.AMQPConnectionError` errors with condition code `amqp:unknown-error` are now categorized into `ServiceBusConnectionError` instead of the general `ServiceBusError`.
 
 ## 7.0.1 (2021-01-12)
 

sdk/servicebus/azure-servicebus/azure/servicebus/exceptions.py

@@ -101,6 +101,8 @@ def _handle_amqp_exception_with_condition(
     elif condition == AMQPErrorCodes.ClientError and "timed out" in str(exception):
         # handle send timeout
         error_cls = OperationTimeoutError
+    elif condition == AMQPErrorCodes.UnknownError and isinstance(exception, AMQPErrors.AMQPConnectionError):
+        error_cls = ServiceBusConnectionError
     else:
         # handle other error codes
         error_cls = _ERROR_CODE_TO_ERROR_MAPPING.get(condition, ServiceBusError)
@@ -491,6 +493,7 @@ class AutoLockRenewTimeout(ServiceBusError):
     AMQPErrorCodes.UnauthorizedAccess: ServiceBusAuthorizationError,
     AMQPErrorCodes.NotImplemented: ServiceBusError,
     AMQPErrorCodes.NotAllowed: ServiceBusError,
+    AMQPErrorCodes.LinkDetachForced: ServiceBusConnectionError,
     ERROR_CODE_MESSAGE_LOCK_LOST: MessageLockLostError,
     ERROR_CODE_MESSAGE_NOT_FOUND: MessageNotFoundError,
     ERROR_CODE_AUTH_FAILED: ServiceBusAuthorizationError,

sdk/servicebus/azure-servicebus/tests/livetest/test_errors.py

@@ -0,0 +1,33 @@
+import logging
+
+from uamqp import errors as AMQPErrors, constants as AMQPConstants
+from azure.servicebus.exceptions import (
+    _create_servicebus_exception,
+    ServiceBusConnectionError,
+    ServiceBusError
+)
+
+
+def test_link_idle_timeout():
+    logger = logging.getLogger("testlogger")
+    amqp_error = AMQPErrors.LinkDetach(AMQPConstants.ErrorCodes.LinkDetachForced, description="Details: AmqpMessageConsumer.IdleTimerExpired: Idle timeout: 00:10:00.")
+    sb_error = _create_servicebus_exception(logger, amqp_error)
+    assert isinstance(sb_error, ServiceBusConnectionError)
+    assert sb_error._retryable
+    assert sb_error._shutdown_handler
+
+
+def test_unknown_connection_error():
+    logger = logging.getLogger("testlogger")
+    amqp_error = AMQPErrors.AMQPConnectionError(AMQPConstants.ErrorCodes.UnknownError)
+    sb_error = _create_servicebus_exception(logger, amqp_error)
+    assert isinstance(sb_error,ServiceBusConnectionError)
+    assert sb_error._retryable
+    assert sb_error._shutdown_handler
+
+    amqp_error = AMQPErrors.AMQPError(AMQPConstants.ErrorCodes.UnknownError)
+    sb_error = _create_servicebus_exception(logger, amqp_error)
+    assert not isinstance(sb_error,ServiceBusConnectionError)
+    assert isinstance(sb_error,ServiceBusError)
+    assert not sb_error._retryable
+    assert sb_error._shutdown_handler