diff --git a/Microsoft.Azure.Cosmos/src/CosmosClientOptions.cs b/Microsoft.Azure.Cosmos/src/CosmosClientOptions.cs
index 09e87237ce..0e2d3110af 100644
--- a/Microsoft.Azure.Cosmos/src/CosmosClientOptions.cs
+++ b/Microsoft.Azure.Cosmos/src/CosmosClientOptions.cs
@@ -444,7 +444,14 @@ public System.Text.Json.JsonSerializerOptions UseSystemTextJsonSerializerWithOpt
         /// <remarks>
         /// <para>This is optimal for latency-sensitive workloads. Does not apply if <see cref="ConnectionMode.Gateway"/> is used.</para>
         /// </remarks>
-        internal bool? EnableAdvancedReplicaSelectionForTcp { get; set; }
+        internal bool? EnableAdvancedReplicaSelectionForTcp { get; set; }
+
+        /// <summary>
+        /// Gets or sets stack trace optimization to reduce stack trace proliferation in high-concurrency scenarios where exceptions are frequently thrown. 
+        /// When enabled, critical SDK components optimize exception handling to minimize performance overhead.
+        /// The default value is 'true'.
+        /// </summary>
+        internal bool EnableAsyncCacheExceptionNoSharing { get; set; } = true;
 
         /// <summary>
         /// (Direct/TCP) Controls the amount of idle time after which unused connections are closed.
diff --git a/Microsoft.Azure.Cosmos/src/DocumentClient.cs b/Microsoft.Azure.Cosmos/src/DocumentClient.cs
index c8950e97fa..6828ed56c1 100644
--- a/Microsoft.Azure.Cosmos/src/DocumentClient.cs
+++ b/Microsoft.Azure.Cosmos/src/DocumentClient.cs
@@ -118,6 +118,7 @@ internal partial class DocumentClient : IDisposable, IAuthorizationTokenProvider
 
         private readonly bool IsLocalQuorumConsistency = false;
         private readonly bool isReplicaAddressValidationEnabled;
+        private readonly bool enableAsyncCacheExceptionNoSharing;
 
         //Fault Injection
         private readonly IChaosInterceptorFactory chaosInterceptorFactory;
@@ -243,7 +244,9 @@ public DocumentClient(Uri serviceEndpoint,
             }
 
             this.Initialize(serviceEndpoint, connectionPolicy, desiredConsistencyLevel);
-            this.initTaskCache = new AsyncCacheNonBlocking<string, bool>(cancellationToken: this.cancellationTokenSource.Token);
+            this.initTaskCache = new AsyncCacheNonBlocking<string, bool>(
+                cancellationToken: this.cancellationTokenSource.Token,
+                enableAsyncCacheExceptionNoSharing: this.enableAsyncCacheExceptionNoSharing);
             this.isReplicaAddressValidationEnabled = ConfigurationManager.IsReplicaAddressValidationEnabled(connectionPolicy);
         }
 
@@ -444,6 +447,7 @@ internal DocumentClient(Uri serviceEndpoint,
         /// <param name="remoteCertificateValidationCallback">This delegate responsible for validating the third party certificate. </param>
         /// <param name="cosmosClientTelemetryOptions">This is distributed tracing flag</param>
         /// <param name="chaosInterceptorFactory">This is the chaos interceptor used for fault injection</param>
+        /// <param name="enableAsyncCacheExceptionNoSharing">A boolean flag indicating if stack trace optimization is enabled.</param>
         /// <remarks>
         /// The service endpoint can be obtained from the Azure Management Portal.
         /// If you are connecting using one of the Master Keys, these can be obtained along with the endpoint from the Azure Management Portal
@@ -472,7 +476,8 @@ internal DocumentClient(Uri serviceEndpoint,
                               string cosmosClientId = null,
                               RemoteCertificateValidationCallback remoteCertificateValidationCallback = null,
                               CosmosClientTelemetryOptions cosmosClientTelemetryOptions = null,
-                              IChaosInterceptorFactory chaosInterceptorFactory = null)
+                              IChaosInterceptorFactory chaosInterceptorFactory = null,
+                              bool enableAsyncCacheExceptionNoSharing = true)
         {
             if (sendingRequestEventArgs != null)
             {
@@ -491,10 +496,13 @@ internal DocumentClient(Uri serviceEndpoint,
                 this.receivedResponse += receivedResponseEventArgs;
             }
 
+            this.enableAsyncCacheExceptionNoSharing = enableAsyncCacheExceptionNoSharing;
             this.cosmosAuthorization = cosmosAuthorization ?? throw new ArgumentNullException(nameof(cosmosAuthorization));
             this.transportClientHandlerFactory = transportClientHandlerFactory;
             this.IsLocalQuorumConsistency = isLocalQuorumConsistency;
-            this.initTaskCache = new AsyncCacheNonBlocking<string, bool>(cancellationToken: this.cancellationTokenSource.Token);
+            this.initTaskCache = new AsyncCacheNonBlocking<string, bool>(
+                cancellationToken: this.cancellationTokenSource.Token,
+                enableAsyncCacheExceptionNoSharing: this.enableAsyncCacheExceptionNoSharing);
             this.chaosInterceptorFactory = chaosInterceptorFactory;
             this.chaosInterceptor = chaosInterceptorFactory?.CreateInterceptor(this);
 
@@ -675,8 +683,9 @@ private async Task OpenPrivateAsync(CancellationToken cancellationToken)
                     storeModel: this.GatewayStoreModel, 
                     tokenProvider: this, 
                     retryPolicy: this.retryPolicy,
-                    telemetryToServiceHelper: this.telemetryToServiceHelper);
-                this.partitionKeyRangeCache = new PartitionKeyRangeCache(this, this.GatewayStoreModel, this.collectionCache, this.GlobalEndpointManager);
+                    telemetryToServiceHelper: this.telemetryToServiceHelper,
+                    enableAsyncCacheExceptionNoSharing: this.enableAsyncCacheExceptionNoSharing);
+                this.partitionKeyRangeCache = new PartitionKeyRangeCache(this, this.GatewayStoreModel, this.collectionCache, this.GlobalEndpointManager, this.enableAsyncCacheExceptionNoSharing);
 
                 DefaultTrace.TraceWarning("{0} occurred while OpenAsync. Exception Message: {1}", ex.ToString(), ex.Message);
             }
@@ -938,7 +947,7 @@ internal virtual void Initialize(Uri serviceEndpoint,
             servicePoint.ConnectionLimit = this.ConnectionPolicy.MaxConnectionLimit;
 #endif
 
-            this.GlobalEndpointManager = new GlobalEndpointManager(this, this.ConnectionPolicy);
+            this.GlobalEndpointManager = new GlobalEndpointManager(this, this.ConnectionPolicy, this.enableAsyncCacheExceptionNoSharing);
             this.PartitionKeyRangeLocation = this.ConnectionPolicy.EnablePartitionLevelFailover || this.ConnectionPolicy.EnablePartitionLevelCircuitBreaker
                 ? new GlobalPartitionEndpointManagerCore(
                     this.GlobalEndpointManager,
@@ -1059,8 +1068,9 @@ private async Task<bool> GetInitializationTaskAsync(IStoreClientFactory storeCli
                     storeModel: this.GatewayStoreModel, 
                     tokenProvider: this, 
                     retryPolicy: this.retryPolicy,
-                    telemetryToServiceHelper: this.telemetryToServiceHelper);
-            this.partitionKeyRangeCache = new PartitionKeyRangeCache(this, this.GatewayStoreModel, this.collectionCache, this.GlobalEndpointManager);
+                    telemetryToServiceHelper: this.telemetryToServiceHelper,
+                    enableAsyncCacheExceptionNoSharing: this.enableAsyncCacheExceptionNoSharing);
+            this.partitionKeyRangeCache = new PartitionKeyRangeCache(this, this.GatewayStoreModel, this.collectionCache, this.GlobalEndpointManager, this.enableAsyncCacheExceptionNoSharing);
             this.ResetSessionTokenRetryPolicy = new ResetSessionTokenRetryPolicyFactory(this.sessionContainer, this.collectionCache, this.retryPolicy);
 
             gatewayStoreModel.SetCaches(this.partitionKeyRangeCache, this.collectionCache);
@@ -6722,7 +6732,8 @@ private void InitializeDirectConnectivity(IStoreClientFactory storeClientFactory
                 this.accountServiceConfiguration,
                 this.ConnectionPolicy,
                 this.httpClient,
-                this.storeClientFactory.GetConnectionStateListener());
+                this.storeClientFactory.GetConnectionStateListener(),
+                this.enableAsyncCacheExceptionNoSharing);
 
             this.CreateStoreModel(subscribeRntbdStatus: true);
         }
diff --git a/Microsoft.Azure.Cosmos/src/Resource/ClientContextCore.cs b/Microsoft.Azure.Cosmos/src/Resource/ClientContextCore.cs
index 86658cf9a3..461960104e 100644
--- a/Microsoft.Azure.Cosmos/src/Resource/ClientContextCore.cs
+++ b/Microsoft.Azure.Cosmos/src/Resource/ClientContextCore.cs
@@ -85,7 +85,8 @@ internal static CosmosClientContext Create(
                cosmosClientId: cosmosClient.Id,
                remoteCertificateValidationCallback: ClientContextCore.SslCustomValidationCallBack(clientOptions.GetServerCertificateCustomValidationCallback()),
                cosmosClientTelemetryOptions: clientOptions.CosmosClientTelemetryOptions,
-               chaosInterceptorFactory: clientOptions.ChaosInterceptorFactory);
+               chaosInterceptorFactory: clientOptions.ChaosInterceptorFactory,
+               enableAsyncCacheExceptionNoSharing: clientOptions.EnableAsyncCacheExceptionNoSharing);
 
             return ClientContextCore.Create(
                 cosmosClient,
diff --git a/Microsoft.Azure.Cosmos/src/Routing/AsyncCache.cs b/Microsoft.Azure.Cosmos/src/Routing/AsyncCache.cs
index ba93c64cf2..14a27d4fdb 100644
--- a/Microsoft.Azure.Cosmos/src/Routing/AsyncCache.cs
+++ b/Microsoft.Azure.Cosmos/src/Routing/AsyncCache.cs
@@ -20,20 +20,26 @@ namespace Microsoft.Azure.Cosmos.Common
     /// <typeparam name="TValue">Type of values.</typeparam>
     internal sealed class AsyncCache<TKey, TValue>
     {
+        private readonly bool enableAsyncCacheExceptionNoSharing;
         private readonly IEqualityComparer<TValue> valueEqualityComparer;
         private readonly IEqualityComparer<TKey> keyEqualityComparer;
 
         private ConcurrentDictionary<TKey, AsyncLazy<TValue>> values;
 
-        public AsyncCache(IEqualityComparer<TValue> valueEqualityComparer, IEqualityComparer<TKey> keyEqualityComparer = null)
+        public AsyncCache(
+            IEqualityComparer<TValue> valueEqualityComparer,
+            IEqualityComparer<TKey> keyEqualityComparer = null,
+            bool enableAsyncCacheExceptionNoSharing = true)
         {
             this.keyEqualityComparer = keyEqualityComparer ?? EqualityComparer<TKey>.Default;
             this.values = new ConcurrentDictionary<TKey, AsyncLazy<TValue>>(this.keyEqualityComparer);
             this.valueEqualityComparer = valueEqualityComparer;
+            this.enableAsyncCacheExceptionNoSharing = enableAsyncCacheExceptionNoSharing;
         }
 
-        public AsyncCache()
-            : this(EqualityComparer<TValue>.Default)
+        public AsyncCache(bool enableAsyncCacheExceptionNoSharing = true)
+            : this(valueEqualityComparer: EqualityComparer<TValue>.Default,
+                  enableAsyncCacheExceptionNoSharing: enableAsyncCacheExceptionNoSharing)
         {
         }
 
diff --git a/Microsoft.Azure.Cosmos/src/Routing/AsyncCacheNonBlocking.cs b/Microsoft.Azure.Cosmos/src/Routing/AsyncCacheNonBlocking.cs
index 2d207e2b25..c0690eadd2 100644
--- a/Microsoft.Azure.Cosmos/src/Routing/AsyncCacheNonBlocking.cs
+++ b/Microsoft.Azure.Cosmos/src/Routing/AsyncCacheNonBlocking.cs
@@ -20,6 +20,7 @@ namespace Microsoft.Azure.Cosmos
     /// </summary>
     internal sealed class AsyncCacheNonBlocking<TKey, TValue> : IDisposable
     {
+        private readonly bool enableAsyncCacheExceptionNoSharing;
         private readonly CancellationTokenSource cancellationTokenSource;
         private readonly ConcurrentDictionary<TKey, AsyncLazyWithRefreshTask<TValue>> values;
         private readonly Func<Exception, bool> removeFromCacheOnBackgroundRefreshException;
@@ -30,7 +31,8 @@ internal sealed class AsyncCacheNonBlocking<TKey, TValue> : IDisposable
         public AsyncCacheNonBlocking(
             Func<Exception, bool> removeFromCacheOnBackgroundRefreshException = null,
             IEqualityComparer<TKey> keyEqualityComparer = null,
-            CancellationToken cancellationToken = default)
+            CancellationToken cancellationToken = default,
+            bool enableAsyncCacheExceptionNoSharing = true)
         {
             this.keyEqualityComparer = keyEqualityComparer ?? EqualityComparer<TKey>.Default;
             this.values = new ConcurrentDictionary<TKey, AsyncLazyWithRefreshTask<TValue>>(this.keyEqualityComparer);
@@ -38,10 +40,13 @@ public AsyncCacheNonBlocking(
             this.cancellationTokenSource = cancellationToken == default
                 ? new CancellationTokenSource()
                 : CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+            this.enableAsyncCacheExceptionNoSharing = enableAsyncCacheExceptionNoSharing;
         }
 
-        public AsyncCacheNonBlocking()
-            : this(removeFromCacheOnBackgroundRefreshException: null, keyEqualityComparer: null)
+        public AsyncCacheNonBlocking(bool enableAsyncCacheExceptionNoSharing = true)
+            : this(removeFromCacheOnBackgroundRefreshException: null,
+                  keyEqualityComparer: null,
+                  enableAsyncCacheExceptionNoSharing: enableAsyncCacheExceptionNoSharing)
         {
         }
 
@@ -279,7 +284,7 @@ public AsyncLazyWithRefreshTask(
 
             public bool IsValueCreated => this.value != null;
 
-            public Task<T> GetValueAsync(
+            public Task<T> GetValueAsync(
                 Func<T, Task<T>> createValueFunc)
             {
                 // The task was already created so just return it.
diff --git a/Microsoft.Azure.Cosmos/src/Routing/ClientCollectionCache.cs b/Microsoft.Azure.Cosmos/src/Routing/ClientCollectionCache.cs
index 534a0b8a22..995e9020ba 100644
--- a/Microsoft.Azure.Cosmos/src/Routing/ClientCollectionCache.cs
+++ b/Microsoft.Azure.Cosmos/src/Routing/ClientCollectionCache.cs
@@ -33,7 +33,9 @@ public ClientCollectionCache(
             IStoreModel storeModel,
             ICosmosAuthorizationTokenProvider tokenProvider,
             IRetryPolicyFactory retryPolicy,
-            TelemetryToServiceHelper telemetryToServiceHelper)
+            TelemetryToServiceHelper telemetryToServiceHelper,
+            bool enableAsyncCacheExceptionNoSharing = true)
+            : base(enableAsyncCacheExceptionNoSharing)
         {
             this.storeModel = storeModel ?? throw new ArgumentNullException("storeModel");
             this.tokenProvider = tokenProvider;
diff --git a/Microsoft.Azure.Cosmos/src/Routing/CollectionCache.cs b/Microsoft.Azure.Cosmos/src/Routing/CollectionCache.cs
index 32a83ffed7..caf490a8bb 100644
--- a/Microsoft.Azure.Cosmos/src/Routing/CollectionCache.cs
+++ b/Microsoft.Azure.Cosmos/src/Routing/CollectionCache.cs
@@ -28,10 +28,17 @@ internal abstract class CollectionCache
         /// </summary>
         protected class InternalCache
         {
-            internal InternalCache()
+            internal InternalCache(
+                bool enableAsyncCacheExceptionNoSharing = true)
             {
-                this.collectionInfoByName = new AsyncCache<string, ContainerProperties>(new CollectionRidComparer());
-                this.collectionInfoById = new AsyncCache<string, ContainerProperties>(new CollectionRidComparer());
+                this.collectionInfoByName = new AsyncCache<string, ContainerProperties>(
+                    new CollectionRidComparer(),
+                    enableAsyncCacheExceptionNoSharing: enableAsyncCacheExceptionNoSharing);
+
+                this.collectionInfoById = new AsyncCache<string, ContainerProperties>(
+                    new CollectionRidComparer(),
+                    enableAsyncCacheExceptionNoSharing: enableAsyncCacheExceptionNoSharing);
+
                 this.collectionInfoByNameLastRefreshTime = new ConcurrentDictionary<string, DateTime>();
                 this.collectionInfoByIdLastRefreshTime = new ConcurrentDictionary<string, DateTime>();
             }
@@ -48,11 +55,12 @@ internal InternalCache()
         /// </summary>
         protected readonly InternalCache[] cacheByApiList;
 
-        protected CollectionCache()
+        protected CollectionCache(
+            bool enableAsyncCacheExceptionNoSharing = true)
         {
             this.cacheByApiList = new InternalCache[2];
-            this.cacheByApiList[0] = new InternalCache(); // for API version < 2018-12-31
-            this.cacheByApiList[1] = new InternalCache(); // for API version >= 2018-12-31
+            this.cacheByApiList[0] = new InternalCache(enableAsyncCacheExceptionNoSharing); // for API version < 2018-12-31
+            this.cacheByApiList[1] = new InternalCache(enableAsyncCacheExceptionNoSharing); // for API version >= 2018-12-31
         }
 
         /// <summary>
diff --git a/Microsoft.Azure.Cosmos/src/Routing/GatewayAddressCache.cs b/Microsoft.Azure.Cosmos/src/Routing/GatewayAddressCache.cs
index af115c957d..74e157e7a9 100644
--- a/Microsoft.Azure.Cosmos/src/Routing/GatewayAddressCache.cs
+++ b/Microsoft.Azure.Cosmos/src/Routing/GatewayAddressCache.cs
@@ -71,14 +71,15 @@ public GatewayAddressCache(
             IConnectionStateListener connectionStateListener,
             long suboptimalPartitionForceRefreshIntervalInSeconds = 600,
             bool enableTcpConnectionEndpointRediscovery = false,
-            bool replicaAddressValidationEnabled = false)
+            bool replicaAddressValidationEnabled = false,
+            bool enableAsyncCacheExceptionNoSharing = true)
         {
             this.addressEndpoint = new Uri(serviceEndpoint + "/" + Paths.AddressPathSegment);
             this.protocol = protocol;
             this.tokenProvider = tokenProvider;
             this.serviceEndpoint = serviceEndpoint;
             this.serviceConfigReader = serviceConfigReader;
-            this.serverPartitionAddressCache = new AsyncCacheNonBlocking<PartitionKeyRangeIdentity, PartitionAddressInformation>();
+            this.serverPartitionAddressCache = new AsyncCacheNonBlocking<PartitionKeyRangeIdentity, PartitionAddressInformation>(enableAsyncCacheExceptionNoSharing);
             this.suboptimalServerPartitionTimestamps = new ConcurrentDictionary<PartitionKeyRangeIdentity, DateTime>();
             this.serverPartitionAddressToPkRangeIdMap = new ConcurrentDictionary<ServerKey, HashSet<PartitionKeyRangeIdentity>>();
             this.suboptimalMasterPartitionTimestamp = DateTime.MaxValue;
diff --git a/Microsoft.Azure.Cosmos/src/Routing/GlobalAddressResolver.cs b/Microsoft.Azure.Cosmos/src/Routing/GlobalAddressResolver.cs
index 56f5f2167e..88e142e4d3 100644
--- a/Microsoft.Azure.Cosmos/src/Routing/GlobalAddressResolver.cs
+++ b/Microsoft.Azure.Cosmos/src/Routing/GlobalAddressResolver.cs
@@ -39,6 +39,7 @@ internal sealed class GlobalAddressResolver : IAddressResolverExtension, IDispos
         private readonly ConcurrentDictionary<Uri, EndpointCache> addressCacheByEndpoint;
         private readonly bool enableTcpConnectionEndpointRediscovery;
         private readonly bool isReplicaAddressValidationEnabled;
+        private readonly bool enableAsyncCacheExceptionNoSharing;
         private readonly IConnectionStateListener connectionStateListener;
         private IOpenConnectionsHandler openConnectionsHandler;
 
@@ -52,7 +53,8 @@ public GlobalAddressResolver(
             IServiceConfigurationReader serviceConfigReader,
             ConnectionPolicy connectionPolicy,
             CosmosHttpClient httpClient,
-            IConnectionStateListener connectionStateListener)
+            IConnectionStateListener connectionStateListener,
+            bool enableAsyncCacheExceptionNoSharing = true)
         {
             this.endpointManager = endpointManager;
             this.partitionKeyRangeLocationCache = partitionKeyRangeLocationCache;
@@ -72,6 +74,8 @@ public GlobalAddressResolver(
 
             this.isReplicaAddressValidationEnabled = ConfigurationManager.IsReplicaAddressValidationEnabled(connectionPolicy);
 
+            this.enableAsyncCacheExceptionNoSharing = enableAsyncCacheExceptionNoSharing;
+
             this.maxEndpoints = maxBackupReadEndpoints + 2; // for write and alternate write endpoint (during failover)
 
             this.addressCacheByEndpoint = new ConcurrentDictionary<Uri, EndpointCache>();
@@ -344,7 +348,8 @@ private EndpointCache GetOrAddEndpoint(Uri endpoint)
                         this.openConnectionsHandler,
                         this.connectionStateListener,
                         enableTcpConnectionEndpointRediscovery: this.enableTcpConnectionEndpointRediscovery,
-                        replicaAddressValidationEnabled: this.isReplicaAddressValidationEnabled);
+                        replicaAddressValidationEnabled: this.isReplicaAddressValidationEnabled,
+                        enableAsyncCacheExceptionNoSharing: this.enableAsyncCacheExceptionNoSharing);
 
                     string location = this.endpointManager.GetLocation(endpoint);
                     AddressResolver addressResolver = new AddressResolver(null, new NullRequestSigner(), location);
diff --git a/Microsoft.Azure.Cosmos/src/Routing/GlobalEndpointManager.cs b/Microsoft.Azure.Cosmos/src/Routing/GlobalEndpointManager.cs
index db3e7dbcb1..6fe1f9c858 100644
--- a/Microsoft.Azure.Cosmos/src/Routing/GlobalEndpointManager.cs
+++ b/Microsoft.Azure.Cosmos/src/Routing/GlobalEndpointManager.cs
@@ -33,7 +33,7 @@ internal class GlobalEndpointManager : IGlobalEndpointManager
         private readonly Uri defaultEndpoint;
         private readonly ConnectionPolicy connectionPolicy;
         private readonly IDocumentClientInternal owner;
-        private readonly AsyncCache<string, AccountProperties> databaseAccountCache = new AsyncCache<string, AccountProperties>();
+        private readonly AsyncCache<string, AccountProperties> databaseAccountCache;
         private readonly TimeSpan MinTimeBetweenAccountRefresh = TimeSpan.FromSeconds(15);
         private readonly int backgroundRefreshLocationTimeIntervalInMS = GlobalEndpointManager.DefaultBackgroundRefreshLocationTimeIntervalInMS;
         private readonly object backgroundAccountRefreshLock = new object();
@@ -42,7 +42,10 @@ internal class GlobalEndpointManager : IGlobalEndpointManager
         private bool isBackgroundAccountRefreshActive = false;
         private DateTime LastBackgroundRefreshUtc = DateTime.MinValue;
 
-        public GlobalEndpointManager(IDocumentClientInternal owner, ConnectionPolicy connectionPolicy)
+        public GlobalEndpointManager(
+            IDocumentClientInternal owner,
+            ConnectionPolicy connectionPolicy,
+            bool enableAsyncCacheExceptionNoSharing = true)
         {
             this.locationCache = new LocationCache(
                 new ReadOnlyCollection<string>(connectionPolicy.PreferredLocations),
@@ -56,6 +59,7 @@ public GlobalEndpointManager(IDocumentClientInternal owner, ConnectionPolicy con
             this.connectionPolicy = connectionPolicy;
 
             this.connectionPolicy.PreferenceChanged += this.OnPreferenceChanged;
+            this.databaseAccountCache = new AsyncCache<string, AccountProperties>(enableAsyncCacheExceptionNoSharing);
 
 #if !(NETSTANDARD15 || NETSTANDARD16)
 #if NETSTANDARD20
diff --git a/Microsoft.Azure.Cosmos/src/Routing/PartitionKeyRangeCache.cs b/Microsoft.Azure.Cosmos/src/Routing/PartitionKeyRangeCache.cs
index d849e4d3c0..5fab62c6b0 100644
--- a/Microsoft.Azure.Cosmos/src/Routing/PartitionKeyRangeCache.cs
+++ b/Microsoft.Azure.Cosmos/src/Routing/PartitionKeyRangeCache.cs
@@ -36,10 +36,12 @@ public PartitionKeyRangeCache(
             ICosmosAuthorizationTokenProvider authorizationTokenProvider,
             IStoreModel storeModel,
             CollectionCache collectionCache,
-            IGlobalEndpointManager endpointManager)
+            IGlobalEndpointManager endpointManager,
+            bool enableAsyncCacheExceptionNoSharing = true)
         {
             this.routingMapCache = new AsyncCacheNonBlocking<string, CollectionRoutingMap>(
-                    keyEqualityComparer: StringComparer.Ordinal);
+                    keyEqualityComparer: StringComparer.Ordinal,
+                    enableAsyncCacheExceptionNoSharing: enableAsyncCacheExceptionNoSharing);
             this.authorizationTokenProvider = authorizationTokenProvider;
             this.storeModel = storeModel;
             this.collectionCache = collectionCache;
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Performance.Tests/Mocks/MockDocumentClient.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Performance.Tests/Mocks/MockDocumentClient.cs
index 891b56351b..9453b37aa0 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Performance.Tests/Mocks/MockDocumentClient.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Performance.Tests/Mocks/MockDocumentClient.cs
@@ -152,7 +152,7 @@ ValueTask<string> ICosmosAuthorizationTokenProvider.GetUserAuthorizationTokenAsy
 
         private void Init()
         {
-            this.collectionCache = new Mock<ClientCollectionCache>(null, new ServerStoreModel(null), null, null, null);
+            this.collectionCache = new Mock<ClientCollectionCache>(null, new ServerStoreModel(null), null, null, null, false);
 
             ContainerProperties containerProperties = ContainerProperties.CreateWithResourceId("test");
             containerProperties.PartitionKey = partitionKeyDefinition;
@@ -181,7 +181,7 @@ private void Init()
                     },
                 string.Empty);
 
-            this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, null, null, null);
+            this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, null, null, null, false);
             this.partitionKeyRangeCache.Setup(
                         m => m.TryLookupAsync(
                             It.IsAny<string>(),
@@ -209,7 +209,7 @@ private void Init()
                     It.IsAny<bool>()))
                 .Returns(Task.FromResult((IReadOnlyList<PartitionKeyRange>)result));
 
-            this.globalEndpointManager = new Mock<GlobalEndpointManager>(this, new ConnectionPolicy());
+            this.globalEndpointManager = new Mock<GlobalEndpointManager>(this, new ConnectionPolicy(), false);
 
             this.telemetryToServiceHelper = TelemetryToServiceHelper.CreateAndInitializeClientConfigAndTelemetryJob("perf-test-client",
                                                                 this.ConnectionPolicy,
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncBatcherTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncBatcherTests.cs
index 02819588ab..a250395267 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncBatcherTests.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncBatcherTests.cs
@@ -799,7 +799,7 @@ private class ClientWithSplitDetection : MockDocumentClient
 
             public ClientWithSplitDetection()
             {
-                this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(MockBehavior.Strict, null, null, null, null);
+                this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(MockBehavior.Strict, null, null, null, null, false);
                 this.partitionKeyRangeCache.Setup(
                         m => m.TryGetOverlappingRangesAsync(
                             It.IsAny<string>(),
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncContainerExecutorTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncContainerExecutorTests.cs
index 93aa72519a..d77afc627a 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncContainerExecutorTests.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncContainerExecutorTests.cs
@@ -377,7 +377,7 @@ private class ClientWithSplitDetection : MockDocumentClient
 
             public ClientWithSplitDetection()
             {
-                this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(MockBehavior.Strict, null, null, null, null);
+                this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(MockBehavior.Strict, null, null, null, null, false);
                 this.partitionKeyRangeCache.Setup(
                         m => m.TryGetOverlappingRangesAsync(
                             It.IsAny<string>(),
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncOperationContextTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncOperationContextTests.cs
index a291209dc4..b686449c42 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncOperationContextTests.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Batch/BatchAsyncOperationContextTests.cs
@@ -287,7 +287,7 @@ private class ClientWithSplitDetection : MockDocumentClient
 
             public ClientWithSplitDetection()
             {
-                this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(MockBehavior.Strict, null, null, null, null);
+                this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(MockBehavior.Strict, null, null, null, null, false);
                 this.partitionKeyRangeCache.Setup(
                         m => m.TryGetOverlappingRangesAsync(
                             It.IsAny<string>(),
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/BulkPartitionKeyRangeGoneRetryPolicyTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/BulkPartitionKeyRangeGoneRetryPolicyTests.cs
index 10cd9868cc..fe75375e6d 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/BulkPartitionKeyRangeGoneRetryPolicyTests.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/BulkPartitionKeyRangeGoneRetryPolicyTests.cs
@@ -145,7 +145,7 @@ private class ClientWithSplitDetection : MockDocumentClient
 
             public ClientWithSplitDetection()
             {
-                this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(MockBehavior.Strict, null, null, null, null);
+                this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(MockBehavior.Strict, null, null, null, null, false);
                 this.partitionKeyRangeCache.Setup(
                         m => m.TryGetOverlappingRangesAsync(
                             It.IsAny<string>(),
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/CancellationTokenTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/CancellationTokenTests.cs
index ef31029960..446b885f0f 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/CancellationTokenTests.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/CancellationTokenTests.cs
@@ -206,7 +206,7 @@ private static GatewayStoreModel MockGatewayStoreModel(Func<HttpRequestMessage,
             Mock<IDocumentClientInternal> mockDocumentClient = new Mock<IDocumentClientInternal>();
             mockDocumentClient.Setup(client => client.ServiceEndpoint).Returns(new Uri("https://foo"));
 
-            Mock<GlobalEndpointManager> endpointManager = new Mock<GlobalEndpointManager>(mockDocumentClient.Object, new ConnectionPolicy());
+            Mock<GlobalEndpointManager> endpointManager = new Mock<GlobalEndpointManager>(mockDocumentClient.Object, new ConnectionPolicy(), false);
             endpointManager.Setup(gep => gep.ResolveServiceEndpoint(It.IsAny<DocumentServiceRequest>())).Returns(new Uri("http://localhost"));
             ISessionContainer sessionContainer = new SessionContainer(string.Empty);
             HttpMessageHandler messageHandler = new MockMessageHandler(sendFunc);
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/ChangeFeed/PartitionSynchronizerCoreTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/ChangeFeed/PartitionSynchronizerCoreTests.cs
index 3974ef50b5..140e8bb098 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/ChangeFeed/PartitionSynchronizerCoreTests.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/ChangeFeed/PartitionSynchronizerCoreTests.cs
@@ -61,8 +61,9 @@ public async Task HandlePartitionGoneAsync_PKRangeBasedLease_Split()
             Mock<Routing.PartitionKeyRangeCache> pkRangeCache = new Mock<Routing.PartitionKeyRangeCache>(
                 Mock.Of<ICosmosAuthorizationTokenProvider>(),
                 Mock.Of<Documents.IStoreModel>(),
-                Mock.Of<Common.CollectionCache>(),
-                this.endpointManager);
+                new Mock<Common.CollectionCache>(false).Object,
+                this.endpointManager,
+                false);
 
             List<Documents.PartitionKeyRange> resultingRanges = new List<Documents.PartitionKeyRange>()
             {
@@ -125,8 +126,9 @@ public async Task HandlePartitionGoneAsync_EpkBasedLease_Split()
             Mock<Routing.PartitionKeyRangeCache> pkRangeCache = new Mock<Routing.PartitionKeyRangeCache>(
                 Mock.Of<ICosmosAuthorizationTokenProvider>(),
                 Mock.Of<Documents.IStoreModel>(),
-                Mock.Of<Common.CollectionCache>(),
-                this.endpointManager);
+                new Mock<Common.CollectionCache>(false).Object,
+                this.endpointManager,
+                false);
 
             List<Documents.PartitionKeyRange> resultingRanges = new List<Documents.PartitionKeyRange>()
             {
@@ -194,8 +196,9 @@ public async Task HandlePartitionGoneAsync_PKRangeBasedLease_Merge()
             Mock<Routing.PartitionKeyRangeCache> pkRangeCache = new Mock<Routing.PartitionKeyRangeCache>(
                 Mock.Of<ICosmosAuthorizationTokenProvider>(),
                 Mock.Of<Documents.IStoreModel>(),
-                Mock.Of<Common.CollectionCache>(),
-                this.endpointManager);
+                new Mock<Common.CollectionCache>(false).Object,
+                this.endpointManager,
+                false);
 
             List<Documents.PartitionKeyRange> resultingRanges = new List<Documents.PartitionKeyRange>()
             {
@@ -253,8 +256,9 @@ public async Task HandlePartitionGoneAsync_EpkBasedLease_Merge()
             Mock<Routing.PartitionKeyRangeCache> pkRangeCache = new Mock<Routing.PartitionKeyRangeCache>(
                 Mock.Of<ICosmosAuthorizationTokenProvider>(),
                 Mock.Of<Documents.IStoreModel>(),
-                Mock.Of<Common.CollectionCache>(),
-                this.endpointManager);
+                new Mock<Common.CollectionCache>(false).Object,
+                this.endpointManager,
+                false);
 
             List<Documents.PartitionKeyRange> resultingRanges = new List<Documents.PartitionKeyRange>()
             {
@@ -302,8 +306,9 @@ public async Task CreateMissingLeases_NoLeases()
             Mock<Routing.PartitionKeyRangeCache> pkRangeCache = new Mock<Routing.PartitionKeyRangeCache>(
                 Mock.Of<ICosmosAuthorizationTokenProvider>(),
                 Mock.Of<Documents.IStoreModel>(),
-                Mock.Of<Common.CollectionCache>(),
-                this.endpointManager);
+                new Mock<Common.CollectionCache>(false).Object,
+                this.endpointManager,
+                false);
 
             List<Documents.PartitionKeyRange> resultingRanges = new List<Documents.PartitionKeyRange>()
             {
@@ -348,8 +353,9 @@ public async Task CreateMissingLeases_SomePKRangeLeases()
             Mock<Routing.PartitionKeyRangeCache> pkRangeCache = new Mock<Routing.PartitionKeyRangeCache>(
                 Mock.Of<ICosmosAuthorizationTokenProvider>(),
                 Mock.Of<Documents.IStoreModel>(),
-                Mock.Of<Common.CollectionCache>(),
-                this.endpointManager);
+                new Mock<Common.CollectionCache>(false).Object,
+                this.endpointManager,
+                false);
 
             List<Documents.PartitionKeyRange> resultingRanges = new List<Documents.PartitionKeyRange>()
             {
@@ -400,8 +406,9 @@ public async Task CreateMissingLeases_SomePKRangeAndEPKLeases()
             Mock<Routing.PartitionKeyRangeCache> pkRangeCache = new Mock<Routing.PartitionKeyRangeCache>(
                 Mock.Of<ICosmosAuthorizationTokenProvider>(),
                 Mock.Of<Documents.IStoreModel>(),
-                Mock.Of<Common.CollectionCache>(),
-                this.endpointManager);
+                new Mock<Common.CollectionCache>(false).Object,
+                this.endpointManager,
+                false);
 
             List<Documents.PartitionKeyRange> resultingRanges = new List<Documents.PartitionKeyRange>()
             {
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/GatewayAddressCacheTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/GatewayAddressCacheTests.cs
index 6f58cf70cb..c1dff0e79e 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/GatewayAddressCacheTests.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/GatewayAddressCacheTests.cs
@@ -60,7 +60,7 @@ public GatewayAddressCacheTests()
                 }
             };
 
-            this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, null, null, null);
+            this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, null, null, null, false);
             this.partitionKeyRangeCache
                 .Setup(m => m.TryGetOverlappingRangesAsync(
                     It.IsAny<string>(),
@@ -636,7 +636,7 @@ public async Task GlobalAddressResolver_OpenConnectionsToAllReplicasAsync_WithVa
             containerProperties.Id = "TestId";
             containerProperties.PartitionKeyPath = "/pk";
 
-            Mock<CollectionCache> mockCollectionCahce = new(MockBehavior.Strict);
+            Mock<CollectionCache> mockCollectionCahce = new(MockBehavior.Strict, false);
             mockCollectionCahce
                 .Setup(x => x.ResolveByNameAsync(
                     It.IsAny<string>(),
@@ -715,7 +715,7 @@ public async Task GlobalAddressResolver_OpenConnectionsToAllReplicasAsync_WhenHa
             containerProperties.Id = "TestId";
             containerProperties.PartitionKeyPath = "/pk";
 
-            Mock<CollectionCache> mockCollectionCahce = new(MockBehavior.Strict);
+            Mock<CollectionCache> mockCollectionCahce = new(MockBehavior.Strict, false);
             mockCollectionCahce
                 .Setup(x => x.ResolveByNameAsync(
                     It.IsAny<string>(),
@@ -794,7 +794,7 @@ public async Task GlobalAddressResolver_OpenConnectionsToAllReplicasAsync_WhenIn
             containerProperties.Id = "TestId";
             containerProperties.PartitionKeyPath = "/pk";
 
-            Mock<CollectionCache> mockCollectionCahce = new(MockBehavior.Strict);
+            Mock<CollectionCache> mockCollectionCahce = new(MockBehavior.Strict, false);
             mockCollectionCahce
                 .Setup(x => x.ResolveByNameAsync(
                     It.IsAny<string>(),
@@ -806,7 +806,7 @@ public async Task GlobalAddressResolver_OpenConnectionsToAllReplicasAsync_WhenIn
                 .Returns(Task.FromResult(containerProperties));
 
             string exceptionMessage = "Failed to lookup partition key ranges.";
-            Mock<PartitionKeyRangeCache> partitionKeyRangeCache = new(null, null, null, null);
+            Mock<PartitionKeyRangeCache> partitionKeyRangeCache = new(null, null, null, null, false);
             partitionKeyRangeCache
                 .Setup(m => m.TryGetOverlappingRangesAsync(
                     It.IsAny<string>(),
@@ -882,7 +882,7 @@ public async Task GlobalAddressResolver_OpenConnectionsToAllReplicasAsync_WhenNu
                 RequestTimeout = TimeSpan.FromSeconds(120)
             };
 
-            Mock<CollectionCache> mockCollectionCahce = new(MockBehavior.Strict);
+            Mock<CollectionCache> mockCollectionCahce = new(MockBehavior.Strict, false);
             mockCollectionCahce
                 .Setup(x => x.ResolveByNameAsync(
                     It.IsAny<string>(),
@@ -1644,7 +1644,7 @@ public async Task GlobalAddressResolver_TryOpenConnectionToUnhealthyEndpointsAsy
             containerProperties.Id = "TestId";
             containerProperties.PartitionKeyPath = "/pk";
 
-            Mock<CollectionCache> mockCollectionCahce = new(MockBehavior.Strict);
+            Mock<CollectionCache> mockCollectionCahce = new(MockBehavior.Strict, false);
             mockCollectionCahce
                 .Setup(x => x.ResolveByNameAsync(
                     It.IsAny<string>(),
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/GatewayStoreModelTest.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/GatewayStoreModelTest.cs
index c7e248a9d2..61bca004e3 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/GatewayStoreModelTest.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/GatewayStoreModelTest.cs
@@ -280,8 +280,8 @@ await GatewayStoreModel.ApplySessionTokenAsync(
                            dsr,
                            ConsistencyLevel.Session,
                            new Mock<ISessionContainer>().Object,
-                           partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null).Object,
-                           clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null).Object,
+                           partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null, false).Object,
+                           clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null, false).Object,
                            globalEndpointManager: Mock.Of<IGlobalEndpointManager>());
 
                         Assert.IsNull(dsr.Headers[HttpConstants.HttpHeaders.SessionToken]);
@@ -307,8 +307,8 @@ await GatewayStoreModel.ApplySessionTokenAsync(
                     dsrQueryPlan,
                     ConsistencyLevel.Session,
                     new Mock<ISessionContainer>().Object,
-                    partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null).Object,
-                    clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null).Object,
+                    partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null, false).Object,
+                    clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null, false).Object,
                     globalEndpointManager: Mock.Of<IGlobalEndpointManager>());
 
                 Assert.IsNull(dsrQueryPlan.Headers[HttpConstants.HttpHeaders.SessionToken]);
@@ -361,8 +361,8 @@ await GatewayStoreModel.ApplySessionTokenAsync(
                                 dsr,
                                 ConsistencyLevel.Session,
                                 new Mock<ISessionContainer>().Object,
-                                partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null).Object,
-                                clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null).Object,
+                                partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null, false).Object,
+                                clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null, false).Object,
                                 globalEndpointManager: Mock.Of<IGlobalEndpointManager>());
 
                             Assert.AreEqual(dsrSessionToken, dsr.Headers[HttpConstants.HttpHeaders.SessionToken]);
@@ -391,8 +391,8 @@ await GatewayStoreModel.ApplySessionTokenAsync(
                                 dsrNoSessionToken,
                                 ConsistencyLevel.Session,
                                 sessionContainer,
-                                partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null).Object,
-                                clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null).Object,
+                                partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null, false).Object,
+                                clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null, false).Object,
                                 globalEndpointManager: globalEndpointManager.Object);
 
                             if (dsrNoSessionToken.IsReadOnlyRequest || dsrNoSessionToken.OperationType == OperationType.Batch || multiMaster)
@@ -426,13 +426,13 @@ await GatewayStoreModel.ApplySessionTokenAsync(
                         containerProperties.Id = "TestId";
                         containerProperties.PartitionKeyPath = "/pk";
 
-                        Mock<CollectionCache> mockCollectionCahce = new Mock<CollectionCache>(MockBehavior.Strict);
+                        Mock<CollectionCache> mockCollectionCahce = new Mock<CollectionCache>(MockBehavior.Strict, false);
                         mockCollectionCahce.Setup(x => x.ResolveCollectionAsync(
                             dsr,
                             It.IsAny<CancellationToken>(),
                             NoOpTrace.Singleton)).Returns(Task.FromResult(containerProperties));
 
-                        Mock<PartitionKeyRangeCache> mockPartitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(MockBehavior.Strict, null, null, null, null);
+                        Mock<PartitionKeyRangeCache> mockPartitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(MockBehavior.Strict, null, null, null, null, false);
                         mockPartitionKeyRangeCache.Setup(x => x.TryGetPartitionKeyRangeByIdAsync(
                             containerProperties.ResourceId,
                             partitionKeyRangeId,
@@ -479,8 +479,8 @@ await GatewayStoreModel.ApplySessionTokenAsync(
                     dsrSprocExecute,
                     ConsistencyLevel.Session,
                     new Mock<ISessionContainer>().Object,
-                    partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null).Object,
-                    clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null).Object,
+                    partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null, false).Object,
+                    clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null, false).Object,
                     globalEndpointManager: Mock.Of<IGlobalEndpointManager>());
 
                 Assert.AreEqual(sessionToken, dsrSprocExecute.Headers[HttpConstants.HttpHeaders.SessionToken]);
@@ -518,8 +518,8 @@ await GatewayStoreModel.ApplySessionTokenAsync(
                     dsrNoSessionToken,
                     ConsistencyLevel.Session,
                     sessionContainer,
-                    partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null).Object,
-                    clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null).Object,
+                    partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null, false).Object,
+                    clientCollectionCache: new Mock<ClientCollectionCache>(new SessionContainer("testhost"), gatewayStoreModel, null, null, null, false).Object,
                     globalEndpointManager: globalEndpointManager.Object);
 
                 if (isWriteRequest && multiMaster)
@@ -968,8 +968,8 @@ public async Task GatewayStoreModel_AvoidGlobalSessionToken()
                 eventSource,
                 null,
                 MockCosmosUtil.CreateCosmosHttpClient(() => new HttpClient()));
-            Mock<ClientCollectionCache> clientCollectionCache = new Mock<ClientCollectionCache>(new SessionContainer("testhost"), storeModel, null, null, null);
-            Mock<PartitionKeyRangeCache> partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, storeModel, clientCollectionCache.Object, endpointManager);
+            Mock<ClientCollectionCache> clientCollectionCache = new Mock<ClientCollectionCache>(new SessionContainer("testhost"), storeModel, null, null, null, false);
+            Mock<PartitionKeyRangeCache> partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, storeModel, clientCollectionCache.Object, endpointManager, false);
 
             sessionContainer.SetSessionToken(
                     ResourceId.NewDocumentCollectionId(42, 129).DocumentCollectionId.ToString(),
@@ -1063,9 +1063,9 @@ Task<HttpResponseMessage> sendFunc(HttpRequestMessage request)
                 null,
                 MockCosmosUtil.CreateCosmosHttpClient(() => new HttpClient(messageHandler)));
 
-            Mock<ClientCollectionCache> clientCollectionCache = new Mock<ClientCollectionCache>(new SessionContainer("testhost"), storeModel, null, null, null);
+            Mock<ClientCollectionCache> clientCollectionCache = new Mock<ClientCollectionCache>(new SessionContainer("testhost"), storeModel, null, null, null, false);
 
-            Mock<PartitionKeyRangeCache> partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, storeModel, clientCollectionCache.Object, endpointManager);
+            Mock<PartitionKeyRangeCache> partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, storeModel, clientCollectionCache.Object, endpointManager, false);
             storeModel.SetCaches(partitionKeyRangeCache.Object, clientCollectionCache.Object);
 
             INameValueCollection headers = new RequestNameValueCollection();
@@ -1133,8 +1133,8 @@ await GatewayStoreModel.ApplySessionTokenAsync(
                     documentServiceRequestToChild,
                     ConsistencyLevel.Session,
                     sessionContainer,
-                    partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null).Object,
-                    clientCollectionCache: new Mock<ClientCollectionCache>(sessionContainer, gatewayStoreModel, null, null, null).Object,
+                    partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null, false).Object,
+                    clientCollectionCache: new Mock<ClientCollectionCache>(sessionContainer, gatewayStoreModel, null, null, null, false).Object,
                     globalEndpointManager: globalEndpointManager.Object);
 
                 Assert.AreEqual($"{childPKRangeId}:{parentSession}", documentServiceRequestToChild.Headers[HttpConstants.HttpHeaders.SessionToken]);
@@ -1199,8 +1199,8 @@ await GatewayStoreModel.ApplySessionTokenAsync(
                     documentServiceRequestToChild,
                     ConsistencyLevel.Session,
                     sessionContainer,
-                    partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null).Object,
-                    clientCollectionCache: new Mock<ClientCollectionCache>(sessionContainer, gatewayStoreModel, null, null, null).Object,
+                    partitionKeyRangeCache: new Mock<PartitionKeyRangeCache>(null, null, null, null, false).Object,
+                    clientCollectionCache: new Mock<ClientCollectionCache>(sessionContainer, gatewayStoreModel, null, null, null, false).Object,
                     globalEndpointManager: globalEndpointManager.Object);
 
                 Assert.AreEqual($"{childPKRangeId}:{tokenWithAllMax}", documentServiceRequestToChild.Headers[HttpConstants.HttpHeaders.SessionToken]);
@@ -1269,8 +1269,8 @@ static async Task<HttpResponseMessage> messageHandler(HttpRequestMessage request
                 null,
                 MockCosmosUtil.CreateCosmosHttpClient(() => new HttpClient(httpMessageHandler)));
 
-            ClientCollectionCache clientCollectionCache = new Mock<ClientCollectionCache>(new SessionContainer("testhost"), storeModel, null, null, null).Object;
-            PartitionKeyRangeCache partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, storeModel, clientCollectionCache, endpointManager).Object;
+            ClientCollectionCache clientCollectionCache = new Mock<ClientCollectionCache>(new SessionContainer("testhost"), storeModel, null, null, null, false).Object;
+            PartitionKeyRangeCache partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, storeModel, clientCollectionCache, endpointManager, false).Object;
             storeModel.SetCaches(partitionKeyRangeCache, clientCollectionCache);
 
             await executeWithGatewayStoreModel(storeModel);
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/PartitionKeyRangeCacheTest.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/PartitionKeyRangeCacheTest.cs
index d7ecbe9201..41b53d293a 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/PartitionKeyRangeCacheTest.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/PartitionKeyRangeCacheTest.cs
@@ -72,7 +72,7 @@ public async Task TryGetOverlappingRangesAsync_WithFreshContainer_ShouldNotAddSa
             using (ITrace trace = Trace.GetRootTrace(this.TestContext.TestName, TraceComponent.Unknown, TraceLevel.Info))
             {
                 Mock<IStoreModel> mockStoreModel = new();
-                Mock<CollectionCache> mockCollectioNCache = new();
+                Mock<CollectionCache> mockCollectioNCache = new(false);
                 Mock<ICosmosAuthorizationTokenProvider> mockTokenProvider = new();
                 NameValueCollectionWrapper headers = new()
                 {
@@ -98,7 +98,7 @@ public async Task TryGetOverlappingRangesAsync_WithFreshContainer_ShouldNotAddSa
                     .Returns(new ValueTask<string>(authToken));
 
                 // Act.
-                PartitionKeyRangeCache partitionKeyRangeCache = new(mockTokenProvider.Object, mockStoreModel.Object, mockCollectioNCache.Object, endpointManager);
+                PartitionKeyRangeCache partitionKeyRangeCache = new(mockTokenProvider.Object, mockStoreModel.Object, mockCollectioNCache.Object, endpointManager, enableAsyncCacheExceptionNoSharing: false);
                 IReadOnlyList<PartitionKeyRange> partitionKeyRanges = await partitionKeyRangeCache.TryGetOverlappingRangesAsync(
                     containerRId,
                     FeedRangeEpk.FullRange.Range,
@@ -147,7 +147,7 @@ public async Task TryGetOverlappingRangesAsync_WhenGatewayThrowsServiceUnavailab
             using (ITrace trace = Trace.GetRootTrace(this.TestContext.TestName, TraceComponent.Unknown, TraceLevel.Info))
             {
                 Mock<IStoreModel> mockStoreModel = new();
-                Mock<CollectionCache> mockCollectioNCache = new();
+                Mock<CollectionCache> mockCollectioNCache = new(false);
                 Mock<ICosmosAuthorizationTokenProvider> mockTokenProvider = new();
                 NameValueCollectionWrapper headers = new()
                 {
@@ -207,7 +207,7 @@ public async Task TryGetOverlappingRangesAsync_WhenGatewayThrowsServiceUnavailab
                 mockTokenProvider.Setup(x => x.GetUserAuthorizationTokenAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<INameValueCollection>(), It.IsAny<AuthorizationTokenType>(), It.IsAny<ITrace>()))
                     .Returns(new ValueTask<string>(authToken));
 
-                PartitionKeyRangeCache partitionKeyRangeCache = new(mockTokenProvider.Object, mockStoreModel.Object, mockCollectioNCache.Object, mockedEndpointManager.Object);
+                PartitionKeyRangeCache partitionKeyRangeCache = new(mockTokenProvider.Object, mockStoreModel.Object, mockCollectioNCache.Object, mockedEndpointManager.Object, enableAsyncCacheExceptionNoSharing: false);
 
                 if (shouldSucceed)
                 {
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/PartitionKeyRangeHandlerTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/PartitionKeyRangeHandlerTests.cs
index ea924a0b89..e2a83f73aa 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/PartitionKeyRangeHandlerTests.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/PartitionKeyRangeHandlerTests.cs
@@ -656,7 +656,7 @@ public async Task PartitionKeyRangeGoneTracePlumbingTest()
 
             using GlobalEndpointManager endpointManager = new(mockDocumentClient.Object, new ConnectionPolicy());
 
-            Mock<Common.CollectionCache> collectionCache = new Mock<Common.CollectionCache>(MockBehavior.Strict);
+            Mock<Common.CollectionCache> collectionCache = new Mock<Common.CollectionCache>(MockBehavior.Strict, false);
             collectionCache.Setup(c => c.ResolveCollectionAsync(It.IsAny<DocumentServiceRequest>(), default, trace))
                 .ReturnsAsync(containerProperties);
 
@@ -666,7 +666,8 @@ public async Task PartitionKeyRangeGoneTracePlumbingTest()
                 new Mock<ICosmosAuthorizationTokenProvider>().Object,
                 new Mock<IStoreModel>().Object,
                 collectionCache.Object,
-                endpointManager);
+                endpointManager,
+                false);
             partitionKeyRangeCache.Setup(c => c.TryLookupAsync(collectionRid, null, It.IsAny<DocumentServiceRequest>(), trace))
                 .ReturnsAsync(collectionRoutingMap);
 
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Routing/AsyncCacheNonBlockingTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Routing/AsyncCacheNonBlockingTests.cs
index bbdcc90127..2744319455 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Routing/AsyncCacheNonBlockingTests.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Routing/AsyncCacheNonBlockingTests.cs
@@ -28,7 +28,7 @@ public class AsyncCacheNonBlockingTests
         [ExpectedException(typeof(NotFoundException))]
         public async Task ValidateNegativeScenario(bool forceRefresh)
         {
-            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>();
+            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>(enableAsyncCacheExceptionNoSharing: false);
             await asyncCache.GetAsync(
                 "test",
                 async (_) =>
@@ -42,7 +42,7 @@ await asyncCache.GetAsync(
         [TestMethod]
         public async Task ValidateMultipleBackgroundRefreshesScenario()
         {
-            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>();
+            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>(enableAsyncCacheExceptionNoSharing: false);
 
             string expectedValue = "ResponseValue";
             string response = await asyncCache.GetAsync(
@@ -75,7 +75,7 @@ public async Task ValidateMultipleBackgroundRefreshesScenario()
         [ExpectedException(typeof(NotFoundException))]
         public async Task ValidateNegativeNotAwaitedScenario()
         {
-            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>();
+            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>(enableAsyncCacheExceptionNoSharing: false);
             Task task1 = asyncCache.GetAsync(
                 "test",
                 async (_) =>
@@ -105,7 +105,7 @@ await asyncCache.GetAsync(
         public async Task ValidateNotFoundOnBackgroundRefreshRemovesFromCacheScenario()
         {
             string value1 = "Response1Value";
-            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>();
+            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>(enableAsyncCacheExceptionNoSharing: false);
             string response1 = await asyncCache.GetAsync(
                 "test",
                 async (_) =>
@@ -171,7 +171,7 @@ await asyncCache.GetAsync(
         [Owner("jawilley")]
         public async Task ValidateAsyncCacheNonBlocking()
         {
-            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>();
+            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>(enableAsyncCacheExceptionNoSharing: false);
             string result = await asyncCache.GetAsync(
                 "test",
                 (_) => Task.FromResult("test2"),
@@ -212,7 +212,7 @@ public async Task ValidateAsyncCacheNonBlocking()
         [Owner("jawilley")]
         public async Task ValidateCacheValueIsRemovedAfterException()
         {
-            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>();
+            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>(enableAsyncCacheExceptionNoSharing: false);
             string result = await asyncCache.GetAsync(
                 key: "test",
                 singleValueInitFunc: (_) => Task.FromResult("test2"),
@@ -268,7 +268,7 @@ await asyncCache.GetAsync(
         [Owner("jawilley")]
         public async Task ValidateConcurrentCreateAsyncCacheNonBlocking()
         {
-            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>();
+            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>(enableAsyncCacheExceptionNoSharing: false);
             int totalLazyCalls = 0;
 
             List<Task> tasks = new List<Task>();
@@ -292,7 +292,7 @@ public async Task ValidateConcurrentCreateAsyncCacheNonBlocking()
         [Owner("jawilley")]
         public async Task ValidateConcurrentCreateWithFailureAsyncCacheNonBlocking()
         {
-            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>();
+            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>(enableAsyncCacheExceptionNoSharing: false);
             int totalLazyCalls = 0;
 
             Random random = new Random();
@@ -331,7 +331,7 @@ await asyncCache.GetAsync(
         [Owner("jawilley")]
         public async Task ValidateExceptionScenariosCacheNonBlocking()
         {
-            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>();
+            AsyncCacheNonBlocking<string, string> asyncCache = new AsyncCacheNonBlocking<string, string>(enableAsyncCacheExceptionNoSharing: false);
             int totalLazyCalls = 0;
 
             try
@@ -402,7 +402,7 @@ await asyncCache.GetAsync(
         public async Task Refresh_WhenRefreshRequestedForAnExistingKey_ShouldRefreshTheCache()
         {
             // Arrange.
-            AsyncCacheNonBlocking<string, string> asyncCache = new();
+            AsyncCacheNonBlocking<string, string> asyncCache = new (enableAsyncCacheExceptionNoSharing: false);
 
             // Act and Assert.
             string result = await asyncCache.GetAsync(
@@ -437,7 +437,7 @@ await Task.Delay(
         public async Task Refresh_WhenThrowsDocumentClientException_ShouldRemoveKeyFromTheCache()
         {
             // Arrange.
-            AsyncCacheNonBlocking<string, string> asyncCache = new();
+            AsyncCacheNonBlocking<string, string> asyncCache = new (enableAsyncCacheExceptionNoSharing: false);
 
             // Act and Assert.
             string result = await asyncCache.GetAsync(
@@ -491,7 +491,7 @@ await Task.Delay(
         public async Task Refresh_WhenThrowsOtherException_ShouldNotRemoveKeyFromTheCache()
         {
             // Arrange.
-            AsyncCacheNonBlocking<string, string> asyncCache = new();
+            AsyncCacheNonBlocking<string, string> asyncCache = new (enableAsyncCacheExceptionNoSharing: false);
 
             // Act and Assert.
             string result = await asyncCache.GetAsync(
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Utils/MockDocumentClient.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Utils/MockDocumentClient.cs
index 357fe9295f..1b6f50ed06 100644
--- a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Utils/MockDocumentClient.cs
+++ b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Utils/MockDocumentClient.cs
@@ -156,7 +156,7 @@ internal virtual PartitionKeyRange ResolvePartitionKeyRangeById(string collectio
 
         private void Init()
         {
-            this.collectionCache = new Mock<ClientCollectionCache>(new SessionContainer("testhost"), new ServerStoreModel(null), null, null, null);
+            this.collectionCache = new Mock<ClientCollectionCache>(new SessionContainer("testhost"), new ServerStoreModel(null), null, null, null, true);
             const string pkPath = "/pk";
             this.collectionCache.Setup
                     (m =>
@@ -229,7 +229,7 @@ private void Init()
 
             using GlobalEndpointManager endpointManager = new(mockDocumentClient.Object, new ConnectionPolicy());
 
-            this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, null, null, endpointManager);
+            this.partitionKeyRangeCache = new Mock<PartitionKeyRangeCache>(null, null, null, endpointManager, false);
             this.partitionKeyRangeCache.Setup(
                         m => m.TryLookupAsync(
                             It.IsAny<string>(),
@@ -260,7 +260,7 @@ private void Init()
             ).Returns((string collectionRid, string pkRangeId, ITrace trace, bool forceRefresh) =>
             Task.FromResult<PartitionKeyRange>(this.ResolvePartitionKeyRangeById(collectionRid, pkRangeId, forceRefresh)));
 
-            this.MockGlobalEndpointManager = new Mock<GlobalEndpointManager>(this, new ConnectionPolicy());
+            this.MockGlobalEndpointManager = new Mock<GlobalEndpointManager>(this, new ConnectionPolicy(), false);
             this.MockGlobalEndpointManager.Setup(gep => gep.ResolveServiceEndpoint(It.IsAny<DocumentServiceRequest>())).Returns(new Uri("http://localhost"));
             this.MockGlobalEndpointManager.Setup(gep => gep.InitializeAccountPropertiesAndStartBackgroundRefresh(It.IsAny<AccountProperties>()));
             SessionContainer sessionContainer = new SessionContainer(this.ServiceEndpoint.Host);