diff --git a/eng/code-quality-reports/src/main/java/com/azure/tools/checkstyle/checks/NoImplInPublicAPI.java b/eng/code-quality-reports/src/main/java/com/azure/tools/checkstyle/checks/NoImplInPublicAPI.java
index a21ed603ab72..99e9ee2ce794 100644
--- a/eng/code-quality-reports/src/main/java/com/azure/tools/checkstyle/checks/NoImplInPublicAPI.java
+++ b/eng/code-quality-reports/src/main/java/com/azure/tools/checkstyle/checks/NoImplInPublicAPI.java
@@ -18,8 +18,6 @@ public class NoImplInPublicAPI extends AbstractCheck {
     private static final String PARAM_TYPE_ERROR = "\"%s\" class is in an implementation package, and it should not be used as a parameter type in public API. Alternatively, it can be removed from the implementation package and made public API, after appropriate API review.";
     private static final String RETURN_TYPE_ERROR = "\"%s\" class is in an implementation package, and it should not be a return type from public API. Alternatively, it can be removed from the implementation package and made public API.";
 
-    private static boolean isTrackTwo;
-    private static boolean isImplPackage;
     private Set<String> implementationClassSet = new HashSet<>();
 
     @Override
@@ -43,28 +41,11 @@ public int[] getRequiredTokens() {
 
     @Override
     public void beginTree(DetailAST root) {
-        this.isImplPackage = false;
-        this.isTrackTwo = false;
         this.implementationClassSet.clear();
     }
 
     @Override
     public void visitToken(DetailAST ast) {
-        if (ast.getType() == TokenTypes.PACKAGE_DEF) {
-            String packageName = FullIdent.createFullIdent(ast.findFirstToken(TokenTypes.DOT)).getText();
-            this.isTrackTwo = packageName.startsWith(COM_AZURE);
-            this.isImplPackage = packageName.contains(DOT_IMPLEMENTATION);
-            return;
-        } else {
-            if (this.isTrackTwo) {
-                if (this.isImplPackage) {
-                    return;
-                }
-            } else {
-                return;
-            }
-        }
-
         switch (ast.getType()) {
             case TokenTypes.IMPORT:
                 String importClassPath = FullIdent.createFullIdentBelow(ast).getText();
diff --git a/eng/code-quality-reports/src/main/java/com/azure/tools/checkstyle/checks/ThrowFromClientLoggerCheck.java b/eng/code-quality-reports/src/main/java/com/azure/tools/checkstyle/checks/ThrowFromClientLoggerCheck.java
new file mode 100644
index 000000000000..fb8018d62286
--- /dev/null
+++ b/eng/code-quality-reports/src/main/java/com/azure/tools/checkstyle/checks/ThrowFromClientLoggerCheck.java
@@ -0,0 +1,111 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.tools.checkstyle.checks;
+
+import com.puppycrawl.tools.checkstyle.api.AbstractCheck;
+import com.puppycrawl.tools.checkstyle.api.DetailAST;
+import com.puppycrawl.tools.checkstyle.api.FullIdent;
+import com.puppycrawl.tools.checkstyle.api.TokenTypes;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+/**
+ * To throw an exception, Must throw it through a 'logger.logExceptionAsError', rather than by directly calling 'throw exception'.
+ *
+ * Skip check if throwing exception from
+ * <ol>
+ *   <li>Static method</li>
+ *   <li>Static class</li>
+ *   <li>Constructor</li>
+ * </ol>
+ */
+public class ThrowFromClientLoggerCheck extends AbstractCheck {
+    private static final String LOGGER_LOG_EXCEPTION_AS_ERROR = "logger.logExceptionAsError";
+    private static final String LOGGER_LOG_EXCEPTION_AS_WARNING = "logger.logExceptionAsWarning";
+    private static final String THROW_LOGGER_EXCEPTION_MESSAGE = "Directly throwing an exception is disallowed. Must "
+        + "throw through ''ClientLogger'' API, either of ''%s'' or ''%s'' where ''logger'' is type of ClientLogger from Azure Core package.";
+
+    // A container stores the static status of class, skip this ThrowFromClientLoggerCheck if the class is static
+    private final Deque<Boolean> classStaticDeque = new ArrayDeque<>();
+    // A container stores the static status of method, skip this ThrowFromClientLoggerCheck if the method is static
+    private final Deque<Boolean> methodStaticDeque = new ArrayDeque<>();
+    // The variable is used to indicate if current node is still inside of constructor.
+    private boolean isInConstructor = false;
+
+    @Override
+    public int[] getDefaultTokens() {
+        return getRequiredTokens();
+    }
+
+    @Override
+    public int[] getAcceptableTokens() {
+        return getRequiredTokens();
+    }
+
+    @Override
+    public int[] getRequiredTokens() {
+        return new int[] {
+            TokenTypes.CLASS_DEF,
+            TokenTypes.CTOR_DEF,
+            TokenTypes.LITERAL_THROW,
+            TokenTypes.METHOD_DEF
+        };
+    }
+
+    @Override
+    public void leaveToken(DetailAST token) {
+        switch (token.getType()) {
+            case TokenTypes.CLASS_DEF:
+                classStaticDeque.pop();
+                break;
+            case TokenTypes.CTOR_DEF:
+                isInConstructor = false;
+                break;
+            case TokenTypes.METHOD_DEF:
+                methodStaticDeque.pop();
+                break;
+            default:
+                // Checkstyle complains if there's no default block in switch
+                break;
+        }
+    }
+
+    @Override
+    public void visitToken(DetailAST token) {
+        switch (token.getType()) {
+            case TokenTypes.CLASS_DEF:
+                DetailAST modifiersToken = token.findFirstToken(TokenTypes.MODIFIERS);
+                classStaticDeque.addLast(modifiersToken.branchContains(TokenTypes.LITERAL_STATIC));
+                break;
+            case TokenTypes.CTOR_DEF:
+                isInConstructor = true;
+                break;
+            case TokenTypes.METHOD_DEF:
+                DetailAST methodModifiersToken = token.findFirstToken(TokenTypes.MODIFIERS);
+                methodStaticDeque.addLast(methodModifiersToken.branchContains(TokenTypes.LITERAL_STATIC));
+                break;
+            case TokenTypes.LITERAL_THROW:
+                // Skip check if the throw exception from static class, constructor or static method
+                if (classStaticDeque.isEmpty() || classStaticDeque.peekLast() || isInConstructor
+                    || methodStaticDeque.isEmpty() || methodStaticDeque.peekLast()) {
+                    return;
+                }
+                DetailAST methodCallToken = token.findFirstToken(TokenTypes.EXPR).findFirstToken(TokenTypes.METHOD_CALL);
+                if (methodCallToken == null) {
+                    log(token, String.format(THROW_LOGGER_EXCEPTION_MESSAGE, LOGGER_LOG_EXCEPTION_AS_ERROR, LOGGER_LOG_EXCEPTION_AS_WARNING));
+                    return;
+                }
+
+                String methodCallName = FullIdent.createFullIdent(methodCallToken.findFirstToken(TokenTypes.DOT)).getText();
+                if (!LOGGER_LOG_EXCEPTION_AS_ERROR.equals(methodCallName) && !LOGGER_LOG_EXCEPTION_AS_WARNING.equals(methodCallName)) {
+                    log(token, String.format(THROW_LOGGER_EXCEPTION_MESSAGE, LOGGER_LOG_EXCEPTION_AS_ERROR, LOGGER_LOG_EXCEPTION_AS_WARNING));
+                }
+                break;
+            default:
+                // Checkstyle complains if there's no default block in switch
+                break;
+        }
+    }
+}
diff --git a/eng/code-quality-reports/src/main/resources/checkstyle/checkstyle-suppressions.xml b/eng/code-quality-reports/src/main/resources/checkstyle/checkstyle-suppressions.xml
index e5c597c93aaa..186bbb917d2f 100755
--- a/eng/code-quality-reports/src/main/resources/checkstyle/checkstyle-suppressions.xml
+++ b/eng/code-quality-reports/src/main/resources/checkstyle/checkstyle-suppressions.xml
@@ -96,19 +96,20 @@
 
   <!-- Remove this after fixing: https://github.com/Azure/azure-sdk-for-java/issues/5021 -->
   <suppress checks="com.azure.tools.checkstyle.checks.EnforceFinalFieldsCheck" files=".*\BlobOutputStream.java"/>
-  
+
+  <!-- Don't apply custom Checkstyle rules to files under test package -->
   <suppress checks="com\.azure\.tools\.checkstyle\.checks\.(ExternalDependencyExposedCheck|NoImplInPublicAPI|ServiceClientInstantiationCheck
   |ServiceClientBuilderCheck|ServiceInterfaceCheck|HttpPipelinePolicyCheck|JavaDocFormatting|JavadocThrowsChecks
-  |EnforceFinalFieldsCheck)" files=".*[/\\]src[/\\]test[/\\]java[/\\].*\.java"/>
+  |ThrowFromClientLoggerCheck|EnforceFinalFieldsCheck)" files=".*[/\\]src[/\\]test[/\\]java[/\\].*\.java"/>
 
-  <!-- Custom checkstyle rules that don't apply to files under implementation package -->
-  <suppress checks="com\.azure\.tools\.checkstyle\.checks\.(ExternalDependencyExposedCheck|ServiceClientInstantiationCheck
-  |ServiceClientBuilderCheck|ServiceInterfaceCheck|JavaDocFormatting|JavadocThrowsChecks|EnforceFinalFieldsCheck)"
+  <!-- Don't apply custom Checkstyle rules to files under implementation package -->
+  <suppress checks="com\.azure\.tools\.checkstyle\.checks\.(ExternalDependencyExposedCheck|NoImplInPublicAPI|ServiceClientInstantiationCheck
+  |ServiceClientBuilderCheck|ServiceInterfaceCheck|JavaDocFormatting|JavadocThrowsChecks|EnforceFinalFieldsCheck|ThrowFromClientLoggerCheck)"
   files=".*[/\\]implementation[/\\].*\.java"/>
 
-  <!-- Custom checkstyle rules that don't apply to files under samples package -->
+  <!-- Don't apply custom Checkstyle rules to files under samples package -->
   <suppress checks="com\.azure\.tools\.checkstyle\.checks\.(ExternalDependencyExposedCheck|HttpPipelinePolicyCheck|JavadocPackage
-  |EnforceFinalFieldsCheck)" files=".*[/\\]samples[/\\].*\.java"/>
+  |EnforceFinalFieldsCheck|ThrowFromClientLoggerCheck)" files=".*[/\\]samples[/\\].*\.java"/>
 
   <!-- Don't apply custom Checkstyle rules to files under checkstyle package. -->
   <suppress checks="com\.azure\.tools\.checkstyle\.checks\..+" files=".*[/\\]tools[/\\]checkstyle[/\\].*"/>
@@ -118,4 +119,14 @@
 
   <!-- Suppress warnings for Event Processor until the usage of "Client" is discussed and resolved: https://github.com/Azure/azure-sdk/issues/321 -->
   <suppress checks="com.azure.tools.checkstyle.checks.ServiceClientBuilderCheck" files=".*EventHubClientBuilder.java"/>
+
+  <!-- Any code in any package, it should never be a 'throw' keyword in the client library codebase except for in the client logger -->
+  <suppress checks="com.azure.tools.checkstyle.checks.ThrowFromClientLoggerCheck" files=".*[/\\]com[/\\]azure[/\\]core[/\\]util[/\\]logging[/\\]*"/>
+
+  <!-- Suppress IO exception for now, which need code owner's attention -->
+  <suppress checks="com.azure.tools.checkstyle.checks.ThrowFromClientLoggerCheck" files="com.azure.core.implementation.util.ClientLoggerJavaDocCodeSnippets.java"/>
+  <suppress checks="com.azure.tools.checkstyle.checks.ThrowFromClientLoggerCheck" files="com.azure.storage.blob.BlobInputStream.java"/>
+  <suppress checks="com.azure.tools.checkstyle.checks.ThrowFromClientLoggerCheck" files="com.azure.storage.blob.BlobOutputStream.java"/>
+  <suppress checks="com.azure.tools.checkstyle.checks.ThrowFromClientLoggerCheck" files="com.azure.storage.blob.BlockBlobClient.java"/>
+  <suppress checks="com.azure.tools.checkstyle.checks.ThrowFromClientLoggerCheck" files="com.azure.storage.blob.PageBlobAsyncClient.java"/>
 </suppressions>
diff --git a/eng/code-quality-reports/src/main/resources/checkstyle/checkstyle.xml b/eng/code-quality-reports/src/main/resources/checkstyle/checkstyle.xml
index d948fe1d7454..9e868e047b27 100755
--- a/eng/code-quality-reports/src/main/resources/checkstyle/checkstyle.xml
+++ b/eng/code-quality-reports/src/main/resources/checkstyle/checkstyle.xml
@@ -329,7 +329,7 @@ page at http://checkstyle.sourceforge.net/config.html -->
 
     <!--CUSTOM CHECKS-->
     <!-- Must use 'logger.logAndThrow' but not directly calling 'throw exception' -->
-<!--    <module name="com.azure.tools.checkstyle.checks.ThrownClientLoggerCheck"/>-->
+    <module name="com.azure.tools.checkstyle.checks.ThrowFromClientLoggerCheck"/>
 
     <!--CUSTOM CHECKS-->
     <!-- Any class that implements the HttpPipelinePolicy interface should:
diff --git a/sdk/appconfiguration/azure-data-appconfiguration/src/main/java/com/azure/data/appconfiguration/credentials/ConfigurationClientCredentials.java b/sdk/appconfiguration/azure-data-appconfiguration/src/main/java/com/azure/data/appconfiguration/credentials/ConfigurationClientCredentials.java
index 03658c75af7e..326069301f14 100644
--- a/sdk/appconfiguration/azure-data-appconfiguration/src/main/java/com/azure/data/appconfiguration/credentials/ConfigurationClientCredentials.java
+++ b/sdk/appconfiguration/azure-data-appconfiguration/src/main/java/com/azure/data/appconfiguration/credentials/ConfigurationClientCredentials.java
@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 package com.azure.data.appconfiguration.credentials;
 
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.data.appconfiguration.ConfigurationClientBuilder;
 import com.azure.data.appconfiguration.policy.ConfigurationCredentialsPolicy;
 import com.azure.core.implementation.util.ImplUtils;
@@ -37,6 +38,8 @@
  * @see ConfigurationClientBuilder
  */
 public class ConfigurationClientCredentials {
+    private final ClientLogger logger = new ClientLogger(ConfigurationClientCredentials.class);
+
     private static final String HOST_HEADER = "Host";
     private static final String DATE_HEADER = "Date";
     private static final String CONTENT_HASH_HEADER = "x-ms-content-sha256";
@@ -81,7 +84,7 @@ public Mono<Map<String, String>> getAuthorizationHeadersAsync(URL url, String ht
                 try {
                     return MessageDigest.getInstance("SHA-256");
                 } catch (NoSuchAlgorithmException e) {
-                    throw Exceptions.propagate(e);
+                    throw logger.logExceptionAsError(Exceptions.propagate(e));
                 }
             }, (messageDigest, byteBuffer) -> {
                     if (messageDigest != null) {
diff --git a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/PagedList.java b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/PagedList.java
index 3ff643e67723..d6066d6c3ad3 100644
--- a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/PagedList.java
+++ b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/PagedList.java
@@ -4,6 +4,7 @@
 package com.azure.core.management;
 
 import com.azure.core.exception.HttpResponseException;
+import com.azure.core.util.logging.ClientLogger;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -21,6 +22,8 @@
  * @param <E> the element type.
  */
 public abstract class PagedList<E> implements List<E> {
+    private final ClientLogger logger = new ClientLogger(PagedList.class);
+
     /** The actual items in the list. */
     private final List<E> items;
     /** Stores the latest page fetched. */
@@ -67,7 +70,7 @@ private void cachePage(String nextPageLink) {
                 }
             }
         } catch (IOException ex) {
-            throw new RuntimeException(ex);
+            throw logger.logExceptionAsError(new RuntimeException(ex));
         }
     }
 
@@ -137,6 +140,8 @@ protected void setCurrentPage(Page<E> currentPage) {
      * The implementation of {@link ListIterator} for PagedList.
      */
     private class ListItr implements ListIterator<E> {
+        private final ClientLogger logger = new ClientLogger(ListItr.class);
+
         /**
          * index of next element to return.
          */
@@ -164,7 +169,7 @@ public boolean hasNext() {
         public E next() {
             if (this.nextIndex >= items.size()) {
                 if (!hasNextPage()) {
-                    throw new NoSuchElementException();
+                    throw logger.logExceptionAsError(new NoSuchElementException());
                 } else {
                     loadNextPage();
                 }
@@ -179,7 +184,7 @@ public E next() {
                 } catch (IndexOutOfBoundsException ex) {
                     // The nextIndex got invalid means a different instance of iterator
                     // removed item from this index.
-                    throw new ConcurrentModificationException();
+                    throw logger.logExceptionAsError(new ConcurrentModificationException(ex));
                 }
             }
         }
@@ -187,14 +192,14 @@ public E next() {
         @Override
         public void remove() {
             if (this.lastRetIndex < 0) {
-                throw new IllegalStateException();
+                throw logger.logExceptionAsError(new IllegalStateException());
             } else {
                 try {
                     items.remove(this.lastRetIndex);
                     this.nextIndex = this.lastRetIndex;
                     this.lastRetIndex = -1;
                 } catch (IndexOutOfBoundsException ex) {
-                    throw new ConcurrentModificationException();
+                    throw logger.logExceptionAsError(new ConcurrentModificationException(ex));
                 }
             }
         }
@@ -208,16 +213,16 @@ public boolean hasPrevious() {
         public E previous() {
             int i = this.nextIndex - 1;
             if (i < 0) {
-                throw new NoSuchElementException();
+                throw logger.logExceptionAsError(new NoSuchElementException());
             } else if (i >= items.size()) {
-                throw new ConcurrentModificationException();
+                throw logger.logExceptionAsError(new ConcurrentModificationException());
             } else {
                 try {
                     this.nextIndex = i;
                     this.lastRetIndex = i;
                     return items.get(this.lastRetIndex);
                 } catch (IndexOutOfBoundsException ex) {
-                    throw new ConcurrentModificationException();
+                    throw logger.logExceptionAsError(new ConcurrentModificationException(ex));
                 }
             }
         }
@@ -235,12 +240,12 @@ public int previousIndex() {
         @Override
         public void set(E e) {
             if (this.lastRetIndex < 0) {
-                throw new IllegalStateException();
+                throw logger.logExceptionAsError(new IllegalStateException());
             } else {
                 try {
                     items.set(this.lastRetIndex, e);
                 } catch (IndexOutOfBoundsException ex) {
-                    throw new ConcurrentModificationException();
+                    throw logger.logExceptionAsError(new ConcurrentModificationException(ex));
                 }
             }
         }
@@ -252,7 +257,7 @@ public void add(E e) {
                 this.nextIndex = this.nextIndex + 1;
                 this.lastRetIndex = -1;
             } catch (IndexOutOfBoundsException ex) {
-                throw new ConcurrentModificationException();
+                throw logger.logExceptionAsError(new ConcurrentModificationException(ex));
             }
         }
     }
diff --git a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/AzureAsyncOperationPollStrategy.java b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/AzureAsyncOperationPollStrategy.java
index 54d1386f5cf4..8054078d0557 100644
--- a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/AzureAsyncOperationPollStrategy.java
+++ b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/AzureAsyncOperationPollStrategy.java
@@ -11,6 +11,7 @@
 import com.azure.core.management.AsyncOperationResource;
 import com.azure.core.management.CloudException;
 import com.azure.core.management.OperationState;
+import com.azure.core.util.logging.ClientLogger;
 import reactor.core.publisher.Mono;
 
 import java.io.IOException;
@@ -23,6 +24,8 @@
  * running operation.
  */
 public final class AzureAsyncOperationPollStrategy extends PollStrategy {
+    private final ClientLogger logger = new ClientLogger(AzureAsyncOperationPollStrategy.class);
+
     private AzureAsyncOperationPollStrategyData data;
 
     /**
@@ -100,7 +103,7 @@ public HttpRequest createPollRequest() {
                 pollUrl = data.originalResourceUrl;
             }
         } else {
-            throw new IllegalStateException("Polling is completed and did not succeed. Cannot create a polling request.");
+            throw logger.logExceptionAsError(new IllegalStateException("Polling is completed and did not succeed. Cannot create a polling request."));
         }
 
         return new HttpRequest(HttpMethod.GET, pollUrl);
@@ -122,7 +125,7 @@ public Mono<HttpResponse> updateFromAsync(HttpResponse httpPollResponse) {
                                     } catch (IOException ignored) { }
                                     //
                                     if (operationResource == null || operationResource.status() == null) {
-                                        throw new CloudException("The polling response does not contain a valid body", bufferedHttpPollResponse, null);
+                                        throw logger.logExceptionAsError(new CloudException("The polling response does not contain a valid body", bufferedHttpPollResponse, null));
                                     } else {
                                         final String status = operationResource.status();
                                         setStatus(status);
@@ -133,7 +136,7 @@ public Mono<HttpResponse> updateFromAsync(HttpResponse httpPollResponse) {
                                             clearDelayInMilliseconds();
 
                                             if (!data.pollingSucceeded) {
-                                                throw new CloudException("Async operation failed with provisioning state: " + status, bufferedHttpPollResponse);
+                                                throw logger.logExceptionAsError(new CloudException("Async operation failed with provisioning state: " + status, bufferedHttpPollResponse));
                                             }
 
                                             if (operationResource.id() != null) {
diff --git a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/AzureProxy.java b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/AzureProxy.java
index d892d00ce718..ecc119b3c06b 100644
--- a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/AzureProxy.java
+++ b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/AzureProxy.java
@@ -30,6 +30,7 @@
 import com.azure.core.management.annotations.AzureHost;
 import com.azure.core.management.serializer.AzureJacksonAdapter;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -52,6 +53,7 @@
  */
 public final class AzureProxy extends RestProxy {
     private static long defaultPollingDelayInMilliseconds = 30 * 1000;
+    private final ClientLogger logger = new ClientLogger(AzureProxy.class);
 
     /**
      * Create a new instance of RestProxy.
@@ -226,7 +228,7 @@ protected Object handleHttpResponse(final HttpRequest httpRequest, Mono<HttpDeco
         if (TypeUtil.isTypeOrSubTypeOf(returnType, Flux.class)) {
             final Type operationStatusType = ((ParameterizedType) returnType).getActualTypeArguments()[0];
             if (!TypeUtil.isTypeOrSubTypeOf(operationStatusType, OperationStatus.class)) {
-                throw new InvalidReturnTypeException("AzureProxy only supports swagger interface methods that return Flux (such as " + methodParser.fullyQualifiedMethodName() + "()) if the Flux's inner type that is OperationStatus (not " + returnType.toString() + ").");
+                throw logger.logExceptionAsError(new InvalidReturnTypeException("AzureProxy only supports swagger interface methods that return Flux (such as " + methodParser.fullyQualifiedMethodName() + "()) if the Flux's inner type that is OperationStatus (not " + returnType.toString() + ")."));
             } else {
                 // Get ResultTypeT in OperationStatus<ResultTypeT>
                 final Type operationStatusResultType = ((ParameterizedType) operationStatusType).getActualTypeArguments()[0];
@@ -257,7 +259,7 @@ protected Object handleResumeOperation(final HttpRequest httpRequest,
                                            Context context) {
         final Type operationStatusType = ((ParameterizedType) returnType).getActualTypeArguments()[0];
         if (!TypeUtil.isTypeOrSubTypeOf(operationStatusType, OperationStatus.class)) {
-            throw new InvalidReturnTypeException("AzureProxy only supports swagger interface methods that return Flux (such as " + methodParser.fullyQualifiedMethodName() + "()) if the Flux's inner type that is OperationStatus (not " + returnType.toString() + ").");
+            throw logger.logExceptionAsError(new InvalidReturnTypeException("AzureProxy only supports swagger interface methods that return Flux (such as " + methodParser.fullyQualifiedMethodName() + "()) if the Flux's inner type that is OperationStatus (not " + returnType.toString() + ")."));
         }
 
         PollStrategy.PollStrategyData pollStrategyData =
@@ -307,7 +309,7 @@ private Mono<PollStrategy> createPollStrategy(final HttpRequest originalHttpRequ
                                         if (pollStrategy == null) {
                                             pollStrategy = LocationPollStrategy.tryToCreate(AzureProxy.this, methodParser, originalHttpRequest, originalHttpResponse, delayInMilliseconds);
                                             if (pollStrategy == null) {
-                                                throw new CloudException("Response does not contain an Azure-AsyncOperation or Location header.", originalHttpResponse);
+                                                throw logger.logExceptionAsError(new CloudException("Response does not contain an Azure-AsyncOperation or Location header.", originalHttpResponse));
                                             }
                                         }
                                     }
@@ -342,7 +344,7 @@ private Mono<PollStrategy> createProvisioningStateOrCompletedPollStrategy(final
             pollStrategyMono = bufferedOriginalHttpResponse.bodyAsString()
                     .map(originalHttpResponseBody -> {
                         if (originalHttpResponseBody == null || originalHttpResponseBody.isEmpty()) {
-                            throw new CloudException("The HTTP response does not contain a body.", bufferedOriginalHttpResponse);
+                            throw logger.logExceptionAsError(new CloudException("The HTTP response does not contain a body.", bufferedOriginalHttpResponse));
                         }
                         PollStrategy pollStrategy;
                         try {
@@ -358,7 +360,7 @@ private Mono<PollStrategy> createProvisioningStateOrCompletedPollStrategy(final
                                                 AzureProxy.this, methodParser, bufferedOriginalHttpResponse));
                             }
                         } catch (IOException e) {
-                            throw Exceptions.propagate(e);
+                            throw logger.logExceptionAsError(Exceptions.propagate(e));
                         }
                         return pollStrategy;
                     });
diff --git a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/CompletedPollStrategy.java b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/CompletedPollStrategy.java
index cfda53c45496..0ab2012dbc36 100644
--- a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/CompletedPollStrategy.java
+++ b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/CompletedPollStrategy.java
@@ -9,6 +9,7 @@
 import com.azure.core.implementation.SwaggerMethodParser;
 import com.azure.core.management.OperationState;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -20,6 +21,8 @@
  * further polling.
  */
 public class CompletedPollStrategy extends PollStrategy {
+    private final ClientLogger logger = new ClientLogger(CompletedPollStrategy.class);
+
     private final HttpResponse firstHttpResponse;
     private CompletedPollStrategyData data;
 
@@ -65,7 +68,7 @@ PollStrategy initializeStrategy(RestProxy restProxy,
 
     @Override
     public HttpRequest createPollRequest() {
-        throw new UnsupportedOperationException();
+        throw logger.logExceptionAsError(new UnsupportedOperationException());
     }
 
     @Override
diff --git a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/LocationPollStrategy.java b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/LocationPollStrategy.java
index 89d7af92d12f..f1232c4b3906 100644
--- a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/LocationPollStrategy.java
+++ b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/LocationPollStrategy.java
@@ -8,6 +8,7 @@
 import com.azure.core.http.HttpResponse;
 import com.azure.core.implementation.RestProxy;
 import com.azure.core.implementation.SwaggerMethodParser;
+import com.azure.core.util.logging.ClientLogger;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Mono;
 
@@ -21,6 +22,8 @@
  * operation.
  */
 public final class LocationPollStrategy extends PollStrategy {
+    private final ClientLogger logger = new ClientLogger(LocationPollStrategy.class);
+
     LocationPollStrategyData data;
 
     /**
@@ -91,7 +94,7 @@ public Mono<HttpResponse> updateFromAsync(HttpResponse httpPollResponse) {
                             try {
                                 data.locationUrl = new URL(newLocationUrl);
                             } catch (MalformedURLException mfue) {
-                                throw Exceptions.propagate(mfue);
+                                throw logger.logExceptionAsError(Exceptions.propagate(mfue));
                             }
                         }
                     } else {
diff --git a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/ProvisioningStatePollStrategy.java b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/ProvisioningStatePollStrategy.java
index affef85b7a56..6c07ee8d3dd8 100644
--- a/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/ProvisioningStatePollStrategy.java
+++ b/sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/ProvisioningStatePollStrategy.java
@@ -10,6 +10,7 @@
 import com.azure.core.implementation.SwaggerMethodParser;
 import com.azure.core.management.CloudException;
 import com.azure.core.management.OperationState;
+import com.azure.core.util.logging.ClientLogger;
 import reactor.core.publisher.Mono;
 
 import java.io.IOException;
@@ -20,7 +21,7 @@
  * state property is in a completed state.
  */
 public final class ProvisioningStatePollStrategy extends PollStrategy {
-
+    private final ClientLogger logger = new ClientLogger(ProvisioningStatePollStrategy.class);
 
     private ProvisioningStatePollStrategyData data;
     ProvisioningStatePollStrategy(ProvisioningStatePollStrategyData data) {
@@ -87,9 +88,9 @@ Mono<HttpResponse> updateFromAsync(HttpResponse pollResponse) {
                         }
 
                         if (resource == null || resource.properties() == null || resource.properties().provisioningState() == null) {
-                            throw new CloudException("The polling response does not contain a valid body", bufferedHttpPollResponse, null);
+                            throw logger.logExceptionAsError(new CloudException("The polling response does not contain a valid body", bufferedHttpPollResponse, null));
                         } else if (OperationState.isFailedOrCanceled(resource.properties().provisioningState())) {
-                            throw new CloudException("Async operation failed with provisioning state: " + resource.properties().provisioningState(), bufferedHttpPollResponse);
+                            throw logger.logExceptionAsError(new CloudException("Async operation failed with provisioning state: " + resource.properties().provisioningState(), bufferedHttpPollResponse));
                         } else {
                             setStatus(resource.properties().provisioningState());
                         }
diff --git a/sdk/core/azure-core-test/src/main/java/com/azure/core/test/policy/RecordNetworkCallPolicy.java b/sdk/core/azure-core-test/src/main/java/com/azure/core/test/policy/RecordNetworkCallPolicy.java
index df9d27cabe25..7c0643f0ff8a 100644
--- a/sdk/core/azure-core-test/src/main/java/com/azure/core/test/policy/RecordNetworkCallPolicy.java
+++ b/sdk/core/azure-core-test/src/main/java/com/azure/core/test/policy/RecordNetworkCallPolicy.java
@@ -10,9 +10,8 @@
 import com.azure.core.http.policy.HttpPipelinePolicy;
 import com.azure.core.test.models.NetworkCallRecord;
 import com.azure.core.test.models.RecordedData;
+import com.azure.core.util.logging.ClientLogger;
 import io.netty.handler.codec.http.HttpResponseStatus;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Mono;
 
@@ -31,7 +30,8 @@
  */
 public class RecordNetworkCallPolicy implements HttpPipelinePolicy {
     private static final int DEFAULT_BUFFER_LENGTH = 1024;
-    private final Logger logger = LoggerFactory.getLogger(RecordNetworkCallPolicy.class);
+
+    private final ClientLogger logger = new ClientLogger(RecordNetworkCallPolicy.class);
     private final RecordedData recordedData;
 
     /**
@@ -73,9 +73,7 @@ public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineN
                 // Remove pre-added header if this is a waiting or redirection
                 if (body != null && body.contains("<Status>InProgress</Status>")
                     || Integer.parseInt(responseData.get("StatusCode")) == HttpResponseStatus.TEMPORARY_REDIRECT.code()) {
-                    if (logger.isInfoEnabled()) {
-                        logger.info("Waiting for a response or redirection.");
-                    }
+                    logger.info("Waiting for a response or redirection.");
                 } else {
                     recordedData.addNetworkCall(networkCallRecord);
                 }
@@ -129,7 +127,7 @@ private Mono<Map<String, String>> extractResponseData(final HttpResponse respons
 
                     content = new String(output.toByteArray(), StandardCharsets.UTF_8);
                 } catch (IOException e) {
-                    throw Exceptions.propagate(e);
+                    throw logger.logExceptionAsError(Exceptions.propagate(e));
                 }
 
                 responseData.remove("content-encoding");
diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/credentials/BasicAuthenticationCredential.java b/sdk/core/azure-core/src/main/java/com/azure/core/credentials/BasicAuthenticationCredential.java
index e390c24486e9..6a1426039eee 100644
--- a/sdk/core/azure-core/src/main/java/com/azure/core/credentials/BasicAuthenticationCredential.java
+++ b/sdk/core/azure-core/src/main/java/com/azure/core/credentials/BasicAuthenticationCredential.java
@@ -4,6 +4,7 @@
 package com.azure.core.credentials;
 
 import com.azure.core.implementation.util.Base64Util;
+import com.azure.core.util.logging.ClientLogger;
 import reactor.core.publisher.Mono;
 
 import java.io.UnsupportedEncodingException;
@@ -13,6 +14,7 @@
  * Basic Auth credentials for use with a REST Service Client.
  */
 public class BasicAuthenticationCredential implements TokenCredential {
+    private final ClientLogger logger = new ClientLogger(BasicAuthenticationCredential.class);
     /**
      * Basic auth user name.
      */
@@ -45,7 +47,7 @@ public Mono<AccessToken> getToken(String... scopes) {
             encodedCredential = Base64Util.encodeToString(credential.getBytes("UTF8"));
         } catch (UnsupportedEncodingException e) {
             // The encoding is hard-coded, so if it's unsupported, it needs to be fixed right here.
-            throw new RuntimeException(e);
+            throw logger.logExceptionAsError(new RuntimeException(e));
         }
 
         return Mono.just(new AccessToken(encodedCredential, OffsetDateTime.MAX));
diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/http/policy/CookiePolicy.java b/sdk/core/azure-core/src/main/java/com/azure/core/http/policy/CookiePolicy.java
index a9e84bb6ad44..89bac5c9cbfc 100644
--- a/sdk/core/azure-core/src/main/java/com/azure/core/http/policy/CookiePolicy.java
+++ b/sdk/core/azure-core/src/main/java/com/azure/core/http/policy/CookiePolicy.java
@@ -7,6 +7,7 @@
 import com.azure.core.http.HttpPipelineCallContext;
 import com.azure.core.http.HttpPipelineNextPolicy;
 import com.azure.core.http.HttpResponse;
+import com.azure.core.util.logging.ClientLogger;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Mono;
 
@@ -25,6 +26,7 @@
  * The Pipeline policy that which stores cookies based on the response Set-Cookie header and adds cookies to requests.
  */
 public class CookiePolicy implements HttpPipelinePolicy {
+    private final ClientLogger logger = new ClientLogger(CookiePolicy.class);
     private final CookieHandler cookies = new CookieManager();
 
     @Override
@@ -51,7 +53,7 @@ public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineN
                 try {
                     cookies.put(uri, responseHeaders);
                 } catch (IOException e) {
-                    throw Exceptions.propagate(e);
+                    throw logger.logExceptionAsError(Exceptions.propagate(e));
                 }
                 return httpResponse;
             });
diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/RestProxy.java b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/RestProxy.java
index 248f5a300cd9..6ad09c85285c 100644
--- a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/RestProxy.java
+++ b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/RestProxy.java
@@ -35,6 +35,7 @@
 import com.azure.core.implementation.util.ImplUtils;
 import com.azure.core.implementation.util.TypeUtil;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import io.netty.buffer.ByteBuf;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
@@ -66,6 +67,7 @@
  * deserialized Java object.
  */
 public class RestProxy implements InvocationHandler {
+    private final ClientLogger logger = new ClientLogger(RestProxy.class);
     private final HttpPipeline httpPipeline;
     private final SerializerAdapter serializer;
     private final SwaggerInterfaceParser interfaceParser;
@@ -148,7 +150,7 @@ public Object invoke(Object proxy, final Method method, Object[] args) {
             }
 
         } catch (Exception e) {
-            throw Exceptions.propagate(e);
+            throw logger.logExceptionAsError(Exceptions.propagate(e));
         }
     }
 
@@ -426,7 +428,7 @@ private Response<?> createResponse(HttpDecodedResponse response, Type entityType
             cls = (Class<? extends Response<?>>) (Object) PagedResponseBase.class;
 
             if (bodyAsObject != null && !TypeUtil.isTypeOrSubTypeOf(bodyAsObject.getClass(), Page.class)) {
-                throw new RuntimeException("Unable to create PagedResponse<T>. Body must be of a type that implements: " + Page.class);
+                throw logger.logExceptionAsError(new RuntimeException("Unable to create PagedResponse<T>. Body must be of a type that implements: " + Page.class));
             }
         }
 
@@ -443,7 +445,7 @@ private Response<?> createResponse(HttpDecodedResponse response, Type entityType
             .collect(Collectors.toList());
 
         if (constructors.isEmpty()) {
-            throw new RuntimeException("Cannot find suitable constructor for class " + cls);
+            throw logger.logExceptionAsError(new RuntimeException("Cannot find suitable constructor for class " + cls));
         }
 
         // try to create an instance using our list of potential candidates
@@ -461,14 +463,14 @@ private Response<?> createResponse(HttpDecodedResponse response, Type entityType
                     case 5:
                         return ctor.newInstance(httpRequest, responseStatusCode, responseHeaders, bodyAsObject, response.decodedHeaders().block());
                     default:
-                        throw new IllegalStateException("Response constructor with expected parameters not found.");
+                        throw logger.logExceptionAsError(new IllegalStateException("Response constructor with expected parameters not found."));
                 }
             } catch (IllegalAccessException | InvocationTargetException | InstantiationException e) {
-                throw reactor.core.Exceptions.propagate(e);
+                throw logger.logExceptionAsError(reactor.core.Exceptions.propagate(e));
             }
         }
         // error
-        throw new RuntimeException("Cannot find suitable constructor for class " + cls);
+        throw logger.logExceptionAsError(new RuntimeException("Cannot find suitable constructor for class " + cls));
     }
 
     protected final Mono<?> handleBodyReturnType(final HttpDecodedResponse response, final SwaggerMethodParser methodParser, final Type entityType) {
diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/UrlBuilder.java b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/UrlBuilder.java
index fb9d7e56ccd4..f33717511cad 100644
--- a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/UrlBuilder.java
+++ b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/UrlBuilder.java
@@ -3,6 +3,8 @@
 
 package com.azure.core.implementation.http;
 
+import com.azure.core.util.logging.ClientLogger;
+
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.LinkedHashMap;
@@ -12,6 +14,8 @@
  * A builder class that is used to create URLs.
  */
 public final class UrlBuilder {
+    private final ClientLogger logger = new ClientLogger(UrlBuilder.class);
+
     private String scheme;
     private String host;
     private String port;
@@ -190,7 +194,7 @@ private UrlBuilder with(String text, UrlTokenizerState startState) {
                             if (nameValue.length == 2) {
                                 setQueryParameter(nameValue[0], nameValue[1]);
                             } else {
-                                throw new IllegalArgumentException("Malformed query entry: " + entry);
+                                throw logger.logExceptionAsError(new IllegalArgumentException("Malformed query entry: " + entry));
                             }
                         }
                     }
diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/serializer/jackson/JacksonAdapter.java b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/serializer/jackson/JacksonAdapter.java
index abfe37c5caf6..e0e07489c717 100644
--- a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/serializer/jackson/JacksonAdapter.java
+++ b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/serializer/jackson/JacksonAdapter.java
@@ -7,6 +7,7 @@
 import com.azure.core.implementation.serializer.MalformedValueException;
 import com.azure.core.implementation.serializer.SerializerAdapter;
 import com.azure.core.implementation.serializer.SerializerEncoding;
+import com.azure.core.util.logging.ClientLogger;
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.core.JsonParseException;
@@ -29,6 +30,8 @@
  * Implementation of {@link SerializerAdapter} for Jackson.
  */
 public class JacksonAdapter implements SerializerAdapter {
+    private final ClientLogger logger = new ClientLogger(JacksonAdapter.class);
+
     /**
      * An instance of {@link ObjectMapper} to serialize/deserialize objects.
      */
@@ -156,7 +159,7 @@ public <T> T deserialize(String value, final Type type, SerializerEncoding encod
                 return (T) serializer().readValue(value, javaType);
             }
         } catch (JsonParseException jpe) {
-            throw new MalformedValueException(jpe.getMessage(), jpe);
+            throw logger.logExceptionAsError(new MalformedValueException(jpe.getMessage(), jpe));
         }
     }
 
diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/util/Context.java b/sdk/core/azure-core/src/main/java/com/azure/core/util/Context.java
index bb58ed298737..2945cf00f5d4 100644
--- a/sdk/core/azure-core/src/main/java/com/azure/core/util/Context.java
+++ b/sdk/core/azure-core/src/main/java/com/azure/core/util/Context.java
@@ -5,6 +5,8 @@
 
 import com.azure.core.implementation.annotation.Immutable;
 import com.azure.core.implementation.util.ImplUtils;
+import com.azure.core.util.logging.ClientLogger;
+
 import java.util.Map;
 import java.util.Optional;
 
@@ -16,6 +18,8 @@
  */
 @Immutable
 public class Context {
+    private final ClientLogger logger = new ClientLogger(Context.class);
+
     // All fields must be immutable.
     //
     /**
@@ -60,7 +64,7 @@ private Context(Context parent, Object key, Object value) {
      */
     public Context addData(Object key, Object value) {
         if (key == null) {
-            throw new IllegalArgumentException("key cannot be null");
+            throw logger.logExceptionAsError(new IllegalArgumentException("key cannot be null"));
         }
         return new Context(this, key, value);
     }
@@ -99,7 +103,7 @@ public static Context of(Map<Object, Object> keyValues) {
      */
     public Optional<Object> getData(Object key) {
         if (key == null) {
-            throw new IllegalArgumentException("key cannot be null");
+            throw logger.logExceptionAsError(new IllegalArgumentException("key cannot be null"));
         }
         for (Context c = this; c != null; c = c.parent) {
             if (key.equals(c.key)) {
diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/util/polling/Poller.java b/sdk/core/azure-core/src/main/java/com/azure/core/util/polling/Poller.java
index a95935a135ac..210f1982128d 100644
--- a/sdk/core/azure-core/src/main/java/com/azure/core/util/polling/Poller.java
+++ b/sdk/core/azure-core/src/main/java/com/azure/core/util/polling/Poller.java
@@ -168,7 +168,7 @@ public Poller(Duration pollInterval, Function<PollResponse<T>, Mono<PollResponse
      */
     public void cancelOperation() throws UnsupportedOperationException {
         if (this.cancelOperation == null) {
-            throw new UnsupportedOperationException("Cancel operation is not supported on this service/resource.");
+            throw logger.logExceptionAsError(new UnsupportedOperationException("Cancel operation is not supported on this service/resource."));
         }
 
         // We can not cancel an operation if it was never started
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubAsyncClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubAsyncClient.java
index a3a6c04861ec..99a22b209835 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubAsyncClient.java
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubAsyncClient.java
@@ -248,10 +248,10 @@ public EventHubAsyncConsumer createConsumer(String consumerGroup, String partiti
         Objects.requireNonNull(partitionId);
 
         if (ImplUtils.isNullOrEmpty(consumerGroup)) {
-            throw new IllegalArgumentException("'consumerGroup' cannot be an empty string.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("'consumerGroup' cannot be an empty string."));
         }
         if (ImplUtils.isNullOrEmpty(partitionId)) {
-            throw new IllegalArgumentException("'partitionId' cannot be an empty string.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("'partitionId' cannot be an empty string."));
         }
 
         final EventHubConsumerOptions clonedOptions = options.clone();
@@ -293,8 +293,8 @@ public void close() {
                     connection.close();
                 }
             } catch (IOException exception) {
-                throw new AmqpException(false, "Unable to close connection to service", exception,
-                    new ErrorContext(connectionOptions.host()));
+                throw logger.logExceptionAsError(new AmqpException(false, "Unable to close connection to service", exception,
+                    new ErrorContext(connectionOptions.host())));
             }
         }
     }
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubAsyncProducer.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubAsyncProducer.java
index 466899165ff4..31ff44b93dee 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubAsyncProducer.java
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubAsyncProducer.java
@@ -324,13 +324,13 @@ private void verifyPartitionKey(String partitionKey) {
         }
 
         if (isPartitionSender) {
-            throw new IllegalArgumentException(String.format(Locale.US,
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format(Locale.US,
                 "BatchOptions.partitionKey() cannot be set when an EventHubProducer is created with"
                     + "EventHubProducerOptions.partitionId() set. This EventHubProducer can only send events to partition '%s'.",
-                senderOptions.partitionId()));
+                senderOptions.partitionId())));
         } else if (partitionKey.length() > MAX_PARTITION_KEY_LENGTH) {
-            throw new IllegalArgumentException(String.format(Locale.US,
-                "PartitionKey '%s' exceeds the maximum allowed length: '%s'.", partitionKey, MAX_PARTITION_KEY_LENGTH));
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format(Locale.US,
+                "PartitionKey '%s' exceeds the maximum allowed length: '%s'.", partitionKey, MAX_PARTITION_KEY_LENGTH)));
         }
     }
 
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java
index cdd80c591111..31201630c112 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java
@@ -12,6 +12,7 @@
 import com.azure.core.util.configuration.BaseConfigurations;
 import com.azure.core.util.configuration.Configuration;
 import com.azure.core.util.configuration.ConfigurationManager;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.messaging.eventhubs.implementation.CBSAuthorizationType;
 import com.azure.messaging.eventhubs.implementation.ClientConstants;
 import com.azure.messaging.eventhubs.implementation.ConnectionOptions;
@@ -63,6 +64,7 @@
  */
 @ServiceClientBuilder(serviceClients = {EventHubAsyncClient.class, EventHubClient.class, EventProcessor.class})
 public class EventHubClientBuilder {
+    private final ClientLogger logger = new ClientLogger(EventHubClientBuilder.class);
 
     private static final String AZURE_EVENT_HUBS_CONNECTION_STRING = "AZURE_EVENT_HUBS_CONNECTION_STRING";
     private static final RetryOptions DEFAULT_RETRY = new RetryOptions()
@@ -118,7 +120,7 @@ public EventHubClientBuilder connectionString(String connectionString) {
             tokenCredential = new EventHubSharedAccessKeyCredential(properties.sharedAccessKeyName(),
                 properties.sharedAccessKey(), ClientConstants.TOKEN_VALIDITY);
         } catch (InvalidKeyException | NoSuchAlgorithmException e) {
-            throw new AzureException("Could not create the EventHubSharedAccessKeyCredential.", e);
+            throw logger.logExceptionAsError(new AzureException("Could not create the EventHubSharedAccessKeyCredential.", e));
         }
 
         return credential(properties.endpoint().getHost(), properties.eventHubName(), tokenCredential);
@@ -140,7 +142,7 @@ public EventHubClientBuilder connectionString(String connectionString) {
      */
     public EventHubClientBuilder connectionString(String connectionString, String eventHubName) {
         if (ImplUtils.isNullOrEmpty(eventHubName)) {
-            throw new IllegalArgumentException("'eventHubName' cannot be null or empty");
+            throw logger.logExceptionAsError(new IllegalArgumentException("'eventHubName' cannot be null or empty"));
         }
 
         final ConnectionStringProperties properties = new ConnectionStringProperties(connectionString);
@@ -149,14 +151,14 @@ public EventHubClientBuilder connectionString(String connectionString, String ev
             tokenCredential = new EventHubSharedAccessKeyCredential(properties.sharedAccessKeyName(),
                 properties.sharedAccessKey(), ClientConstants.TOKEN_VALIDITY);
         } catch (InvalidKeyException | NoSuchAlgorithmException e) {
-            throw new AzureException("Could not create the EventHubSharedAccessKeyCredential.", e);
+            throw logger.logExceptionAsError(new AzureException("Could not create the EventHubSharedAccessKeyCredential.", e));
         }
 
         if (!ImplUtils.isNullOrEmpty(properties.eventHubName())) {
-            throw new IllegalArgumentException(String.format(Locale.US,
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format(Locale.US,
                 "'connectionString' contains an Event Hub name [%s].  Please use the"
                     + " credentials(String connectionString) overload. Or supply a 'connectionString' without"
-                    + " 'EntityPath' in it.", properties.eventHubName()));
+                    + " 'EntityPath' in it.", properties.eventHubName())));
         }
 
         return credential(properties.endpoint().getHost(), eventHubName, tokenCredential);
@@ -190,10 +192,10 @@ public EventHubClientBuilder configuration(Configuration configuration) {
      */
     public EventHubClientBuilder credential(String host, String eventHubName, TokenCredential credential) {
         if (ImplUtils.isNullOrEmpty(host)) {
-            throw new IllegalArgumentException("'host' cannot be null or empty");
+            throw logger.logExceptionAsError(new IllegalArgumentException("'host' cannot be null or empty"));
         }
         if (ImplUtils.isNullOrEmpty(eventHubName)) {
-            throw new IllegalArgumentException("'eventHubName' cannot be null or empty.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("'eventHubName' cannot be null or empty."));
         }
 
         Objects.requireNonNull(credential);
@@ -407,9 +409,9 @@ private ConnectionOptions getConnectionOptions() {
             final String connectionString = configuration.get(AZURE_EVENT_HUBS_CONNECTION_STRING);
 
             if (ImplUtils.isNullOrEmpty(connectionString)) {
-                throw new IllegalArgumentException("Credentials have not been set using 'EventHubClientBuilder.credentials(String)'"
+                throw logger.logExceptionAsError(new IllegalArgumentException("Credentials have not been set using 'EventHubClientBuilder.credentials(String)'"
                     + "EventHubClientBuilder.credentials(String, String, TokenCredential). And the connection string is"
-                    + "not set in the '" + AZURE_EVENT_HUBS_CONNECTION_STRING + "' environment variable.");
+                    + "not set in the '" + AZURE_EVENT_HUBS_CONNECTION_STRING + "' environment variable."));
             }
 
             connectionString(connectionString);
@@ -423,7 +425,7 @@ private ConnectionOptions getConnectionOptions() {
         // is not AMQP_WEB_SOCKETS.
         if (proxyConfiguration != null && proxyConfiguration.isProxyAddressConfigured()
             && transport != TransportType.AMQP_WEB_SOCKETS) {
-            throw new IllegalArgumentException("Cannot use a proxy when TransportType is not AMQP.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Cannot use a proxy when TransportType is not AMQP."));
         }
 
         if (proxyConfiguration == null) {
@@ -456,7 +458,7 @@ private ProxyConfiguration getDefaultProxyConfiguration(Configuration configurat
 
         final String[] hostPort = proxyAddress.split(":");
         if (hostPort.length < 2) {
-            throw new IllegalArgumentException("HTTP_PROXY cannot be parsed into a proxy");
+            throw logger.logExceptionAsError(new IllegalArgumentException("HTTP_PROXY cannot be parsed into a proxy"));
         }
 
         final String host = hostPort[0];
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubSharedAccessKeyCredential.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubSharedAccessKeyCredential.java
index ae23b11a6eb3..d63483c780cf 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubSharedAccessKeyCredential.java
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubSharedAccessKeyCredential.java
@@ -7,6 +7,7 @@
 import com.azure.core.credentials.TokenCredential;
 import com.azure.core.implementation.annotation.Immutable;
 import com.azure.core.implementation.util.ImplUtils;
+import com.azure.core.util.logging.ClientLogger;
 import reactor.core.publisher.Mono;
 
 import javax.crypto.Mac;
@@ -33,6 +34,8 @@ public class EventHubSharedAccessKeyCredential implements TokenCredential {
     private static final String SHARED_ACCESS_SIGNATURE_FORMAT = "SharedAccessSignature sr=%s&sig=%s&se=%s&skn=%s";
     private static final String HASH_ALGORITHM = "HMACSHA256";
 
+    private final ClientLogger logger = new ClientLogger(EventHubSharedAccessKeyCredential.class);
+
     private final String keyName;
     private final Mac hmac;
     private final Duration tokenValidity;
@@ -87,7 +90,7 @@ public EventHubSharedAccessKeyCredential(String keyName, String sharedAccessKey,
     @Override
     public Mono<AccessToken> getToken(String... scopes) {
         if (scopes.length != 1) {
-            throw new IllegalArgumentException("'scopes' should only contain a single argument that is the token audience or resource name.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("'scopes' should only contain a single argument that is the token audience or resource name."));
         }
 
         return Mono.fromCallable(() -> generateSharedAccessSignature(scopes[0]));
@@ -95,7 +98,7 @@ public Mono<AccessToken> getToken(String... scopes) {
 
     private AccessToken generateSharedAccessSignature(final String resource) throws UnsupportedEncodingException {
         if (ImplUtils.isNullOrEmpty(resource)) {
-            throw new IllegalArgumentException("resource cannot be empty");
+            throw logger.logExceptionAsError(new IllegalArgumentException("resource cannot be empty"));
         }
 
         final String utf8Encoding = UTF_8.name();
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ManagementChannel.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ManagementChannel.java
index 44e3fc8a87de..60341d96247c 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ManagementChannel.java
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ManagementChannel.java
@@ -132,12 +132,12 @@ private <T> Mono<T> getProperties(Map<String, Object> properties, Function<Map<?
 
             return channelMono.flatMap(x -> x.sendWithAck(request, provider.getReactorDispatcher())).map(message -> {
                 if (!(message.getBody() instanceof AmqpValue)) {
-                    throw new IllegalArgumentException("Expected message.getBody() to be AmqpValue, but is: " + message.getBody());
+                    throw logger.logExceptionAsError(new IllegalArgumentException("Expected message.getBody() to be AmqpValue, but is: " + message.getBody()));
                 }
 
                 AmqpValue body = (AmqpValue) message.getBody();
                 if (!(body.getValue() instanceof Map)) {
-                    throw new IllegalArgumentException("Expected message.getBody().getValue() to be of type Map");
+                    throw logger.logExceptionAsError(new IllegalArgumentException("Expected message.getBody().getValue() to be of type Map"));
                 }
 
                 Map<?, ?> map = (Map<?, ?>) body.getValue();
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ReactorDispatcher.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ReactorDispatcher.java
index b11077be4410..8dc68be81578 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ReactorDispatcher.java
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ReactorDispatcher.java
@@ -83,13 +83,13 @@ private void throwIfSchedulerError() {
         final RejectedExecutionException rejectedException = this.reactor.attachments()
             .get(RejectedExecutionException.class, RejectedExecutionException.class);
         if (rejectedException != null) {
-            throw new RejectedExecutionException(rejectedException.getMessage(), rejectedException);
+            throw logger.logExceptionAsError(new RejectedExecutionException(rejectedException.getMessage(), rejectedException));
         }
 
         // throw when the pipe is in closed state - in which case,
         // signalling the new event-dispatch will fail
         if (!this.ioSignal.sink().isOpen()) {
-            throw new RejectedExecutionException("ReactorDispatcher instance is closed.");
+            throw logger.logExceptionAsError(new RejectedExecutionException("ReactorDispatcher instance is closed."));
         }
     }
 
@@ -118,7 +118,7 @@ public void run(Selectable selectable) {
                 logger.info("WorkScheduler.run() failed with an error: %s", ignorePipeClosedDuringReactorShutdown);
             } catch (IOException ioException) {
                 logger.error("WorkScheduler.run() failed with an error: %s", ioException);
-                throw new RuntimeException(ioException);
+                throw logger.logExceptionAsError(new RuntimeException(ioException));
             }
 
             Work topWork;
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ReactorProvider.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ReactorProvider.java
index 82657ca04de0..8c7c731247ed 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ReactorProvider.java
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ReactorProvider.java
@@ -3,6 +3,7 @@
 
 package com.azure.messaging.eventhubs.implementation;
 
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.messaging.eventhubs.implementation.handler.CustomIOHandler;
 import com.azure.messaging.eventhubs.implementation.handler.ReactorHandler;
 import org.apache.qpid.proton.Proton;
@@ -15,6 +16,7 @@
 import java.util.Objects;
 
 public class ReactorProvider {
+    private final ClientLogger logger = new ClientLogger(ReactorProvider.class);
     private final Object lock = new Object();
     private Reactor reactor;
     private ReactorDispatcher reactorDispatcher;
@@ -57,7 +59,7 @@ private Reactor createReactor(final int maxFrameSize, final Handler globalHandle
         Objects.requireNonNull(globalHandler);
 
         if (maxFrameSize <= 0) {
-            throw new IllegalArgumentException("'maxFrameSize' must be a positive number.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("'maxFrameSize' must be a positive number."));
         }
 
         final ReactorOptions reactorOptions = new ReactorOptions();
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/RequestResponseChannel.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/RequestResponseChannel.java
index 4f3c5e0f70f5..56b47b4c49a2 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/RequestResponseChannel.java
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/RequestResponseChannel.java
@@ -104,13 +104,13 @@ Mono<Message> sendWithAck(final Message message, final ReactorDispatcher dispatc
         start();
 
         if (message == null) {
-            throw new IllegalArgumentException("message cannot be null");
+            throw logger.logExceptionAsError(new IllegalArgumentException("message cannot be null"));
         }
         if (message.getMessageId() != null) {
-            throw new IllegalArgumentException("message.getMessageId() should be null");
+            throw logger.logExceptionAsError(new IllegalArgumentException("message.getMessageId() should be null"));
         }
         if (message.getReplyTo() != null) {
-            throw new IllegalArgumentException("message.getReplyTo() should be null");
+            throw logger.logExceptionAsError(new IllegalArgumentException("message.getReplyTo() should be null"));
         }
 
         final UnsignedLong messageId = UnsignedLong.valueOf(requestId.incrementAndGet());
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/TokenResourceProvider.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/TokenResourceProvider.java
index 47a03bd1d7ff..33519b5e2962 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/TokenResourceProvider.java
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/TokenResourceProvider.java
@@ -3,6 +3,8 @@
 
 package com.azure.messaging.eventhubs.implementation;
 
+import com.azure.core.util.logging.ClientLogger;
+
 import java.util.Locale;
 import java.util.Objects;
 
@@ -13,6 +15,7 @@ class TokenResourceProvider {
     private static final String TOKEN_AUDIENCE_FORMAT = "amqp://%s/%s";
     private static final String AZURE_ACTIVE_DIRECTORY_SCOPE = "https://eventhubs.azure.net//.default";
 
+    private final ClientLogger logger = new ClientLogger(TokenResourceProvider.class);
     private final CBSAuthorizationType authorizationType;
     private final String host;
 
@@ -31,8 +34,8 @@ String getResourceString(String resource) {
             case SHARED_ACCESS_SIGNATURE:
                 return String.format(Locale.US, TOKEN_AUDIENCE_FORMAT, host, resource);
             default:
-                throw new IllegalArgumentException(String.format(Locale.US,
-                    "'%s' is not supported authorization type for token audience.", authorizationType));
+                throw logger.logExceptionAsError(new IllegalArgumentException(String.format(Locale.US,
+                    "'%s' is not supported authorization type for token audience.", authorizationType)));
         }
     }
 }
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/handler/ConnectionHandler.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/handler/ConnectionHandler.java
index db80eb963fab..02326c954050 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/handler/ConnectionHandler.java
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/handler/ConnectionHandler.java
@@ -263,7 +263,7 @@ private void notifyErrorContext(Connection connection, ErrorCondition condition)
         }
 
         if (condition == null) {
-            throw new IllegalStateException("notifyErrorContext does not have an ErrorCondition.");
+            throw logger.logExceptionAsError(new IllegalStateException("notifyErrorContext does not have an ErrorCondition."));
         }
 
         // if the remote-peer abruptly closes the connection without issuing close frame issue one
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventHubConsumerOptions.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventHubConsumerOptions.java
index 3db1c4c07d0c..a44e45aae0f2 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventHubConsumerOptions.java
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventHubConsumerOptions.java
@@ -6,6 +6,7 @@
 import com.azure.core.amqp.RetryOptions;
 import com.azure.core.implementation.annotation.Fluent;
 import com.azure.core.implementation.util.ImplUtils;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.messaging.eventhubs.EventHubAsyncClient;
 import com.azure.messaging.eventhubs.EventHubAsyncConsumer;
 import reactor.core.scheduler.Scheduler;
@@ -22,6 +23,8 @@
  */
 @Fluent
 public class EventHubConsumerOptions implements Cloneable {
+    private final ClientLogger logger = new ClientLogger(EventHubConsumerOptions.class);
+
     /**
      * The maximum length, in characters, for the identifier assigned to an {@link EventHubAsyncConsumer}.
      */
@@ -61,8 +64,8 @@ public EventHubConsumerOptions() {
      */
     public EventHubConsumerOptions identifier(String identifier) {
         if (!ImplUtils.isNullOrEmpty(identifier) && identifier.length() > MAXIMUM_IDENTIFIER_LENGTH) {
-            throw new IllegalArgumentException(String.format(Locale.US,
-                "identifier length cannot exceed %s", MAXIMUM_IDENTIFIER_LENGTH));
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format(Locale.US,
+                "identifier length cannot exceed %s", MAXIMUM_IDENTIFIER_LENGTH)));
         }
 
         this.identifier = identifier;
@@ -89,7 +92,7 @@ public EventHubConsumerOptions identifier(String identifier) {
      */
     public EventHubConsumerOptions ownerLevel(Long priority) {
         if (priority != null && priority < 0) {
-            throw new IllegalArgumentException("'priority' cannot be a negative value. Please specify a zero or positive long value.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("'priority' cannot be a negative value. Please specify a zero or positive long value."));
         }
 
         this.ownerLevel = priority;
@@ -119,13 +122,13 @@ public EventHubConsumerOptions retry(RetryOptions retry) {
      */
     public EventHubConsumerOptions prefetchCount(int prefetchCount) {
         if (prefetchCount < MINIMUM_PREFETCH_COUNT) {
-            throw new IllegalArgumentException(String.format(Locale.US,
-                "PrefetchCount, '%s' has to be above %s", prefetchCount, MINIMUM_PREFETCH_COUNT));
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format(Locale.US,
+                "PrefetchCount, '%s' has to be above %s", prefetchCount, MINIMUM_PREFETCH_COUNT)));
         }
 
         if (prefetchCount > MAXIMUM_PREFETCH_COUNT) {
-            throw new IllegalArgumentException(String.format(Locale.US,
-                "PrefetchCount, '%s', has to be below %s", prefetchCount, MAXIMUM_PREFETCH_COUNT));
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format(Locale.US,
+                "PrefetchCount, '%s', has to be below %s", prefetchCount, MAXIMUM_PREFETCH_COUNT)));
         }
 
         this.prefetchCount = prefetchCount;
diff --git a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/KeyClientBuilder.java b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/KeyClientBuilder.java
index 3bd9f4bb3fe7..2ab46793e7dc 100644
--- a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/KeyClientBuilder.java
+++ b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/KeyClientBuilder.java
@@ -17,6 +17,7 @@
 import com.azure.core.http.policy.UserAgentPolicy;
 import com.azure.core.util.configuration.ConfigurationManager;
 import com.azure.core.implementation.annotation.ServiceClientBuilder;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.security.keyvault.keys.implementation.AzureKeyVaultConfiguration;
 
 import java.net.MalformedURLException;
@@ -55,6 +56,8 @@
  */
 @ServiceClientBuilder(serviceClients = KeyClient.class)
 public final class KeyClientBuilder {
+    private final ClientLogger logger = new ClientLogger(KeyClientBuilder.class);
+
     private final List<HttpPipelinePolicy> policies;
     private TokenCredential credential;
     private HttpPipeline pipeline;
@@ -110,7 +113,7 @@ public KeyAsyncClient buildAsyncClient() {
         URL buildEndpoint = getBuildEndpoint(buildConfiguration);
 
         if (buildEndpoint == null) {
-            throw new IllegalStateException(KeyVaultErrorCodeStrings.getErrorString(KeyVaultErrorCodeStrings.VAULT_END_POINT_REQUIRED));
+            throw logger.logExceptionAsError(new IllegalStateException(KeyVaultErrorCodeStrings.getErrorString(KeyVaultErrorCodeStrings.VAULT_END_POINT_REQUIRED)));
         }
 
         if (pipeline != null) {
@@ -118,7 +121,7 @@ public KeyAsyncClient buildAsyncClient() {
         }
 
         if (credential == null) {
-            throw new IllegalStateException(KeyVaultErrorCodeStrings.getErrorString(KeyVaultErrorCodeStrings.CREDENTIAL_REQUIRED));
+            throw logger.logExceptionAsError(new IllegalStateException(KeyVaultErrorCodeStrings.getErrorString(KeyVaultErrorCodeStrings.CREDENTIAL_REQUIRED)));
         }
 
         // Closest to API goes first, closest to wire goes last.
@@ -150,7 +153,7 @@ public KeyClientBuilder endpoint(String endpoint) {
         try {
             this.endpoint = new URL(endpoint);
         } catch (MalformedURLException e) {
-            throw new IllegalArgumentException("The Azure Key Vault endpoint url is malformed.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("The Azure Key Vault endpoint url is malformed."));
         }
         return this;
     }
diff --git a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/CryptographyAsyncClient.java b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/CryptographyAsyncClient.java
index 053fb631ade8..ddbec14ef3c1 100644
--- a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/CryptographyAsyncClient.java
+++ b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/CryptographyAsyncClient.java
@@ -113,7 +113,7 @@ private void initializeCryptoClients() {
                 localKeyCryptographyClient = new SymmetricKeyCryptographyClient(key, cryptographyServiceClient);
                 break;
             default:
-                throw new IllegalArgumentException(String.format("The Json Web Key Type: %s  is not supported.", key.kty().toString()));
+                throw logger.logExceptionAsError(new IllegalArgumentException(String.format("The Json Web Key Type: %s  is not supported.", key.kty().toString())));
         }
     }
 
@@ -524,7 +524,7 @@ Mono<VerifyResult> verifyData(SignatureAlgorithm algorithm, byte[] data, byte[]
 
     private void unpackAndValidateId(String keyId) {
         if (ImplUtils.isNullOrEmpty(keyId)) {
-            throw new IllegalArgumentException("Key Id is invalid");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Key Id is invalid"));
         }
         try {
             URL url = new URL(keyId);
@@ -533,14 +533,14 @@ private void unpackAndValidateId(String keyId) {
             String keyName = (tokens.length >= 3 ? tokens[2] : null);
             String version = (tokens.length >= 4 ? tokens[3] : null);
             if (Strings.isNullOrEmpty(endpoint)) {
-                throw new IllegalArgumentException("Key endpoint in key id is invalid");
+                throw logger.logExceptionAsError(new IllegalArgumentException("Key endpoint in key id is invalid"));
             } else if (Strings.isNullOrEmpty(keyName)) {
-                throw new IllegalArgumentException("Key name in key id is invalid");
+                throw logger.logExceptionAsError(new IllegalArgumentException("Key name in key id is invalid"));
             } else if (Strings.isNullOrEmpty(version)) {
-                throw new IllegalArgumentException("Key version in key id is invalid");
+                throw logger.logExceptionAsError(new IllegalArgumentException("Key version in key id is invalid"));
             }
         } catch (MalformedURLException e) {
-            throw new IllegalArgumentException("The key identifier is malformed", e);
+            throw logger.logExceptionAsError(new IllegalArgumentException("The key identifier is malformed", e));
         }
     }
 
diff --git a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/CryptographyClientBuilder.java b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/CryptographyClientBuilder.java
index 6d9b40d25e73..e212ba95cce1 100644
--- a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/CryptographyClientBuilder.java
+++ b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/CryptographyClientBuilder.java
@@ -16,6 +16,7 @@
 import com.azure.core.implementation.http.policy.spi.HttpPolicyProviders;
 import com.azure.core.util.configuration.Configuration;
 import com.azure.core.util.configuration.ConfigurationManager;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.security.keyvault.keys.KeyVaultCredentialPolicy;
 import com.azure.security.keyvault.keys.implementation.AzureKeyVaultConfiguration;
 import com.azure.security.keyvault.keys.models.webkey.JsonWebKey;
@@ -55,6 +56,8 @@
 @ServiceClientBuilder(serviceClients = CryptographyClient.class)
 public final class CryptographyClientBuilder {
     private final List<HttpPipelinePolicy> policies;
+    private final ClientLogger logger = new ClientLogger(CryptographyClientBuilder.class);
+
     private TokenCredential credential;
     private HttpPipeline pipeline;
     private JsonWebKey jsonWebKey;
@@ -110,7 +113,7 @@ public CryptographyAsyncClient buildAsyncClient() {
         Configuration buildConfiguration = (configuration == null) ? ConfigurationManager.getConfiguration().clone() : configuration;
 
         if (jsonWebKey == null && Strings.isNullOrEmpty(keyId)) {
-            throw new IllegalStateException("Json Web Key or jsonWebKey identifier are required to create cryptography client");
+            throw logger.logExceptionAsError(new IllegalStateException("Json Web Key or jsonWebKey identifier are required to create cryptography client"));
         }
 
         if (pipeline != null) {
@@ -122,7 +125,7 @@ public CryptographyAsyncClient buildAsyncClient() {
         }
 
         if (credential == null) {
-            throw new IllegalStateException("Key Vault credentials " + "are required to build the Cryptography async client");
+            throw logger.logExceptionAsError(new IllegalStateException("Key Vault credentials " + "are required to build the Cryptography async client"));
         }
 
         // Closest to API goes first, closest to wire goes last.
diff --git a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/EcKeyCryptographyClient.java b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/EcKeyCryptographyClient.java
index d93d63725596..257d168a9c64 100644
--- a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/EcKeyCryptographyClient.java
+++ b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/EcKeyCryptographyClient.java
@@ -4,6 +4,7 @@
 package com.azure.security.keyvault.keys.cryptography;
 
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.security.keyvault.keys.cryptography.models.DecryptResult;
 import com.azure.security.keyvault.keys.cryptography.models.EncryptionAlgorithm;
 import com.azure.security.keyvault.keys.cryptography.models.EncryptResult;
@@ -23,6 +24,8 @@
 import java.security.MessageDigest;
 
 class EcKeyCryptographyClient extends LocalKeyCryptographyClient {
+    private final ClientLogger logger = new ClientLogger(EcKeyCryptographyClient.class);
+
     private KeyPair keyPair;
     private final CryptographyServiceClient serviceClient;
     private final Provider provider;
@@ -54,13 +57,12 @@ private KeyPair getKeyPair(JsonWebKey key) {
 
     @Override
     Mono<EncryptResult> encryptAsync(EncryptionAlgorithm algorithm, byte[] plaintext, byte[] iv, byte[] authenticationData, Context context, JsonWebKey key) {
-        throw new UnsupportedOperationException("Encrypt operation is not supported for EC key");
+        throw logger.logExceptionAsError(new UnsupportedOperationException("Encrypt operation is not supported for EC key"));
     }
 
     @Override
     Mono<DecryptResult> decryptAsync(EncryptionAlgorithm algorithm, byte[] cipherText, byte[] iv, byte[] authenticationData, byte[] authenticationTag, Context context, JsonWebKey key) {
-
-        throw new UnsupportedOperationException("Decrypt operation is not supported for EC key");
+        throw logger.logExceptionAsError(new UnsupportedOperationException("Decrypt operation is not supported for EC key"));
     }
 
     @Override
@@ -144,14 +146,12 @@ Mono<VerifyResult> verifyAsync(SignatureAlgorithm algorithm, byte[] digest, byte
 
     @Override
     Mono<KeyWrapResult> wrapKeyAsync(KeyWrapAlgorithm algorithm, byte[] key, Context context, JsonWebKey webKey) {
-
         return Mono.error(new UnsupportedOperationException("Wrap key operation is not supported for EC key"));
     }
 
     @Override
     Mono<KeyUnwrapResult> unwrapKeyAsync(KeyWrapAlgorithm algorithm, byte[] encryptedKey, Context context, JsonWebKey key) {
-
-        throw new UnsupportedOperationException("Unwrap key operation is not supported for Ec key");
+        throw logger.logExceptionAsError(new UnsupportedOperationException("Unwrap key operation is not supported for Ec key"));
     }
 
     @Override
diff --git a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/Ecdsa.java b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/Ecdsa.java
index bd4ed3c0029c..6d091d1b4ad0 100644
--- a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/Ecdsa.java
+++ b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/Ecdsa.java
@@ -3,12 +3,15 @@
 
 package com.azure.security.keyvault.keys.cryptography;
 
+import com.azure.core.util.logging.ClientLogger;
+
 import java.security.GeneralSecurityException;
 import java.security.KeyPair;
 import java.security.Provider;
 import java.security.Signature;
 
 abstract class Ecdsa extends AsymmetricSignatureAlgorithm {
+    private final ClientLogger logger = new ClientLogger(Ecdsa.class);
 
     protected Ecdsa() {
         super("NONEwithECDSA");
@@ -54,7 +57,7 @@ public boolean verify(byte[] digest, byte[] signature) throws GeneralSecurityExc
 
         private void checkDigestLength(byte[] digest) {
             if (digest.length != getDigestLength()) {
-                throw new IllegalArgumentException("Invalid digest length.");
+                throw logger.logExceptionAsError(new IllegalArgumentException("Invalid digest length."));
             }
         }
 
diff --git a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/SymmetricKeyCryptographyClient.java b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/SymmetricKeyCryptographyClient.java
index 1597cced6e73..0121a4b3e761 100644
--- a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/SymmetricKeyCryptographyClient.java
+++ b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/cryptography/SymmetricKeyCryptographyClient.java
@@ -4,6 +4,7 @@
 package com.azure.security.keyvault.keys.cryptography;
 
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.security.keyvault.keys.cryptography.models.DecryptResult;
 import com.azure.security.keyvault.keys.cryptography.models.EncryptionAlgorithm;
 import com.azure.security.keyvault.keys.cryptography.models.EncryptResult;
@@ -19,6 +20,8 @@
 import java.security.NoSuchAlgorithmException;
 
 class SymmetricKeyCryptographyClient extends LocalKeyCryptographyClient {
+    private final ClientLogger logger = new ClientLogger(SymmetricKeyCryptographyClient.class);
+
     private byte[] key;
 
     /*
@@ -127,7 +130,7 @@ Mono<KeyWrapResult> wrapKeyAsync(KeyWrapAlgorithm algorithm, byte[] key, Context
         this.key = getKey(jsonWebKey);
 
         if (key == null || key.length == 0) {
-            throw new IllegalArgumentException("key");
+            throw logger.logExceptionAsError(new IllegalArgumentException("key"));
         }
 
         // Interpret the algorithm
diff --git a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/models/webkey/JsonWebKey.java b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/models/webkey/JsonWebKey.java
index d287d522a03e..c2305af1945d 100644
--- a/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/models/webkey/JsonWebKey.java
+++ b/sdk/keyvault/azure-keyvault-keys/src/main/java/com/azure/security/keyvault/keys/models/webkey/JsonWebKey.java
@@ -3,6 +3,7 @@
 
 package com.azure.security.keyvault.keys.models.webkey;
 
+import com.azure.core.util.logging.ClientLogger;
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -50,6 +51,7 @@
  */
 @JsonAutoDetect(getterVisibility = JsonAutoDetect.Visibility.PUBLIC_ONLY, setterVisibility = JsonAutoDetect.Visibility.PUBLIC_ONLY)
 public class JsonWebKey {
+    private final ClientLogger logger = new ClientLogger(JsonWebKey.class);
 
     /**
      * Key Identifier.
@@ -448,11 +450,11 @@ public String toString() {
         try {
             return mapper.writeValueAsString(this);
         } catch (JsonGenerationException e) {
-            throw new IllegalStateException(e);
+            throw logger.logExceptionAsError(new IllegalStateException(e));
         } catch (JsonMappingException e) {
-            throw new IllegalStateException(e);
+            throw logger.logExceptionAsError(new IllegalStateException(e));
         } catch (IOException e) {
-            throw new IllegalStateException(e);
+            throw logger.logExceptionAsError(new IllegalStateException(e));
         }
     }
 
@@ -559,7 +561,7 @@ private PublicKey getRSAPublicKey(Provider provider) {
 
             return factory.generatePublic(publicKeySpec);
         } catch (GeneralSecurityException e) {
-            throw new IllegalStateException(e);
+            throw logger.logExceptionAsError(new IllegalStateException(e));
         }
     }
 
@@ -578,7 +580,7 @@ private PrivateKey getRSAPrivateKey(Provider provider) {
 
             return factory.generatePrivate(privateKeySpec);
         } catch (GeneralSecurityException e) {
-            throw new IllegalStateException(e);
+            throw logger.logExceptionAsError(new IllegalStateException(e));
         }
     }
 
@@ -610,7 +612,7 @@ private static PrivateKey getECPrivateKey(byte[] d, ECParameterSpec curveSpec, P
      */
     private void checkRSACompatible() {
         if (!KeyType.RSA.equals(kty) && !KeyType.RSA_HSM.equals(kty)) {
-            throw new UnsupportedOperationException("Not an RSA key");
+            throw logger.logExceptionAsError(new UnsupportedOperationException("Not an RSA key"));
         }
     }
 
@@ -746,7 +748,7 @@ public KeyPair toEC(boolean includePrivateParameters, Provider provider) {
         }
 
         if (!KeyType.EC.equals(kty) && !KeyType.EC_HSM.equals(kty)) {
-            throw new IllegalArgumentException("Not an EC key.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Not an EC key."));
         }
 
         try {
@@ -773,7 +775,7 @@ public KeyPair toEC(boolean includePrivateParameters, Provider provider) {
 
             return realKeyPair;
         } catch (GeneralSecurityException e) {
-            throw new IllegalStateException(e);
+            throw logger.logExceptionAsError(new IllegalStateException(e));
         }
     }
 
diff --git a/sdk/keyvault/azure-keyvault-secrets/src/main/java/com/azure/security/keyvault/secrets/SecretClientBuilder.java b/sdk/keyvault/azure-keyvault-secrets/src/main/java/com/azure/security/keyvault/secrets/SecretClientBuilder.java
index b27677a89593..7f67fb75ba26 100644
--- a/sdk/keyvault/azure-keyvault-secrets/src/main/java/com/azure/security/keyvault/secrets/SecretClientBuilder.java
+++ b/sdk/keyvault/azure-keyvault-secrets/src/main/java/com/azure/security/keyvault/secrets/SecretClientBuilder.java
@@ -17,6 +17,7 @@
 import com.azure.core.http.policy.RetryPolicy;
 import com.azure.core.http.policy.UserAgentPolicy;
 import com.azure.core.implementation.util.ImplUtils;
+import com.azure.core.util.logging.ClientLogger;
 
 import java.net.MalformedURLException;
 import java.net.URL;
@@ -52,6 +53,8 @@
  */
 @ServiceClientBuilder(serviceClients = SecretClient.class)
 public final class SecretClientBuilder {
+    private final ClientLogger logger = new ClientLogger(SecretClientBuilder.class);
+
     private final List<HttpPipelinePolicy> policies;
     private TokenCredential credential;
     private HttpPipeline pipeline;
@@ -108,7 +111,7 @@ public SecretAsyncClient buildAsyncClient() {
         URL buildEndpoint = getBuildEndpoint(buildConfiguration);
 
         if (buildEndpoint == null) {
-            throw new IllegalStateException(KeyVaultErrorCodeStrings.getErrorString(KeyVaultErrorCodeStrings.VAULT_END_POINT_REQUIRED));
+            throw logger.logExceptionAsError(new IllegalStateException(KeyVaultErrorCodeStrings.getErrorString(KeyVaultErrorCodeStrings.VAULT_END_POINT_REQUIRED)));
         }
 
         if (pipeline != null) {
@@ -116,7 +119,7 @@ public SecretAsyncClient buildAsyncClient() {
         }
 
         if (credential == null) {
-            throw new IllegalStateException(KeyVaultErrorCodeStrings.getErrorString(KeyVaultErrorCodeStrings.CREDENTIAL_REQUIRED));
+            throw logger.logExceptionAsError(new IllegalStateException(KeyVaultErrorCodeStrings.getErrorString(KeyVaultErrorCodeStrings.CREDENTIAL_REQUIRED)));
         }
 
         // Closest to API goes first, closest to wire goes last.
@@ -148,7 +151,7 @@ public SecretClientBuilder endpoint(String endpoint) {
         try {
             this.endpoint = new URL(endpoint);
         } catch (MalformedURLException e) {
-            throw new IllegalArgumentException("The Azure Key Vault endpoint url is malformed.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("The Azure Key Vault endpoint url is malformed."));
         }
         return this;
     }
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobAsyncClient.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobAsyncClient.java
index dc4405fdd26f..e3890c79e1ae 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobAsyncClient.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobAsyncClient.java
@@ -10,6 +10,7 @@
 import com.azure.core.implementation.http.UrlBuilder;
 import com.azure.core.implementation.util.FluxUtil;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.implementation.AzureBlobStorageBuilder;
 import com.azure.storage.blob.implementation.AzureBlobStorageImpl;
 import com.azure.storage.blob.models.AccessTier;
@@ -82,6 +83,8 @@ public class BlobAsyncClient {
     private static final int BLOB_DEFAULT_DOWNLOAD_BLOCK_SIZE = 4 * Constants.MB;
     private static final int BLOB_MAX_DOWNLOAD_BLOCK_SIZE = 100 * Constants.MB;
 
+    private final ClientLogger logger = new ClientLogger(BlobAsyncClient.class);
+
     final AzureBlobStorageImpl azureBlobStorage;
     protected final String snapshot;
 
@@ -176,7 +179,7 @@ public URL getBlobUrl() {
             }
             return urlBuilder.toURL();
         } catch (MalformedURLException e) {
-            throw new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(), azureBlobStorage.getUrl()), e);
+            throw logger.logExceptionAsError(new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(), azureBlobStorage.getUrl()), e));
         }
     }
 
@@ -525,7 +528,7 @@ public Mono<Void> downloadToFile(String filePath, BlobRange range, Integer block
     Mono<Void> downloadToFile(String filePath, BlobRange range, Integer blockSize, ReliableDownloadOptions options,
                               BlobAccessConditions accessConditions, boolean rangeGetContentMD5, Context context) {
         if (blockSize < 0 || blockSize > BLOB_MAX_DOWNLOAD_BLOCK_SIZE) {
-            throw new IllegalArgumentException("Block size should not exceed 100MB");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Block size should not exceed 100MB"));
         }
 
         return Mono.using(() -> downloadToFileResourceSupplier(filePath),
@@ -542,7 +545,7 @@ private AsynchronousFileChannel downloadToFileResourceSupplier(String filePath)
         try {
             return AsynchronousFileChannel.open(Paths.get(filePath), StandardOpenOption.READ, StandardOpenOption.WRITE);
         } catch (IOException e) {
-            throw new UncheckedIOException(e);
+            throw logger.logExceptionAsError(new UncheckedIOException(e));
         }
     }
 
@@ -550,7 +553,7 @@ private void downloadToFileCleanup(AsynchronousFileChannel channel) {
         try {
             channel.close();
         } catch (IOException e) {
-            throw new UncheckedIOException(e);
+            throw logger.logExceptionAsError(new UncheckedIOException(e));
         }
     }
 
@@ -942,7 +945,7 @@ Mono<Response<String>> acquireLeaseWithResponse(String proposedId, int duration,
         if (!(duration == -1 || (duration >= 15 && duration <= 60))) {
             // Throwing is preferred to Mono.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new IllegalArgumentException("Duration must be -1 or between 15 and 60.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Duration must be -1 or between 15 and 60."));
         }
 
         return postProcessResponse(this.azureBlobStorage.blobs().acquireLeaseWithRestResponseAsync(
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobClient.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobClient.java
index 24ad2ee72ca8..55b504ef5364 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobClient.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobClient.java
@@ -7,6 +7,7 @@
 import com.azure.core.http.rest.SimpleResponse;
 import com.azure.core.http.rest.VoidResponse;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.models.AccessTier;
 import com.azure.storage.blob.models.BlobAccessConditions;
 import com.azure.storage.blob.models.BlobHTTPHeaders;
@@ -52,6 +53,8 @@
  * Docs</a> for more information.
  */
 public class BlobClient {
+    private final ClientLogger logger = new ClientLogger(BlobClient.class);
+
     private final BlobAsyncClient blobAsyncClient;
 
     /**
@@ -359,7 +362,7 @@ public VoidResponse downloadWithResponse(OutputStream stream, BlobRange range, R
                     try {
                         stream.write(bf.array());
                     } catch (IOException e) {
-                        throw new UncheckedIOException(e);
+                        throw logger.logExceptionAsError(new UncheckedIOException(e));
                     }
                 }).map(bf -> res))
             .last()
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobClientBuilder.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobClientBuilder.java
index 79e93dfac771..b157a3ec84fc 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobClientBuilder.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobClientBuilder.java
@@ -19,6 +19,7 @@
 import com.azure.core.implementation.util.ImplUtils;
 import com.azure.core.util.configuration.Configuration;
 import com.azure.core.util.configuration.ConfigurationManager;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.implementation.AzureBlobStorageBuilder;
 import com.azure.storage.blob.implementation.AzureBlobStorageImpl;
 import com.azure.storage.blob.models.LeaseAccessConditions;
@@ -75,6 +76,8 @@ public final class BlobClientBuilder {
     private static final String ENDPOINT_PROTOCOL = "defaultendpointsprotocol";
     private static final String ENDPOINT_SUFFIX = "endpointsuffix";
 
+    private final ClientLogger logger = new ClientLogger(BlobClientBuilder.class);
+
     private final List<HttpPipelinePolicy> additionalPolicies;
 
     private String endpoint;
@@ -260,7 +263,7 @@ public BlobClientBuilder endpoint(String endpoint) {
                 this.sharedKeyCredential = null;
             }
         } catch (MalformedURLException ex) {
-            throw new IllegalArgumentException("The Azure Storage Blob endpoint url is malformed.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("The Azure Storage Blob endpoint url is malformed."));
         }
 
         return this;
@@ -370,7 +373,7 @@ public BlobClientBuilder connectionString(String connectionString) {
         String endpointSuffix = connectionKVPs.get(ENDPOINT_SUFFIX);
 
         if (ImplUtils.isNullOrEmpty(accountName) || ImplUtils.isNullOrEmpty(accountKey)) {
-            throw new IllegalArgumentException("Connection string must contain 'AccountName' and 'AccountKey'.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Connection string must contain 'AccountName' and 'AccountKey'."));
         }
 
         if (!ImplUtils.isNullOrEmpty(endpointProtocol) && !ImplUtils.isNullOrEmpty(endpointSuffix)) {
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobInputStream.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobInputStream.java
index 1cfd6a66816d..486ac834e010 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobInputStream.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobInputStream.java
@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 package com.azure.storage.blob;
 
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.models.BlobAccessConditions;
 import com.azure.storage.blob.models.BlobRange;
 import com.azure.storage.blob.models.StorageException;
@@ -16,6 +17,8 @@
  * Provides an input stream to read a given blob resource.
  */
 public final class BlobInputStream extends InputStream {
+    private final ClientLogger logger = new ClientLogger(BlobInputStream.class);
+
     /**
      * Holds the reference to the blob this stream is associated with.
      */
@@ -323,7 +326,7 @@ public int read(final byte[] b) throws IOException {
     @Override
     public int read(final byte[] b, final int off, final int len) throws IOException {
         if (off < 0 || len < 0 || len > b.length - off) {
-            throw new IndexOutOfBoundsException();
+            throw logger.logExceptionAsError(new IndexOutOfBoundsException());
         }
 
         return this.readInternal(b, off, len);
@@ -418,7 +421,7 @@ public synchronized long skip(final long n) throws IOException {
         }
 
         if (n < 0 || this.currentAbsoluteReadPosition + n > this.streamLength + this.blobRangeOffset) {
-            throw new IndexOutOfBoundsException();
+            throw logger.logExceptionAsError(new IndexOutOfBoundsException());
         }
 
         this.reposition(this.currentAbsoluteReadPosition + n);
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobOutputStream.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobOutputStream.java
index fd2f17c2b019..94a8f6959c90 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobOutputStream.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobOutputStream.java
@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 package com.azure.storage.blob;
 
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.models.AppendBlobAccessConditions;
 import com.azure.storage.blob.models.AppendPositionAccessConditions;
 import com.azure.storage.blob.models.BlobAccessConditions;
@@ -32,6 +33,8 @@
 
 
 public final class BlobOutputStream extends OutputStream {
+    private final ClientLogger logger = new ClientLogger(BlobOutputStream.class);
+
     /**
      * Holds the {@link BlobAccessConditions} object that represents the access conditions for the blob.
      */
@@ -350,7 +353,7 @@ public void write(final byte[] data) throws IOException {
     @Override
     public void write(final byte[] data, final int offset, final int length) throws IOException {
         if (offset < 0 || length < 0 || length > data.length - offset) {
-            throw new IndexOutOfBoundsException();
+            throw logger.logExceptionAsError(new IndexOutOfBoundsException());
         }
 
         this.writeInternal(data, offset, length);
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobServiceAsyncClient.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobServiceAsyncClient.java
index 5d4aa782986f..dccade3750d7 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobServiceAsyncClient.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobServiceAsyncClient.java
@@ -10,6 +10,7 @@
 import com.azure.core.http.rest.VoidResponse;
 import com.azure.core.implementation.util.FluxUtil;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.implementation.AzureBlobStorageBuilder;
 import com.azure.storage.blob.implementation.AzureBlobStorageImpl;
 import com.azure.storage.blob.models.ContainerItem;
@@ -56,6 +57,8 @@
  * responses to a {@link java.util.concurrent.CompletableFuture} object through {@link Mono#toFuture()}.
  */
 public final class BlobServiceAsyncClient {
+    private final ClientLogger logger = new ClientLogger(BlobServiceAsyncClient.class);
+
     private final AzureBlobStorageImpl azureBlobStorage;
 
     /**
@@ -155,7 +158,7 @@ public URL getAccountUrl() {
         try {
             return new URL(azureBlobStorage.getUrl());
         } catch (MalformedURLException e) {
-            throw new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(), azureBlobStorage.getUrl()), e);
+            throw logger.logExceptionAsError(new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(), azureBlobStorage.getUrl()), e));
         }
     }
 
@@ -313,7 +316,7 @@ public Mono<Response<UserDelegationKey>> getUserDelegationKeyWithResponse(Offset
     Mono<Response<UserDelegationKey>> getUserDelegationKeyWithResponse(OffsetDateTime start, OffsetDateTime expiry, Context context) {
         Utility.assertNotNull("expiry", expiry);
         if (start != null && !start.isBefore(expiry)) {
-            throw new IllegalArgumentException("`start` must be null or a datetime before `expiry`.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("`start` must be null or a datetime before `expiry`."));
         }
 
         return postProcessResponse(
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobServiceClientBuilder.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobServiceClientBuilder.java
index 7115ba0e1017..e8955361a3b1 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobServiceClientBuilder.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobServiceClientBuilder.java
@@ -19,6 +19,7 @@
 import com.azure.core.implementation.util.ImplUtils;
 import com.azure.core.util.configuration.Configuration;
 import com.azure.core.util.configuration.ConfigurationManager;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.implementation.AzureBlobStorageBuilder;
 import com.azure.storage.common.credentials.SASTokenCredential;
 import com.azure.storage.common.credentials.SharedKeyCredential;
@@ -59,6 +60,8 @@ public final class BlobServiceClientBuilder {
     private static final String ENDPOINT_PROTOCOL = "defaultendpointsprotocol";
     private static final String ENDPOINT_SUFFIX = "endpointsuffix";
 
+    private final ClientLogger logger = new ClientLogger(BlobServiceClientBuilder.class);
+
     private final List<HttpPipelinePolicy> additionalPolicies;
 
     private String endpoint;
@@ -146,7 +149,7 @@ public BlobServiceClientBuilder endpoint(String endpoint) {
                 this.sharedKeyCredential = null;
             }
         } catch (MalformedURLException ex) {
-            throw new IllegalArgumentException("The Azure Storage endpoint url is malformed.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("The Azure Storage endpoint url is malformed."));
         }
 
         return this;
@@ -227,7 +230,7 @@ public BlobServiceClientBuilder connectionString(String connectionString) {
         String endpointSuffix = connectionKVPs.get(ENDPOINT_SUFFIX);
 
         if (ImplUtils.isNullOrEmpty(accountName) || ImplUtils.isNullOrEmpty(accountKey)) {
-            throw new IllegalArgumentException("Connection string must contain 'AccountName' and 'AccountKey'.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Connection string must contain 'AccountName' and 'AccountKey'."));
         }
 
         if (!ImplUtils.isNullOrEmpty(endpointProtocol) && !ImplUtils.isNullOrEmpty(endpointSuffix)) {
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlockBlobAsyncClient.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlockBlobAsyncClient.java
index 7d03f4e1578f..b3a59a588641 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlockBlobAsyncClient.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlockBlobAsyncClient.java
@@ -9,6 +9,7 @@
 import com.azure.core.http.rest.VoidResponse;
 import com.azure.core.implementation.util.FluxUtil;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.implementation.AzureBlobStorageImpl;
 import com.azure.storage.blob.models.BlobAccessConditions;
 import com.azure.storage.blob.models.BlobHTTPHeaders;
@@ -67,6 +68,8 @@
  * object through {@link Mono#toFuture()}.
  */
 public final class BlockBlobAsyncClient extends BlobAsyncClient {
+    private final ClientLogger logger = new ClientLogger(BlockBlobAsyncClient.class);
+
     static final int BLOB_DEFAULT_UPLOAD_BLOCK_SIZE = 4 * Constants.MB;
     static final int BLOB_MAX_UPLOAD_BLOCK_SIZE = 100 * Constants.MB;
 
@@ -180,7 +183,7 @@ public Mono<Void> uploadFromFile(String filePath) {
     public Mono<Void> uploadFromFile(String filePath, Integer blockSize, BlobHTTPHeaders headers, Metadata metadata,
                                      BlobAccessConditions accessConditions) {
         if (blockSize < 0 || blockSize > BLOB_MAX_UPLOAD_BLOCK_SIZE) {
-            throw new IllegalArgumentException("Block size should not exceed 100MB");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Block size should not exceed 100MB"));
         }
 
         return Mono.using(() -> uploadFileResourceSupplier(filePath),
@@ -198,7 +201,7 @@ public Mono<Void> uploadFromFile(String filePath, Integer blockSize, BlobHTTPHea
                         try {
                             channel.close();
                         } catch (IOException e) {
-                            throw new UncheckedIOException(e);
+                            throw logger.logExceptionAsError(new UncheckedIOException(e));
                         }
                     });
             }, this::uploadFileCleanup);
@@ -209,7 +212,7 @@ private AsynchronousFileChannel uploadFileResourceSupplier(String filePath) {
         try {
             return AsynchronousFileChannel.open(Paths.get(filePath), StandardOpenOption.READ);
         } catch (IOException e) {
-            throw new UncheckedIOException(e);
+            throw logger.logExceptionAsError(new UncheckedIOException(e));
         }
     }
 
@@ -217,7 +220,7 @@ private void uploadFileCleanup(AsynchronousFileChannel channel) {
         try {
             channel.close();
         } catch (IOException e) {
-            throw new UncheckedIOException(e);
+            throw logger.logExceptionAsError(new UncheckedIOException(e));
         }
     }
 
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ContainerAsyncClient.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ContainerAsyncClient.java
index 784791a20a20..3ee3cdddf812 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ContainerAsyncClient.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ContainerAsyncClient.java
@@ -9,6 +9,7 @@
 import com.azure.core.http.rest.VoidResponse;
 import com.azure.core.implementation.util.FluxUtil;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.implementation.AzureBlobStorageBuilder;
 import com.azure.storage.blob.implementation.AzureBlobStorageImpl;
 import com.azure.storage.blob.models.BlobFlatListSegment;
@@ -73,6 +74,7 @@ public final class ContainerAsyncClient {
 
     public static final String LOG_CONTAINER_NAME = "$logs";
 
+    private final ClientLogger logger = new ClientLogger(ContainerAsyncClient.class);
     private final AzureBlobStorageImpl azureBlobStorage;
 
     /**
@@ -238,7 +240,7 @@ public URL getContainerUrl() {
         try {
             return new URL(azureBlobStorage.getUrl());
         } catch (MalformedURLException e) {
-            throw new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(), azureBlobStorage.getUrl()), e);
+            throw logger.logExceptionAsError(new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(), azureBlobStorage.getUrl()), e));
         }
     }
 
@@ -337,7 +339,7 @@ Mono<VoidResponse> deleteWithResponse(ContainerAccessConditions accessConditions
         if (!validateNoEtag(accessConditions.modifiedAccessConditions())) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new UnsupportedOperationException("ETag access conditions are not supported for this API.");
+            throw logger.logExceptionAsError(new UnsupportedOperationException("ETag access conditions are not supported for this API."));
         }
 
         return postProcessResponse(this.azureBlobStorage.containers().deleteWithRestResponseAsync(null, null, null,
@@ -407,8 +409,8 @@ Mono<VoidResponse> setMetadataWithResponse(Metadata metadata, ContainerAccessCon
             || accessConditions.modifiedAccessConditions().ifUnmodifiedSince() != null) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new UnsupportedOperationException(
-                "If-Modified-Since is the only HTTP access condition supported for this API");
+            throw logger.logExceptionAsError(new UnsupportedOperationException(
+                "If-Modified-Since is the only HTTP access condition supported for this API"));
         }
 
         return postProcessResponse(this.azureBlobStorage.containers().setMetadataWithRestResponseAsync(null, null,
@@ -492,7 +494,7 @@ Mono<VoidResponse> setAccessPolicyWithResponse(PublicAccessType accessType, List
         if (!validateNoEtag(accessConditions.modifiedAccessConditions())) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new UnsupportedOperationException("ETag access conditions are not supported for this API.");
+            throw logger.logExceptionAsError(new UnsupportedOperationException("ETag access conditions are not supported for this API."));
         }
 
         /*
@@ -708,7 +710,7 @@ public Flux<BlobItem> listBlobsHierarchy(String delimiter, ListBlobsOptions opti
     private Mono<ContainersListBlobHierarchySegmentResponse> listBlobsHierarchySegment(String marker, String delimiter, ListBlobsOptions options) {
         options = options == null ? new ListBlobsOptions() : options;
         if (options.details().snapshots()) {
-            throw new UnsupportedOperationException("Including snapshots in a hierarchical listing is not supported.");
+            throw logger.logExceptionAsError(new UnsupportedOperationException("Including snapshots in a hierarchical listing is not supported."));
         }
 
         return postProcessResponse(this.azureBlobStorage.containers().listBlobHierarchySegmentWithRestResponseAsync(null, delimiter,
@@ -845,8 +847,8 @@ Mono<Response<String>> acquireLeaseWithResponse(String proposedID, int duration,
         if (!this.validateNoEtag(modifiedAccessConditions)) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new UnsupportedOperationException(
-                "ETag access conditions are not supported for this API.");
+            throw logger.logExceptionAsError(new UnsupportedOperationException(
+                "ETag access conditions are not supported for this API."));
         }
 
         return postProcessResponse(this.azureBlobStorage.containers().acquireLeaseWithRestResponseAsync(null, null, duration, proposedID,
@@ -882,8 +884,8 @@ Mono<Response<String>> renewLeaseWithResponse(String leaseID, ModifiedAccessCond
         if (!this.validateNoEtag(modifiedAccessConditions)) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new UnsupportedOperationException(
-                "ETag access conditions are not supported for this API.");
+            throw logger.logExceptionAsError(new UnsupportedOperationException(
+                "ETag access conditions are not supported for this API."));
         }
 
         return postProcessResponse(this.azureBlobStorage.containers().renewLeaseWithRestResponseAsync(null, leaseID, null, null,
@@ -919,8 +921,8 @@ Mono<VoidResponse> releaseLeaseWithResponse(String leaseID, ModifiedAccessCondit
         if (!this.validateNoEtag(modifiedAccessConditions)) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new UnsupportedOperationException(
-                "ETag access conditions are not supported for this API.");
+            throw logger.logExceptionAsError(new UnsupportedOperationException(
+                "ETag access conditions are not supported for this API."));
         }
 
         return postProcessResponse(this.azureBlobStorage.containers().releaseLeaseWithRestResponseAsync(
@@ -962,8 +964,8 @@ Mono<Response<Duration>> breakLeaseWithResponse(Integer breakPeriodInSeconds, Mo
         if (!this.validateNoEtag(modifiedAccessConditions)) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new UnsupportedOperationException(
-                "ETag access conditions are not supported for this API.");
+            throw logger.logExceptionAsError(new UnsupportedOperationException(
+                "ETag access conditions are not supported for this API."));
         }
 
         return postProcessResponse(this.azureBlobStorage.containers().breakLeaseWithRestResponseAsync(null,
@@ -1003,8 +1005,8 @@ Mono<Response<String>> changeLeaseWithResponse(String leaseId, String proposedID
         if (!this.validateNoEtag(modifiedAccessConditions)) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new UnsupportedOperationException(
-                "ETag access conditions are not supported for this API.");
+            throw logger.logExceptionAsError(new UnsupportedOperationException(
+                "ETag access conditions are not supported for this API."));
         }
 
         return postProcessResponse(this.azureBlobStorage.containers().changeLeaseWithRestResponseAsync(null,
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ContainerClientBuilder.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ContainerClientBuilder.java
index b6e9886bfec1..2b24a87edcfa 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ContainerClientBuilder.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ContainerClientBuilder.java
@@ -19,6 +19,7 @@
 import com.azure.core.implementation.util.ImplUtils;
 import com.azure.core.util.configuration.Configuration;
 import com.azure.core.util.configuration.ConfigurationManager;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.implementation.AzureBlobStorageBuilder;
 import com.azure.storage.common.credentials.SASTokenCredential;
 import com.azure.storage.common.credentials.SharedKeyCredential;
@@ -59,6 +60,8 @@ public final class ContainerClientBuilder {
     private static final String ENDPOINT_PROTOCOL = "defaultendpointsprotocol";
     private static final String ENDPOINT_SUFFIX = "endpointsuffix";
 
+    private final ClientLogger logger = new ClientLogger(ContainerClientBuilder.class);
+
     private final List<HttpPipelinePolicy> additionalPolicies;
 
     private String endpoint;
@@ -152,7 +155,7 @@ public ContainerClientBuilder endpoint(String endpoint) {
                 this.sharedKeyCredential = null;
             }
         } catch (MalformedURLException ex) {
-            throw new IllegalArgumentException("The Azure Storage Blob endpoint url is malformed.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("The Azure Storage Blob endpoint url is malformed."));
         }
 
         return this;
@@ -243,7 +246,7 @@ public ContainerClientBuilder connectionString(String connectionString) {
         String endpointSuffix = connectionKVPs.get(ENDPOINT_SUFFIX);
 
         if (ImplUtils.isNullOrEmpty(accountName) || ImplUtils.isNullOrEmpty(accountKey)) {
-            throw new IllegalArgumentException("Connection string must contain 'AccountName' and 'AccountKey'.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Connection string must contain 'AccountName' and 'AccountKey'."));
         }
 
         if (!ImplUtils.isNullOrEmpty(endpointProtocol) && !ImplUtils.isNullOrEmpty(endpointSuffix)) {
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/PageBlobAsyncClient.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/PageBlobAsyncClient.java
index 4155a2775c84..359444920254 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/PageBlobAsyncClient.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/PageBlobAsyncClient.java
@@ -8,6 +8,7 @@
 import com.azure.core.implementation.http.UrlBuilder;
 import com.azure.core.implementation.util.FluxUtil;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.implementation.AzureBlobStorageImpl;
 import com.azure.storage.blob.models.BlobAccessConditions;
 import com.azure.storage.blob.models.BlobHTTPHeaders;
@@ -66,6 +67,8 @@ public final class PageBlobAsyncClient extends BlobAsyncClient {
      */
     public static final int MAX_PUT_PAGES_BYTES = 4 * Constants.MB;
 
+    private final ClientLogger logger = new ClientLogger(PageBlobAsyncClient.class);
+
     /**
      * Package-private constructor for use by {@link BlobClientBuilder}.
      * @param azureBlobStorage the API client for blob storage
@@ -117,12 +120,12 @@ Mono<Response<PageBlobItem>> createWithResponse(long size, Long sequenceNumber,
         if (size % PAGE_BYTES != 0) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new IllegalArgumentException("size must be a multiple of PageBlobAsyncClient.PAGE_BYTES.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("size must be a multiple of PageBlobAsyncClient.PAGE_BYTES."));
         }
         if (sequenceNumber != null && sequenceNumber < 0) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new IllegalArgumentException("SequenceNumber must be greater than or equal to 0.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("SequenceNumber must be greater than or equal to 0."));
         }
         metadata = metadata == null ? new Metadata() : metadata;
 
@@ -183,7 +186,7 @@ Mono<Response<PageBlobItem>> uploadPagesWithResponse(PageRange pageRange, Flux<B
         if (pageRange == null) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new IllegalArgumentException("pageRange cannot be null.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("pageRange cannot be null."));
         }
         String pageRangeStr = pageRangeToString(pageRange);
 
@@ -252,7 +255,7 @@ Mono<Response<PageBlobItem>> uploadPagesFromURLWithResponse(PageRange range, URL
         if (range == null) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new IllegalArgumentException("range cannot be null.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("range cannot be null."));
         }
 
         String rangeString = pageRangeToString(range);
@@ -312,7 +315,7 @@ Mono<Response<PageBlobItem>> clearPagesWithResponse(PageRange pageRange,
         if (pageRange == null) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new IllegalArgumentException("pageRange cannot be null.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("pageRange cannot be null."));
         }
         String pageRangeStr = pageRangeToString(pageRange);
 
@@ -395,7 +398,7 @@ Mono<Response<PageList>> getPageRangesDiffWithResponse(BlobRange blobRange, Stri
         accessConditions = accessConditions == null ? new BlobAccessConditions() : accessConditions;
 
         if (prevSnapshot == null) {
-            throw new IllegalArgumentException("prevSnapshot cannot be null");
+            throw logger.logExceptionAsError(new IllegalArgumentException("prevSnapshot cannot be null"));
         }
 
         return postProcessResponse(this.azureBlobStorage.pageBlobs().getPageRangesDiffWithRestResponseAsync(
@@ -437,7 +440,7 @@ Mono<Response<PageBlobItem>> resizeWithResponse(long size, BlobAccessConditions
         if (size % PAGE_BYTES != 0) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new IllegalArgumentException("size must be a multiple of PageBlobAsyncClient.PAGE_BYTES.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("size must be a multiple of PageBlobAsyncClient.PAGE_BYTES."));
         }
         accessConditions = accessConditions == null ? new BlobAccessConditions() : accessConditions;
 
@@ -480,7 +483,7 @@ Mono<Response<PageBlobItem>> updateSequenceNumberWithResponse(SequenceNumberActi
         if (sequenceNumber != null && sequenceNumber < 0) {
             // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
             // subscription.
-            throw new IllegalArgumentException("SequenceNumber must be greater than or equal to 0.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("SequenceNumber must be greater than or equal to 0."));
         }
         accessConditions = accessConditions == null ? new BlobAccessConditions() : accessConditions;
         sequenceNumber = action == SequenceNumberActionType.INCREMENT ? null : sequenceNumber;
diff --git a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ServiceSASSignatureValues.java b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ServiceSASSignatureValues.java
index a62f85d78643..92dfe6c16203 100644
--- a/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ServiceSASSignatureValues.java
+++ b/sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/ServiceSASSignatureValues.java
@@ -3,6 +3,7 @@
 
 package com.azure.storage.blob;
 
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.blob.models.UserDelegationKey;
 import com.azure.storage.common.Constants;
 import com.azure.storage.common.IPRange;
@@ -32,6 +33,7 @@
  * for additional samples.</p>
  */
 final class ServiceSASSignatureValues {
+    private final ClientLogger logger = new ClientLogger(ServiceSASSignatureValues.class);
 
     private String version = Constants.HeaderConstants.TARGET_STORAGE_VERSION;
 
@@ -270,7 +272,7 @@ public ServiceSASSignatureValues canonicalName(String urlString, String accountN
         try {
             url = new URL(urlString);
         } catch (MalformedURLException e) {
-            throw new RuntimeException(e);
+            throw logger.logExceptionAsError(new RuntimeException(e));
         }
 
         this.canonicalName = String.format("/blob/%s%s", accountName, url.getPath());
@@ -467,7 +469,7 @@ private void assertGenerateOK(boolean usingUserDelegation) {
         }
 
         if (Constants.UrlConstants.SAS_CONTAINER_CONSTANT.equals(this.resource) && this.snapshotId != null) {
-            throw new IllegalArgumentException("Cannot set a snapshotId without resource being a blob.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Cannot set a snapshotId without resource being a blob."));
         }
     }
 
diff --git a/sdk/storage/azure-storage-common/pom.xml b/sdk/storage/azure-storage-common/pom.xml
index 7eb369b5174a..86182b01d6f6 100644
--- a/sdk/storage/azure-storage-common/pom.xml
+++ b/sdk/storage/azure-storage-common/pom.xml
@@ -36,7 +36,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core</artifactId>
-      <version>1.0.0-preview.3</version>
+      <version>1.0.0-preview.4</version>
     </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
diff --git a/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/policy/RequestRetryOptions.java b/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/policy/RequestRetryOptions.java
index e1835e114cb6..46ea76e8719e 100644
--- a/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/policy/RequestRetryOptions.java
+++ b/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/policy/RequestRetryOptions.java
@@ -3,6 +3,8 @@
 
 package com.azure.storage.common.policy;
 
+import com.azure.core.util.logging.ClientLogger;
+
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -11,6 +13,7 @@
  * functionality.
  */
 public final class RequestRetryOptions {
+    private final ClientLogger logger = new ClientLogger(RequestRetryOptions.class);
 
     private final int maxTries;
     private final int tryTimeout;
@@ -154,7 +157,7 @@ long calculateDelayInMs(int tryCount) {
                 delay = this.retryDelayInMs;
                 break;
             default:
-                throw new IllegalArgumentException("Invalid retry policy type.");
+                throw logger.logExceptionAsError(new IllegalArgumentException("Invalid retry policy type."));
         }
 
         return Math.min(delay, this.maxRetryDelayInMs);
diff --git a/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/policy/SASTokenCredentialPolicy.java b/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/policy/SASTokenCredentialPolicy.java
index 4af3edf7f229..00625019ca36 100644
--- a/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/policy/SASTokenCredentialPolicy.java
+++ b/sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/policy/SASTokenCredentialPolicy.java
@@ -8,6 +8,7 @@
 import com.azure.core.http.HttpResponse;
 import com.azure.core.http.policy.HttpPipelinePolicy;
 import com.azure.core.implementation.util.ImplUtils;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.common.credentials.SASTokenCredential;
 import reactor.core.publisher.Mono;
 
@@ -18,6 +19,8 @@
  * Policy that adds the SAS token to the request URL's query.
  */
 public final class SASTokenCredentialPolicy implements HttpPipelinePolicy {
+    private final ClientLogger logger = new ClientLogger(SASTokenCredentialPolicy.class);
+
     private final SASTokenCredential credential;
 
     /**
@@ -38,7 +41,7 @@ public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineN
             String newURL = requestURL.toString() + delimiter + credential.sasToken();
             context.httpRequest().url(new URL(newURL));
         } catch (MalformedURLException ex) {
-            throw new IllegalStateException(ex);
+            throw logger.logExceptionAsError(new IllegalStateException(ex));
         }
 
         return next.process();
diff --git a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/DirectoryAsyncClient.java b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/DirectoryAsyncClient.java
index cca1489a8e02..54004bc7cf3b 100644
--- a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/DirectoryAsyncClient.java
+++ b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/DirectoryAsyncClient.java
@@ -8,6 +8,7 @@
 import com.azure.core.http.rest.SimpleResponse;
 import com.azure.core.http.rest.VoidResponse;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.common.credentials.SASTokenCredential;
 import com.azure.storage.common.credentials.SharedKeyCredential;
 import com.azure.storage.file.implementation.AzureFileStorageBuilder;
@@ -53,6 +54,8 @@
  * @see SASTokenCredential
  */
 public class DirectoryAsyncClient {
+    private final ClientLogger logger = new ClientLogger(DirectoryAsyncClient.class);
+
     private final AzureFileStorageImpl azureFileStorageClient;
     private final String shareName;
     private final String directoryPath;
@@ -100,8 +103,8 @@ public URL getDirectoryUrl() {
         try {
             return new URL(azureFileStorageClient.getUrl());
         } catch (MalformedURLException e) {
-            throw new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(),
-                azureFileStorageClient.getUrl()), e);
+            throw logger.logExceptionAsError(new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(),
+                azureFileStorageClient.getUrl()), e));
         }
     }
 
diff --git a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/DirectoryClientBuilder.java b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/DirectoryClientBuilder.java
index 172ad46b6b79..2b320b9074ab 100644
--- a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/DirectoryClientBuilder.java
+++ b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/DirectoryClientBuilder.java
@@ -17,6 +17,7 @@
 import com.azure.core.implementation.http.policy.spi.HttpPolicyProviders;
 import com.azure.core.util.configuration.Configuration;
 import com.azure.core.util.configuration.ConfigurationManager;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.common.Utility;
 import com.azure.storage.common.credentials.SASTokenCredential;
 import com.azure.storage.common.credentials.SharedKeyCredential;
@@ -74,6 +75,8 @@
 @ServiceClientBuilder(serviceClients = {DirectoryClient.class, DirectoryAsyncClient.class})
 public class DirectoryClientBuilder {
     private static final String ACCOUNT_NAME = "accountname";
+
+    private final ClientLogger logger = new ClientLogger(DirectoryClientBuilder.class);
     private final List<HttpPipelinePolicy> policies;
     private final RetryPolicy retryPolicy;
 
@@ -123,7 +126,7 @@ public DirectoryAsyncClient buildAsyncClient() {
         }
 
         if (sasTokenCredential == null && sharedKeyCredential == null) {
-            throw new IllegalArgumentException("Credentials are required for authorization");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Credentials are required for authorization"));
         }
 
         // Closest to API goes first, closest to wire goes last.
@@ -202,7 +205,7 @@ public DirectoryClientBuilder endpoint(String endpoint) {
                 this.sharedKeyCredential = null;
             }
         } catch (MalformedURLException ex) {
-            throw new IllegalArgumentException("The Azure Storage Directory endpoint url is malformed.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("The Azure Storage Directory endpoint url is malformed."));
         }
 
         return this;
@@ -259,8 +262,8 @@ private void getEndPointFromConnectionString(String connectionString) {
         try {
             this.endpoint = new URL(String.format("https://%s.file.core.windows.net", accountName));
         } catch (MalformedURLException e) {
-            throw new IllegalArgumentException(String.format("There is no valid endpoint for the connection string. "
-                                                                 + "Connection String: %s", connectionString));
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format("There is no valid endpoint "
+                + "for the connection string. Connection String: %s", connectionString)));
         }
     }
 
diff --git a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileAsyncClient.java b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileAsyncClient.java
index 9305b5b0acaa..ef49189e112d 100644
--- a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileAsyncClient.java
+++ b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileAsyncClient.java
@@ -130,8 +130,8 @@ public URL getFileUrl() {
         try {
             return new URL(azureFileStorageClient.getUrl());
         } catch (MalformedURLException e) {
-            throw new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(),
-                azureFileStorageClient.getUrl()), e);
+            throw logger.logExceptionAsError(new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(),
+                azureFileStorageClient.getUrl()), e));
         }
     }
 
@@ -284,7 +284,7 @@ private AsynchronousFileChannel channelSetup(String filePath) {
         try {
             return AsynchronousFileChannel.open(Paths.get(filePath), StandardOpenOption.READ, StandardOpenOption.WRITE);
         } catch (IOException e) {
-            throw new UncheckedIOException(e);
+            throw logger.logExceptionAsError(new UncheckedIOException(e));
         }
     }
 
@@ -292,7 +292,7 @@ private void channelCleanUp(AsynchronousFileChannel channel) {
         try {
             channel.close();
         } catch (IOException e) {
-            throw new UncheckedIOException(e);
+            throw logger.logExceptionAsError(new UncheckedIOException(e));
         }
     }
 
diff --git a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileClientBuilder.java b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileClientBuilder.java
index 1f54c018c28f..3cc47d07b076 100644
--- a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileClientBuilder.java
+++ b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileClientBuilder.java
@@ -17,6 +17,7 @@
 import com.azure.core.implementation.http.policy.spi.HttpPolicyProviders;
 import com.azure.core.util.configuration.Configuration;
 import com.azure.core.util.configuration.ConfigurationManager;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.common.Utility;
 import com.azure.storage.common.credentials.SASTokenCredential;
 import com.azure.storage.common.credentials.SharedKeyCredential;
@@ -75,6 +76,8 @@
 @ServiceClientBuilder(serviceClients = {FileClient.class, FileAsyncClient.class})
 public class FileClientBuilder {
     private static final String ACCOUNT_NAME = "accountname";
+
+    private final ClientLogger logger = new ClientLogger(FileClientBuilder.class);
     private final List<HttpPipelinePolicy> policies;
     private final RetryPolicy retryPolicy;
 
@@ -124,7 +127,7 @@ public FileAsyncClient buildAsyncClient() {
         }
 
         if (sasTokenCredential == null && sharedKeyCredential == null) {
-            throw new IllegalArgumentException("Credentials are required for authorization");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Credentials are required for authorization"));
         }
 
         // Closest to API goes first, closest to wire goes last.
@@ -206,7 +209,7 @@ public FileClientBuilder endpoint(String endpoint) {
                 this.sharedKeyCredential = null;
             }
         } catch (MalformedURLException ex) {
-            throw new IllegalArgumentException("The Azure Storage File endpoint url is malformed.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("The Azure Storage File endpoint url is malformed."));
         }
 
         return this;
@@ -262,8 +265,8 @@ private void getEndPointFromConnectionString(String connectionString) {
         try {
             this.endpoint = new URL(String.format("https://%s.file.core.windows.net", accountName));
         } catch (MalformedURLException e) {
-            throw new IllegalArgumentException(String.format("There is no valid endpoint for the connection string. "
-                                                                + "Connection String: %s", connectionString));
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format("There is no valid endpoint for"
+                + " the connection string. Connection String: %s", connectionString)));
         }
     }
 
diff --git a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileServiceAsyncClient.java b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileServiceAsyncClient.java
index e545c18965a5..203daaf7f88d 100644
--- a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileServiceAsyncClient.java
+++ b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileServiceAsyncClient.java
@@ -9,6 +9,7 @@
 import com.azure.core.http.rest.VoidResponse;
 import com.azure.core.implementation.util.ImplUtils;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.common.credentials.SASTokenCredential;
 import com.azure.storage.common.credentials.SharedKeyCredential;
 import com.azure.storage.file.implementation.AzureFileStorageBuilder;
@@ -48,6 +49,7 @@
  * @see SASTokenCredential
  */
 public final class FileServiceAsyncClient {
+    private final ClientLogger logger = new ClientLogger(FileServiceAsyncClient.class);
     private final AzureFileStorageImpl azureFileStorageClient;
 
     /**
@@ -72,8 +74,8 @@ public URL getFileServiceUrl() {
         try {
             return new URL(azureFileStorageClient.getUrl());
         } catch (MalformedURLException e) {
-            throw new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(),
-                azureFileStorageClient.getUrl()), e);
+            throw logger.logExceptionAsError(new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(),
+                azureFileStorageClient.getUrl()), e));
         }
     }
 
diff --git a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileServiceClientBuilder.java b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileServiceClientBuilder.java
index 76522775ea28..e6676528be8d 100644
--- a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileServiceClientBuilder.java
+++ b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/FileServiceClientBuilder.java
@@ -17,6 +17,7 @@
 import com.azure.core.implementation.http.policy.spi.HttpPolicyProviders;
 import com.azure.core.util.configuration.Configuration;
 import com.azure.core.util.configuration.ConfigurationManager;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.common.Utility;
 import com.azure.storage.common.credentials.SASTokenCredential;
 import com.azure.storage.common.credentials.SharedKeyCredential;
@@ -73,6 +74,7 @@
  */
 @ServiceClientBuilder(serviceClients = {FileServiceClient.class, FileServiceAsyncClient.class})
 public final class FileServiceClientBuilder {
+    private final ClientLogger logger = new ClientLogger(FileServiceClientBuilder.class);
     private static final String ACCOUNT_NAME = "accountname";
     private final List<HttpPipelinePolicy> policies;
 
@@ -115,7 +117,7 @@ public FileServiceAsyncClient buildAsyncClient() {
         }
 
         if (sasTokenCredential == null && sharedKeyCredential == null) {
-            throw new IllegalArgumentException("Credentials are required for authorization");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Credentials are required for authorization"));
         }
 
         // Closest to API goes first, closest to wire goes last.
@@ -186,7 +188,7 @@ public FileServiceClientBuilder endpoint(String endpoint) {
                 this.sharedKeyCredential = null;
             }
         } catch (MalformedURLException ex) {
-            throw new IllegalArgumentException("The Azure Storage File Service endpoint url is malformed.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("The Azure Storage File Service endpoint url is malformed."));
         }
 
         return this;
@@ -243,8 +245,8 @@ private void getEndPointFromConnectionString(String connectionString) {
         try {
             this.endpoint = new URL(String.format("https://%s.file.core.windows.net", accountName));
         } catch (MalformedURLException e) {
-            throw new IllegalArgumentException(String.format("There is no valid endpoint for the connection string. "
-                                                                 + "Connection String: %s", connectionString));
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format("There is no valid endpoint for"
+                + " the connection string. Connection String: %s", connectionString)));
         }
     }
 
diff --git a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/ShareAsyncClient.java b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/ShareAsyncClient.java
index 7d7b059f55cf..4513506ac1e0 100644
--- a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/ShareAsyncClient.java
+++ b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/ShareAsyncClient.java
@@ -9,6 +9,7 @@
 import com.azure.core.http.rest.VoidResponse;
 import com.azure.core.implementation.DateTimeRfc1123;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.common.credentials.SASTokenCredential;
 import com.azure.storage.common.credentials.SharedKeyCredential;
 import com.azure.storage.file.implementation.AzureFileStorageBuilder;
@@ -50,6 +51,8 @@
  * @see SASTokenCredential
  */
 public class ShareAsyncClient {
+    private final ClientLogger logger = new ClientLogger(ShareAsyncClient.class);
+
     private final AzureFileStorageImpl azureFileStorageClient;
     private final String shareName;
     private final String snapshot;
@@ -95,8 +98,8 @@ public URL getShareUrl() {
         try {
             return new URL(azureFileStorageClient.getUrl());
         } catch (MalformedURLException e) {
-            throw new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(),
-                azureFileStorageClient.getUrl()), e);
+            throw logger.logExceptionAsError(new RuntimeException(String.format("Invalid URL on %s: %s" + getClass().getSimpleName(),
+                azureFileStorageClient.getUrl()), e));
         }
     }
 
diff --git a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/ShareClientBuilder.java b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/ShareClientBuilder.java
index 2df6451a0dcd..e7c68d979dc3 100644
--- a/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/ShareClientBuilder.java
+++ b/sdk/storage/azure-storage-file/src/main/java/com/azure/storage/file/ShareClientBuilder.java
@@ -17,6 +17,7 @@
 import com.azure.core.implementation.http.policy.spi.HttpPolicyProviders;
 import com.azure.core.util.configuration.Configuration;
 import com.azure.core.util.configuration.ConfigurationManager;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.storage.common.Utility;
 import com.azure.storage.common.credentials.SASTokenCredential;
 import com.azure.storage.common.credentials.SharedKeyCredential;
@@ -73,6 +74,7 @@
  */
 @ServiceClientBuilder(serviceClients = {ShareClient.class, ShareAsyncClient.class})
 public class ShareClientBuilder {
+    private final ClientLogger logger = new ClientLogger(ShareClientBuilder.class);
     private static final String ACCOUNT_NAME = "accountname";
     private final List<HttpPipelinePolicy> policies;
 
@@ -120,7 +122,7 @@ public ShareAsyncClient buildAsyncClient() {
         }
 
         if (sasTokenCredential == null && sharedKeyCredential == null) {
-            throw new IllegalArgumentException("Credentials are required for authorization");
+            throw logger.logExceptionAsError(new IllegalArgumentException("Credentials are required for authorization"));
         }
 
         // Closest to API goes first, closest to wire goes last.
@@ -194,7 +196,7 @@ public ShareClientBuilder endpoint(String endpoint) {
                 this.sharedKeyCredential = null;
             }
         } catch (MalformedURLException ex) {
-            throw new IllegalArgumentException("The Azure Storage File Service endpoint url is malformed.");
+            throw logger.logExceptionAsError(new IllegalArgumentException("The Azure Storage File Service endpoint url is malformed."));
         }
 
         return this;
@@ -251,8 +253,8 @@ private void getEndPointFromConnectionString(String connectionString) {
         try {
             this.endpoint = new URL(String.format("https://%s.file.core.windows.net", accountName));
         } catch (MalformedURLException e) {
-            throw new IllegalArgumentException(String.format("There is no valid endpoint for the connection string. "
-                                                                 + "Connection String: %s", connectionString));
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format("There is no valid endpoint for"
+                + " the connection string. Connection String: %s", connectionString)));
         }
     }
 
diff --git a/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java b/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java
index 86931937896f..413e4462e57a 100644
--- a/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java
+++ b/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueAsyncClient.java
@@ -51,7 +51,7 @@
  * @see SASTokenCredential
  */
 public final class QueueAsyncClient {
-    private static final ClientLogger LOGGER = new ClientLogger(QueueAsyncClient.class);
+    private final ClientLogger logger = new ClientLogger(QueueAsyncClient.class);
     private final AzureQueueStorageImpl client;
     private final String queueName;
 
@@ -97,8 +97,8 @@ public URL getQueueUrl() {
         try {
             return new URL(client.getUrl());
         } catch (MalformedURLException ex) {
-            LOGGER.error("Queue URL is malformed");
-            throw new RuntimeException("Queue URL is malformed");
+            logger.error("Queue URL is malformed");
+            throw logger.logExceptionAsError(new RuntimeException("Queue URL is malformed"));
         }
     }
 
diff --git a/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueClientBuilder.java b/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueClientBuilder.java
index 7d3a3473fd29..e85fce73080d 100644
--- a/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueClientBuilder.java
+++ b/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueClientBuilder.java
@@ -79,7 +79,7 @@
  */
 @ServiceClientBuilder(serviceClients = {QueueClient.class, QueueAsyncClient.class})
 public final class QueueClientBuilder {
-    private static final ClientLogger LOGGER = new ClientLogger(QueueClientBuilder.class);
+    private final ClientLogger logger = new ClientLogger(QueueClientBuilder.class);
     private static final String ACCOUNT_NAME = "accountname";
     private final List<HttpPipelinePolicy> policies;
 
@@ -144,8 +144,7 @@ public QueueAsyncClient buildAsyncClient() {
         Objects.requireNonNull(endpoint);
 
         if (sasTokenCredential == null && sharedKeyCredential == null && bearerTokenCredential == null) {
-            LOGGER.logExceptionAsError(new IllegalArgumentException("Credentials are required for authorization"));
-            return null;
+            throw logger.logExceptionAsError(new IllegalArgumentException("Credentials are required for authorization"));
         }
 
         if (pipeline != null) {
@@ -214,8 +213,7 @@ public QueueClientBuilder endpoint(String endpoint) {
                 this.bearerTokenCredential = null;
             }
         } catch (MalformedURLException ex) {
-            LOGGER.logExceptionAsError(new IllegalArgumentException("The Azure Storage Queue endpoint url is malformed. Endpoint: " + endpoint));
-            return null;
+            throw logger.logExceptionAsError(new IllegalArgumentException("The Azure Storage Queue endpoint url is malformed. Endpoint: " + endpoint));
         }
 
         return this;
@@ -299,7 +297,7 @@ private void getEndPointFromConnectionString(String connectionString) {
         try {
             this.endpoint = new URL(String.format("https://%s.queue.core.windows.net", accountName));
         } catch (MalformedURLException e) {
-            LOGGER.logExceptionAsError(new IllegalArgumentException(String.format("There is no valid account for the connection string. "
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format("There is no valid account for the connection string. "
                 + "Connection String: %s", connectionString)));
         }
     }
diff --git a/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueServiceAsyncClient.java b/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueServiceAsyncClient.java
index 5c5a78f2dc0c..7a4b57be1223 100644
--- a/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueServiceAsyncClient.java
+++ b/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueServiceAsyncClient.java
@@ -47,7 +47,7 @@
  * @see SASTokenCredential
  */
 public final class QueueServiceAsyncClient {
-    private static final ClientLogger LOGGER = new ClientLogger(QueueServiceAsyncClient.class);
+    private final ClientLogger logger = new ClientLogger(QueueServiceAsyncClient.class);
     private final AzureQueueStorageImpl client;
 
     /**
@@ -71,8 +71,8 @@ public URL getQueueServiceUrl() {
         try {
             return new URL(client.getUrl());
         } catch (MalformedURLException ex) {
-            LOGGER.error("Queue Service URL is malformed");
-            throw new RuntimeException("Storage account URL is malformed");
+            logger.error("Queue Service URL is malformed");
+            throw logger.logExceptionAsError(new RuntimeException("Storage account URL is malformed"));
         }
     }
 
diff --git a/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueServiceClientBuilder.java b/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueServiceClientBuilder.java
index 9905dbc42725..51e3a8334528 100644
--- a/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueServiceClientBuilder.java
+++ b/sdk/storage/azure-storage-queue/src/main/java/com/azure/storage/queue/QueueServiceClientBuilder.java
@@ -76,7 +76,7 @@
  */
 @ServiceClientBuilder(serviceClients = {QueueServiceClient.class, QueueServiceAsyncClient.class})
 public final class QueueServiceClientBuilder {
-    private static final ClientLogger LOGGER = new ClientLogger(QueueServiceClientBuilder.class);
+    private final ClientLogger logger = new ClientLogger(QueueServiceClientBuilder.class);
     private static final String ACCOUNT_NAME = "accountname";
     private final List<HttpPipelinePolicy> policies;
 
@@ -119,8 +119,7 @@ public QueueServiceAsyncClient buildAsyncClient() {
         Objects.requireNonNull(endpoint);
 
         if (sasTokenCredential == null && sharedKeyCredential == null && bearerTokenCredential == null) {
-            LOGGER.logExceptionAsError(new IllegalArgumentException("Credentials are required for authorization"));
-            return null;
+            logger.logExceptionAsError(new IllegalArgumentException("Credentials are required for authorization"));
         }
 
         if (pipeline != null) {
@@ -199,8 +198,7 @@ public QueueServiceClientBuilder endpoint(String endpoint) {
                 this.bearerTokenCredential = null;
             }
         } catch (MalformedURLException ex) {
-            LOGGER.logExceptionAsError(new IllegalArgumentException("The Azure Storage Queue endpoint url is malformed."));
-            return null;
+            logger.logExceptionAsError(new IllegalArgumentException("The Azure Storage Queue endpoint url is malformed."));
         }
 
         return this;
@@ -271,8 +269,8 @@ private void getEndPointFromConnectionString(String connectionString) {
         try {
             this.endpoint = new URL(String.format("https://%s.queue.core.windows.net", accountName));
         } catch (MalformedURLException e) {
-            LOGGER.logExceptionAsError(new IllegalArgumentException(String.format("There is no valid account for the connection string. "
-                + "Connection String: %s", connectionString)));
+            throw logger.logExceptionAsError(new IllegalArgumentException(String.format("There is no valid account for the "
+                + "connection string. Connection String: %s", connectionString)));
         }
     }