[service bus] Use ServiceBusException rather than AmqpException

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusClientBuilder.java

@@ -642,7 +642,7 @@ public final class ServiceBusSessionProcessorClientBuilder {
         private final ServiceBusProcessorClientOptions processorClientOptions;
         private final ServiceBusSessionReceiverClientBuilder sessionReceiverClientBuilder;
         private Consumer<ServiceBusReceivedMessageContext> processMessage;
-        private Consumer<Throwable> processError;
+        private Consumer<ServiceBusProcessErrorContext> processError;
 
         private ServiceBusSessionProcessorClientBuilder() {
             sessionReceiverClientBuilder = new ServiceBusSessionReceiverClientBuilder();
@@ -750,7 +750,8 @@ public ServiceBusSessionProcessorClientBuilder processMessage(
          *
          * @return The updated {@link ServiceBusProcessorClientBuilder} object
          */
-        public ServiceBusSessionProcessorClientBuilder processError(Consumer<Throwable> processError) {
+        public ServiceBusSessionProcessorClientBuilder processError(
+            Consumer<ServiceBusProcessErrorContext> processError) {
             this.processError = processError;
             return this;
         }
@@ -1095,7 +1096,7 @@ public final class ServiceBusProcessorClientBuilder {
         private final ServiceBusReceiverClientBuilder serviceBusReceiverClientBuilder;
         private final ServiceBusProcessorClientOptions processorClientOptions;
         private Consumer<ServiceBusReceivedMessageContext> processMessage;
-        private Consumer<Throwable> processError;
+        private Consumer<ServiceBusProcessErrorContext> processError;
 
         private ServiceBusProcessorClientBuilder() {
             serviceBusReceiverClientBuilder = new ServiceBusReceiverClientBuilder();
@@ -1184,7 +1185,7 @@ public ServiceBusProcessorClientBuilder processMessage(
          *
          * @return The updated {@link ServiceBusProcessorClientBuilder} object
          */
-        public ServiceBusProcessorClientBuilder processError(Consumer<Throwable> processError) {
+        public ServiceBusProcessorClientBuilder processError(Consumer<ServiceBusProcessErrorContext> processError) {
             this.processError = processError;
             return this;
         }

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusErrorSource.java

@@ -36,4 +36,12 @@ public final class ServiceBusErrorSource extends ExpandableStringEnum<ServiceBus
     /** Error while session is closed.*/
     public static final ServiceBusErrorSource CLOSE_SESSION = fromString("CLOSE_SESSION",
         ServiceBusErrorSource.class);
+
+    /** Error while sending a message.*/
+    public static final ServiceBusErrorSource SENDING = fromString("SENDING",
+            ServiceBusErrorSource.class);
+
+    /** Error while trying to do an operation on the management link. */
+    public static final ServiceBusErrorSource MANAGEMENT = fromString("MANAGEMENT",
+            ServiceBusErrorSource.class);
 }

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusException.java

@@ -0,0 +1,101 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.messaging.servicebus;
+
+import com.azure.core.amqp.exception.AmqpErrorCondition;
+import com.azure.core.amqp.exception.AmqpException;
+import com.azure.core.exception.AzureException;
+
+/**
+ * Defines {@link ServiceBusException} which has additional information about the operation that caused the
+ * error.
+ * @see ServiceBusErrorSource
+ */
+public final class ServiceBusException extends AzureException {
+    private final transient ServiceBusErrorSource errorSource;
+    private final transient ServiceBusFailureReason reason;
+    private final boolean isTransient;
+
+    /**
+     * @param throwable for the error happened.
+     * @param errorSource indicating which api caused the error.
+     */
+    public ServiceBusException(Throwable throwable, ServiceBusErrorSource errorSource) {
+        super(throwable.getMessage(), throwable);
+        this.errorSource = errorSource;
+
+        if (throwable instanceof AmqpException) {
+            AmqpException amqpException = (AmqpException) throwable;
+            reason = getServiceBusFailureReasonFromException(amqpException);
+            isTransient = amqpException.isTransient();
+        } else {
+            reason = ServiceBusFailureReason.GENERAL_ERROR;
+            isTransient = false;
+        }
+    }
+
+    /**
+     * Gets the {@link ServiceBusErrorSource} in case of any errors.
+     * @return the {@link ServiceBusErrorSource}
+     */
+    public ServiceBusErrorSource getErrorSource() {
+        return errorSource;
+    }
+
+    /**
+     * Gets the {@link ServiceBusFailureReason} in case of any errors.
+     * @return the {@link ServiceBusFailureReason}
+     */
+    public ServiceBusFailureReason getReason() {
+        return reason;
+    }
+
+    /**
+     * A boolean indicating if the exception is a transient error or not.
+     *
+     * @return returns true when user can retry the operation that generated the exception without additional
+     * intervention.
+     */
+    public boolean isTransient() {
+        return isTransient;
+    }
+
+    private ServiceBusFailureReason getServiceBusFailureReasonFromException(AmqpException throwable) {
+        final AmqpErrorCondition errorCondition = throwable.getErrorCondition();
+
+        if (errorCondition == null) {
+            return ServiceBusFailureReason.GENERAL_ERROR;
+        }
+
+        switch (errorCondition) {
+            case NOT_FOUND:
+                return ServiceBusFailureReason.MESSAGING_ENTITY_NOT_FOUND;
+            case MESSAGE_LOCK_LOST:
+                return ServiceBusFailureReason.MESSAGE_LOCK_LOST;
+            case MESSAGE_NOT_FOUND:
+                return ServiceBusFailureReason.MESSAGE_NOT_FOUND;
+            case LINK_PAYLOAD_SIZE_EXCEEDED:
+                return ServiceBusFailureReason.MESSAGE_SIZE_EXCEEDED;
+            case ENTITY_ALREADY_EXISTS:
+                return ServiceBusFailureReason.MESSAGING_ENTITY_ALREADY_EXISTS;
+            case ENTITY_DISABLED_ERROR:
+                return ServiceBusFailureReason.MESSAGING_ENTITY_DISABLED;
+            case RESOURCE_LIMIT_EXCEEDED:
+                return ServiceBusFailureReason.QUOTA_EXCEEDED;
+            case SERVER_BUSY_ERROR:
+                return ServiceBusFailureReason.SERVICE_BUSY;
+            case TIMEOUT_ERROR:
+                return ServiceBusFailureReason.SERVICE_TIMEOUT;
+            case SESSION_CANNOT_BE_LOCKED:
+                return ServiceBusFailureReason.SESSION_CANNOT_BE_LOCKED;
+            case SESSION_LOCK_LOST:
+                return ServiceBusFailureReason.SESSION_LOCK_LOST;
+            case UNAUTHORIZED_ACCESS:
+                return ServiceBusFailureReason.UNAUTHORIZED;
+            case PROTON_IO:         // does this mapping make sense?
+            default:
+                return ServiceBusFailureReason.GENERAL_ERROR;
+        }
+    }
+}

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusFailureReason.java

@@ -0,0 +1,73 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.messaging.servicebus;
+
+import com.azure.core.util.ExpandableStringEnum;
+
+/**
+ * The set of well-known reasons for an Service Bus operation failure that was the cause of an exception.
+ */
+public final class ServiceBusFailureReason extends ExpandableStringEnum<ServiceBusFailureReason> {
+    // NOTE: this list is intended to mirror the reasons we have in .net
+    // https://github.com/Azure/azure-sdk-for-net/blob/master/sdk/servicebus/Azure.Messaging
+    // .ServiceBus/src/Primitives/ServiceBusFailureReason.cs
+
+    /** The exception was the result of a general error within the client library. */
+    public static final ServiceBusFailureReason GENERAL_ERROR = fromString("GENERAL_ERROR",
+        ServiceBusFailureReason.class);
+
+    /** The lock on the message is lost. Callers should call attempt to receive and process the message again. */
+    public static final ServiceBusFailureReason MESSAGE_LOCK_LOST = fromString("MESSAGE_LOCK_LOST",
+        ServiceBusFailureReason.class);
+
+    /** The requested message was not found. */
+    public static final ServiceBusFailureReason MESSAGE_NOT_FOUND = fromString("MESSAGE_NOT_FOUND",
+        ServiceBusFailureReason.class);
+
+    /** A message is larger than the maximum size allowed for its transport. */
+    public static final ServiceBusFailureReason MESSAGE_SIZE_EXCEEDED = fromString("MESSAGE_SIZE_EXCEEDED",
+        ServiceBusFailureReason.class);
+
+    /** An entity with the same name exists under the same namespace. */
+    public static final ServiceBusFailureReason MESSAGING_ENTITY_ALREADY_EXISTS = fromString(
+        "MESSAGING_ENTITY_ALREADY_EXISTS", ServiceBusFailureReason.class);
+
+    /** The Messaging Entity is disabled. Enable the entity again using Portal. */
+    public static final ServiceBusFailureReason MESSAGING_ENTITY_DISABLED = fromString("MESSAGING_ENTITY_DISABLED",
+        ServiceBusFailureReason.class);
+
+    /** A Service Bus resource cannot be found by the Service Bus service. */
+    public static final ServiceBusFailureReason MESSAGING_ENTITY_NOT_FOUND = fromString("MESSAGING_ENTITY_NOT_FOUND",
+        ServiceBusFailureReason.class);
+
+    /** The quota applied to an Service Bus resource has been exceeded while interacting with the Azure Service Bus
+     * service. */
+    public static final ServiceBusFailureReason QUOTA_EXCEEDED = fromString("QUOTA_EXCEEDED",
+        ServiceBusFailureReason.class);
+
+    /** The Azure Service Bus service reports that it is busy in response to a client request to perform an operation
+     * . */
+    public static final ServiceBusFailureReason SERVICE_BUSY = fromString("SERVICE_BUSY",
+        ServiceBusFailureReason.class);
+
+    /** An operation or other request timed out while interacting with the Azure Service Bus service. */
+    public static final ServiceBusFailureReason SERVICE_TIMEOUT = fromString("SERVICE_TIMEOUT",
+        ServiceBusFailureReason.class);
+
+    /** There was a general communications error encountered when interacting with the Azure Service Bus service. */
+    public static final ServiceBusFailureReason SERVICE_COMMUNICATION_PROBLEM = fromString(
+        "SERVICE_COMMUNICATION_PROBLEM", ServiceBusFailureReason.class);
+
+    /** The requested session cannot be locked. */
+    public static final ServiceBusFailureReason SESSION_CANNOT_BE_LOCKED = fromString("SESSION_CANNOT_BE_LOCKED",
+        ServiceBusFailureReason.class);
+
+    /** The lock on the session has expired. Callers should request the session again. */
+    public static final ServiceBusFailureReason SESSION_LOCK_LOST = fromString("SESSION_LOCK_LOST",
+        ServiceBusFailureReason.class);
+
+    /** The user doesn't have access to the entity. */
+    public static final ServiceBusFailureReason UNAUTHORIZED = fromString("UNAUTHORIZED",
+        ServiceBusFailureReason.class);
+}

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusMessageBatch.java

@@ -101,10 +101,12 @@ public boolean tryAddMessage(final ServiceBusMessage serviceBusMessage) {
         try {
             size = getSize(serviceBusMessageUpdated, serviceBusMessageList.isEmpty());
         } catch (BufferOverflowException exception) {
-            throw logger.logExceptionAsWarning(new AmqpException(false, AmqpErrorCondition.LINK_PAYLOAD_SIZE_EXCEEDED,
-                String.format(Locale.US, "Size of the payload exceeded maximum message size: %s kb",
-                    maxMessageSize / 1024),
-                contextProvider.getErrorContext()));
+            final RuntimeException ex = new ServiceBusException(
+                    new AmqpException(false, AmqpErrorCondition.LINK_PAYLOAD_SIZE_EXCEEDED,
+                        String.format(Locale.US, "Size of the payload exceeded maximum message size: %s kb",
+                            maxMessageSize / 1024), contextProvider.getErrorContext()), ServiceBusErrorSource.SENDING);
+
+            throw logger.logExceptionAsWarning(ex);
         }
 
         synchronized (lock) {

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusProcessErrorContext.java

@@ -0,0 +1,60 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.messaging.servicebus;
+
+/**
+ * Context for errors handled by the Service Bus processor.
+ */
+public class ServiceBusProcessErrorContext {
+    private final Throwable exception;
+    private final ServiceBusErrorSource errorSource;
+    private final String fullyQualifiedNamespace;
+    private final String entityPath;
+
+    ServiceBusProcessErrorContext(Throwable throwable, String fullyQualifiedNamespace, String entityPath) {
+        this.exception = throwable;
+        this.fullyQualifiedNamespace = fullyQualifiedNamespace;
+        this.entityPath = entityPath;
+
+        if (throwable instanceof ServiceBusException) {
+            final ServiceBusException serviceBusException = ((ServiceBusException) throwable);
+            this.errorSource = serviceBusException.getErrorSource();
+        } else {
+            this.errorSource = ServiceBusErrorSource.RECEIVE;
+        }
+    }
+
+    /**
+     * Gets the exception that triggered the call to the error event handler.
+     * @return The exception that triggered the call to the error event handler.
+     */
+    public Throwable getException() {
+        return exception;
+    }
+
+    /**
+     * Gets the source associated with the error.
+     * @return The source associated with the error.
+     */
+    public ServiceBusErrorSource getErrorSource() {
+        return errorSource;
+    }
+
+    /**
+     * Gets the namespace name associated with the error event.
+     * @return The namespace name associated with the error event.
+     */
+    public String getFullyQualifiedNamespace() {
+        return fullyQualifiedNamespace;
+    }
+
+    /**
+     * Gets the entity path associated with the error event.
+     * @return The entity path associated with the error event.
+     */
+    public String getEntityPath() {
+        return entityPath;
+    }
+}
+

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusProcessorClient.java

@@ -39,7 +39,7 @@ public final class ServiceBusProcessorClient implements AutoCloseable {
     private final ServiceBusClientBuilder.ServiceBusSessionReceiverClientBuilder sessionReceiverBuilder;
     private final ServiceBusClientBuilder.ServiceBusReceiverClientBuilder receiverBuilder;
     private final Consumer<ServiceBusReceivedMessageContext> processMessage;
-    private final Consumer<Throwable> processError;
+    private final Consumer<ServiceBusProcessErrorContext> processError;
     private final ServiceBusProcessorClientOptions processorOptions;
     private final AtomicReference<Subscription> receiverSubscription = new AtomicReference<>();
     private final AtomicReference<ServiceBusReceiverAsyncClient> asyncClient = new AtomicReference<>();
@@ -55,8 +55,9 @@ public final class ServiceBusProcessorClient implements AutoCloseable {
      * @param processorOptions Options to configure this instance of the processor.
      */
     ServiceBusProcessorClient(ServiceBusClientBuilder.ServiceBusSessionReceiverClientBuilder sessionReceiverBuilder,
-                              Consumer<ServiceBusReceivedMessageContext> processMessage,
-                              Consumer<Throwable> processError, ServiceBusProcessorClientOptions processorOptions) {
+        Consumer<ServiceBusReceivedMessageContext> processMessage,
+        Consumer<ServiceBusProcessErrorContext> processError,
+        ServiceBusProcessorClientOptions processorOptions) {
         this.sessionReceiverBuilder = Objects.requireNonNull(sessionReceiverBuilder,
             "'sessionReceiverBuilder' cannot be null");
         this.processMessage = Objects.requireNonNull(processMessage, "'processMessage' cannot be null");
@@ -75,8 +76,8 @@ public final class ServiceBusProcessorClient implements AutoCloseable {
      * @param processorOptions Options to configure this instance of the processor.
      */
     ServiceBusProcessorClient(ServiceBusClientBuilder.ServiceBusReceiverClientBuilder receiverBuilder,
-                              Consumer<ServiceBusReceivedMessageContext> processMessage,
-                              Consumer<Throwable> processError, ServiceBusProcessorClientOptions processorOptions) {
+        Consumer<ServiceBusReceivedMessageContext> processMessage,
+        Consumer<ServiceBusProcessErrorContext> processError, ServiceBusProcessorClientOptions processorOptions) {
         this.receiverBuilder = Objects.requireNonNull(receiverBuilder, "'receiverBuilder' cannot be null");
         this.processMessage = Objects.requireNonNull(processMessage, "'processMessage' cannot be null");
         this.processError = Objects.requireNonNull(processError, "'processError' cannot be null");
@@ -168,7 +169,7 @@ public void onNext(ServiceBusMessageContext serviceBusMessageContext) {
                                 new ServiceBusReceivedMessageContext(receiverClient, serviceBusMessageContext);
                             processMessage.accept(serviceBusReceivedMessageContext);
                         } catch (Exception ex) {
-                            handleError(new ServiceBusReceiverException(ex, ServiceBusErrorSource.USER_CALLBACK));
+                            handleError(new ServiceBusException(ex, ServiceBusErrorSource.USER_CALLBACK));
                             if (!processorOptions.isDisableAutoComplete()) {
                                 logger.warning("Error when processing message. Abandoning message.", ex);
                                 abandonMessage(serviceBusMessageContext, receiverClient);
@@ -211,7 +212,10 @@ private void abandonMessage(ServiceBusMessageContext serviceBusMessageContext,
 
     private void handleError(Throwable throwable) {
         try {
-            processError.accept(throwable);
+            ServiceBusReceiverAsyncClient client = asyncClient.get();
+            final String fullyQualifiedNamespace = client.getFullyQualifiedNamespace();
+            final String entityPath = client.getEntityPath();
+            processError.accept(new ServiceBusProcessErrorContext(throwable, fullyQualifiedNamespace, entityPath));
         } catch (Exception ex) {
             logger.verbose("Error from error handler. Ignoring error.", ex);
         }