diff --git a/sdk/core/azure-core/inc/azure/core/internal/client_options.hpp b/sdk/core/azure-core/inc/azure/core/internal/client_options.hpp
index d52832a7c..84467241a 100644
--- a/sdk/core/azure-core/inc/azure/core/internal/client_options.hpp
+++ b/sdk/core/azure-core/inc/azure/core/internal/client_options.hpp
@@ -41,30 +41,45 @@ namespace Azure { namespace Core { namespace Internal {
      * @brief Move each policy from \p options into the new instance.
      *
      */
-    ClientOptions(ClientOptions&& options) = default;
+    ClientOptions(ClientOptions&&) = default;
 
     /**
      * @brief Copy each policy to the new instance.
      *
      */
-    ClientOptions(ClientOptions const& options)
-        : Retry(options.Retry), Transport(options.Transport), Telemetry(options.Telemetry),
-          Log(options.Log)
-    {
-      PerOperationPolicies.reserve(options.PerOperationPolicies.size());
-      for (auto& policy : options.PerOperationPolicies)
-      {
-        PerOperationPolicies.emplace_back(policy->Clone());
-      }
-      PerRetryPolicies.reserve(options.PerRetryPolicies.size());
-      for (auto& policy : options.PerRetryPolicies)
-      {
-        PerRetryPolicies.emplace_back(policy->Clone());
-      }
-    }
+    ClientOptions(ClientOptions const& other) { *this = other; }
 
     ClientOptions() = default;
 
+    /**
+     * @brief Move each policy from \p options into the this instance.
+     *
+     */
+    ClientOptions& operator=(ClientOptions&&) = default;
+
+    /**
+     * @brief Copy each policy to the this instance.
+     *
+     */
+    ClientOptions& operator=(const ClientOptions& other)
+    {
+      this->Retry = other.Retry;
+      this->Transport = other.Transport;
+      this->Telemetry = other.Telemetry;
+      this->Log = other.Log;
+      this->PerOperationPolicies.reserve(other.PerOperationPolicies.size());
+      for (auto& policy : other.PerOperationPolicies)
+      {
+        this->PerOperationPolicies.emplace_back(policy->Clone());
+      }
+      this->PerRetryPolicies.reserve(other.PerRetryPolicies.size());
+      for (auto& policy : other.PerRetryPolicies)
+      {
+        this->PerRetryPolicies.emplace_back(policy->Clone());
+      }
+      return *this;
+    }
+
     /**
      * @brief Specify the number of retries and other retry-related options.
      */
diff --git a/sdk/storage/azure-storage-blobs/CHANGELOG.md b/sdk/storage/azure-storage-blobs/CHANGELOG.md
index 49c0e3e3f..ff19225da 100644
--- a/sdk/storage/azure-storage-blobs/CHANGELOG.md
+++ b/sdk/storage/azure-storage-blobs/CHANGELOG.md
@@ -4,12 +4,13 @@
 
 ### New Features
 
-- Added support for customized application ID.
+- Added support for telemetry options.
 
 ### Breaking Changes
 
 - Changed the return type of `StartCopyFromUri` and `StartCopyIncremental` API from a `Response<T>` to the particular `Operation<T>` type called `StartCopyBlobOperation` directly.
 - String conversion functions of extensible enums were renamed from `Get()` to `ToString()`.
+- Moved `SecondaryHostForRetryReads` out of retry options, now it's under `BlobClientOptions`.
 
 ### Other Changes and Improvements
 
diff --git a/sdk/storage/azure-storage-blobs/inc/azure/storage/blobs/blob_options.hpp b/sdk/storage/azure-storage-blobs/inc/azure/storage/blobs/blob_options.hpp
index fd31cad8f..ce3951975 100644
--- a/sdk/storage/azure-storage-blobs/inc/azure/storage/blobs/blob_options.hpp
+++ b/sdk/storage/azure-storage-blobs/inc/azure/storage/blobs/blob_options.hpp
@@ -9,10 +9,10 @@
 #include <string>
 #include <vector>
 
+#include <azure/core/internal/client_options.hpp>
 #include <azure/core/match_conditions.hpp>
 #include <azure/core/modified_conditions.hpp>
 #include <azure/storage/common/access_conditions.hpp>
-#include <azure/storage/common/storage_retry_policy.hpp>
 
 #include "azure/storage/blobs/protocol/blob_rest_client.hpp"
 
@@ -125,20 +125,8 @@ namespace Azure { namespace Storage { namespace Blobs {
   /**
    * @brief Client options used to initalize all kinds of blob clients.
    */
-  struct BlobClientOptions
+  struct BlobClientOptions : Azure::Core::Internal::ClientOptions
   {
-    /**
-     * @brief Transport pipeline policies for authentication, additional HTTP headers, etc., that
-     * are applied to every request.
-     */
-    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> PerOperationPolicies;
-
-    /**
-     * @brief Transport pipeline policies for authentication, additional HTTP headers, etc., that
-     * are applied to every retrial.
-     */
-    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> PerRetryPolicies;
-
     /**
      * @brief Holds the customer provided key used when making requests.
      */
@@ -150,19 +138,13 @@ namespace Azure { namespace Storage { namespace Blobs {
     Azure::Core::Nullable<std::string> EncryptionScope;
 
     /**
-     * @brief Specify the number of retries and other retry-related options.
+     * SecondaryHostForRetryReads specifies whether the retry policy should retry a read
+     * operation against another host. If SecondaryHostForRetryReads is "" (the default) then
+     * operations are not retried against another host. NOTE: Before setting this field, make sure
+     * you understand the issues around reading stale & potentially-inconsistent data at this
+     * webpage: https://docs.microsoft.com/en-us/azure/storage/common/geo-redundant-design.
      */
-    StorageRetryWithSecondaryOptions RetryOptions;
-
-    /**
-     * @brief Customized HTTP client. We're going to use the default one if this is empty.
-     */
-    Azure::Core::Http::TransportOptions TransportOptions;
-
-    /**
-     * @brief The last part of the user agent for telemetry.
-     */
-    std::string ApplicationId;
+    std::string SecondaryHostForRetryReads;
 
     /**
      * API version used by this client.
diff --git a/sdk/storage/azure-storage-blobs/src/blob_client.cpp b/sdk/storage/azure-storage-blobs/src/blob_client.cpp
index b2c2158f5..05d1271d6 100644
--- a/sdk/storage/azure-storage-blobs/src/blob_client.cpp
+++ b/sdk/storage/azure-storage-blobs/src/blob_client.cpp
@@ -10,6 +10,8 @@
 #include <azure/storage/common/reliable_stream.hpp>
 #include <azure/storage/common/shared_key_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
+#include <azure/storage/common/storage_per_retry_policy.hpp>
+#include <azure/storage/common/storage_switch_to_secondary_policy.hpp>
 
 #include "azure/storage/blobs/append_blob_client.hpp"
 #include "azure/storage/blobs/block_blob_client.hpp"
@@ -45,16 +47,28 @@ namespace Azure { namespace Storage { namespace Blobs {
       const BlobClientOptions& options)
       : BlobClient(blobUrl, options)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    BlobClientOptions newOptions = options;
+    newOptions.PerRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::SharedKeyPolicy>(credential));
+
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_blobUrl.GetHost(), newOptions.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = newOptions.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Storage::Details::SharedKeyPolicy>(credential),
-            options));
+        newOptions,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   BlobClient::BlobClient(
@@ -63,35 +77,55 @@ namespace Azure { namespace Storage { namespace Blobs {
       const BlobClientOptions& options)
       : BlobClient(blobUrl, options)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
-    Azure::Core::Http::TokenRequestOptions tokenOptions;
-    tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_blobUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::TokenRequestOptions tokenOptions;
+      tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+      perRetryPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
+              credential, tokenOptions));
+    }
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
-                credential, tokenOptions),
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   BlobClient::BlobClient(const std::string& blobUrl, const BlobClientOptions& options)
       : m_blobUrl(blobUrl), m_customerProvidedKey(options.CustomerProvidedKey),
         m_encryptionScope(options.EncryptionScope)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_blobUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            nullptr,
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   BlockBlobClient BlobClient::AsBlockBlobClient() const { return BlockBlobClient(*this); }
@@ -152,7 +186,7 @@ namespace Azure { namespace Storage { namespace Blobs {
     }
 
     auto downloadResponse = Details::BlobRestClient::Blob::Download(
-        context, *m_pipeline, m_blobUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context), *m_pipeline, m_blobUrl, protocolLayerOptions);
 
     {
       // In case network failure during reading the body
@@ -445,7 +479,7 @@ namespace Azure { namespace Storage { namespace Blobs {
       protocolLayerOptions.EncryptionAlgorithm = m_customerProvidedKey.GetValue().Algorithm;
     }
     auto response = Details::BlobRestClient::Blob::GetProperties(
-        context, *m_pipeline, m_blobUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context), *m_pipeline, m_blobUrl, protocolLayerOptions);
     if (response->Tier.HasValue() && !response->IsAccessTierInferred.HasValue())
     {
       response->IsAccessTierInferred = false;
@@ -659,7 +693,7 @@ namespace Azure { namespace Storage { namespace Blobs {
     Details::BlobRestClient::Blob::GetBlobTagsOptions protocolLayerOptions;
     protocolLayerOptions.IfTags = options.AccessConditions.TagConditions;
     return Details::BlobRestClient::Blob::GetTags(
-        context, *m_pipeline, m_blobUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context), *m_pipeline, m_blobUrl, protocolLayerOptions);
   }
 
 }}} // namespace Azure::Storage::Blobs
diff --git a/sdk/storage/azure-storage-blobs/src/blob_container_client.cpp b/sdk/storage/azure-storage-blobs/src/blob_container_client.cpp
index f4f15d3f3..56e0046ed 100644
--- a/sdk/storage/azure-storage-blobs/src/blob_container_client.cpp
+++ b/sdk/storage/azure-storage-blobs/src/blob_container_client.cpp
@@ -7,6 +7,8 @@
 #include <azure/storage/common/constants.hpp>
 #include <azure/storage/common/shared_key_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
+#include <azure/storage/common/storage_per_retry_policy.hpp>
+#include <azure/storage/common/storage_switch_to_secondary_policy.hpp>
 
 #include "azure/storage/blobs/append_blob_client.hpp"
 #include "azure/storage/blobs/block_blob_client.hpp"
@@ -41,16 +43,28 @@ namespace Azure { namespace Storage { namespace Blobs {
       const BlobClientOptions& options)
       : BlobContainerClient(blobContainerUrl, options)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    BlobClientOptions newOptions = options;
+    newOptions.PerRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::SharedKeyPolicy>(credential));
+
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_blobContainerUrl.GetHost(), newOptions.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = newOptions.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Storage::Details::SharedKeyPolicy>(credential),
-            options));
+        newOptions,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   BlobContainerClient::BlobContainerClient(
@@ -59,19 +73,31 @@ namespace Azure { namespace Storage { namespace Blobs {
       const BlobClientOptions& options)
       : BlobContainerClient(blobContainerUrl, options)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
-    Azure::Core::Http::TokenRequestOptions tokenOptions;
-    tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_blobContainerUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::TokenRequestOptions tokenOptions;
+      tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+      perRetryPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
+              credential, tokenOptions));
+    }
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
-                credential, tokenOptions),
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   BlobContainerClient::BlobContainerClient(
@@ -80,16 +106,24 @@ namespace Azure { namespace Storage { namespace Blobs {
       : m_blobContainerUrl(blobContainerUrl), m_customerProvidedKey(options.CustomerProvidedKey),
         m_encryptionScope(options.EncryptionScope)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_blobContainerUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            nullptr,
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   BlobClient BlobContainerClient::GetBlobClient(const std::string& blobName) const
@@ -193,7 +227,10 @@ namespace Azure { namespace Storage { namespace Blobs {
     Details::BlobRestClient::BlobContainer::GetBlobContainerPropertiesOptions protocolLayerOptions;
     protocolLayerOptions.LeaseId = options.AccessConditions.LeaseId;
     return Details::BlobRestClient::BlobContainer::GetProperties(
-        context, *m_pipeline, m_blobContainerUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context),
+        *m_pipeline,
+        m_blobContainerUrl,
+        protocolLayerOptions);
   }
 
   Azure::Core::Response<Models::SetBlobContainerMetadataResult> BlobContainerClient::SetMetadata(
@@ -219,7 +256,10 @@ namespace Azure { namespace Storage { namespace Blobs {
     protocolLayerOptions.MaxResults = options.PageSizeHint;
     protocolLayerOptions.Include = options.Include;
     auto response = Details::BlobRestClient::BlobContainer::ListBlobsSinglePage(
-        context, *m_pipeline, m_blobContainerUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context),
+        *m_pipeline,
+        m_blobContainerUrl,
+        protocolLayerOptions);
     for (auto& i : response->Items)
     {
       if (i.Details.Tier.HasValue() && !i.Details.IsAccessTierInferred.HasValue())
@@ -252,7 +292,10 @@ namespace Azure { namespace Storage { namespace Blobs {
     protocolLayerOptions.MaxResults = options.PageSizeHint;
     protocolLayerOptions.Include = options.Include;
     auto response = Details::BlobRestClient::BlobContainer::ListBlobsByHierarchySinglePage(
-        context, *m_pipeline, m_blobContainerUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context),
+        *m_pipeline,
+        m_blobContainerUrl,
+        protocolLayerOptions);
     for (auto& i : response->Items)
     {
       if (i.VersionId.HasValue() && !i.IsCurrentVersion.HasValue())
@@ -272,7 +315,10 @@ namespace Azure { namespace Storage { namespace Blobs {
         protocolLayerOptions;
     protocolLayerOptions.LeaseId = options.AccessConditions.LeaseId;
     return Details::BlobRestClient::BlobContainer::GetAccessPolicy(
-        context, *m_pipeline, m_blobContainerUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context),
+        *m_pipeline,
+        m_blobContainerUrl,
+        protocolLayerOptions);
   }
 
   Azure::Core::Response<Models::SetBlobContainerAccessPolicyResult>
diff --git a/sdk/storage/azure-storage-blobs/src/blob_service_client.cpp b/sdk/storage/azure-storage-blobs/src/blob_service_client.cpp
index ff9ae53bc..3a4eb8a03 100644
--- a/sdk/storage/azure-storage-blobs/src/blob_service_client.cpp
+++ b/sdk/storage/azure-storage-blobs/src/blob_service_client.cpp
@@ -7,6 +7,8 @@
 #include <azure/storage/common/constants.hpp>
 #include <azure/storage/common/shared_key_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
+#include <azure/storage/common/storage_per_retry_policy.hpp>
+#include <azure/storage/common/storage_switch_to_secondary_policy.hpp>
 
 #include "azure/storage/blobs/version.hpp"
 
@@ -34,56 +36,89 @@ namespace Azure { namespace Storage { namespace Blobs {
       const std::string& serviceUrl,
       std::shared_ptr<StorageSharedKeyCredential> credential,
       const BlobClientOptions& options)
-      : m_serviceUrl(serviceUrl)
+      : BlobServiceClient(serviceUrl, options)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    BlobClientOptions newOptions = options;
+    newOptions.PerRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::SharedKeyPolicy>(credential));
+
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_serviceUrl.GetHost(), newOptions.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = newOptions.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Storage::Details::SharedKeyPolicy>(credential),
-            options));
+        newOptions,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   BlobServiceClient::BlobServiceClient(
       const std::string& serviceUrl,
       std::shared_ptr<Core::TokenCredential> credential,
       const BlobClientOptions& options)
-      : m_serviceUrl(serviceUrl)
+      : BlobServiceClient(serviceUrl, options)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
-    Azure::Core::Http::TokenRequestOptions tokenOptions;
-    tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_serviceUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::TokenRequestOptions tokenOptions;
+      tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+      perRetryPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
+              credential, tokenOptions));
+    }
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
-                credential, tokenOptions),
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   BlobServiceClient::BlobServiceClient(
       const std::string& serviceUrl,
       const BlobClientOptions& options)
-      : m_serviceUrl(serviceUrl)
+      : m_serviceUrl(serviceUrl), m_customerProvidedKey(options.CustomerProvidedKey),
+        m_encryptionScope(options.EncryptionScope)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_serviceUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            nullptr,
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   BlobContainerClient BlobServiceClient::GetBlobContainerClient(
@@ -106,7 +141,10 @@ namespace Azure { namespace Storage { namespace Blobs {
     protocolLayerOptions.MaxResults = options.PageSizeHint;
     protocolLayerOptions.Include = options.Include;
     return Details::BlobRestClient::Service::ListBlobContainersSinglePage(
-        context, *m_pipeline, m_serviceUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context),
+        *m_pipeline,
+        m_serviceUrl,
+        protocolLayerOptions);
   }
 
   Azure::Core::Response<Models::GetUserDelegationKeyResult> BlobServiceClient::GetUserDelegationKey(
@@ -118,7 +156,10 @@ namespace Azure { namespace Storage { namespace Blobs {
     protocolLayerOptions.StartsOn = options.startsOn;
     protocolLayerOptions.ExpiresOn = expiresOn;
     return Details::BlobRestClient::Service::GetUserDelegationKey(
-        context, *m_pipeline, m_serviceUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context),
+        *m_pipeline,
+        m_serviceUrl,
+        protocolLayerOptions);
   }
 
   Azure::Core::Response<Models::SetServicePropertiesResult> BlobServiceClient::SetProperties(
@@ -140,7 +181,10 @@ namespace Azure { namespace Storage { namespace Blobs {
     (void)options;
     Details::BlobRestClient::Service::GetServicePropertiesOptions protocolLayerOptions;
     return Details::BlobRestClient::Service::GetProperties(
-        context, *m_pipeline, m_serviceUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context),
+        *m_pipeline,
+        m_serviceUrl,
+        protocolLayerOptions);
   }
 
   Azure::Core::Response<Models::GetAccountInfoResult> BlobServiceClient::GetAccountInfo(
@@ -150,7 +194,10 @@ namespace Azure { namespace Storage { namespace Blobs {
     (void)options;
     Details::BlobRestClient::Service::GetAccountInfoOptions protocolLayerOptions;
     return Details::BlobRestClient::Service::GetAccountInfo(
-        context, *m_pipeline, m_serviceUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context),
+        *m_pipeline,
+        m_serviceUrl,
+        protocolLayerOptions);
   }
 
   Azure::Core::Response<Models::GetServiceStatisticsResult> BlobServiceClient::GetStatistics(
@@ -174,7 +221,10 @@ namespace Azure { namespace Storage { namespace Blobs {
     protocolLayerOptions.ContinuationToken = options.ContinuationToken;
     protocolLayerOptions.MaxResults = options.PageSizeHint;
     return Details::BlobRestClient::Service::FindBlobsByTagsSinglePage(
-        context, *m_pipeline, m_serviceUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context),
+        *m_pipeline,
+        m_serviceUrl,
+        protocolLayerOptions);
   }
 
   Azure::Core::Response<BlobContainerClient> BlobServiceClient::CreateBlobContainer(
diff --git a/sdk/storage/azure-storage-blobs/src/block_blob_client.cpp b/sdk/storage/azure-storage-blobs/src/block_blob_client.cpp
index c2e3b8e06..0afc09a5b 100644
--- a/sdk/storage/azure-storage-blobs/src/block_blob_client.cpp
+++ b/sdk/storage/azure-storage-blobs/src/block_blob_client.cpp
@@ -8,6 +8,7 @@
 #include <azure/storage/common/crypt.hpp>
 #include <azure/storage/common/file_io.hpp>
 #include <azure/storage/common/storage_common.hpp>
+#include <azure/storage/common/storage_switch_to_secondary_policy.hpp>
 
 namespace Azure { namespace Storage { namespace Blobs {
 
@@ -320,7 +321,7 @@ namespace Azure { namespace Storage { namespace Blobs {
     protocolLayerOptions.LeaseId = options.AccessConditions.LeaseId;
     protocolLayerOptions.IfTags = options.AccessConditions.TagConditions;
     return Details::BlobRestClient::BlockBlob::GetBlockList(
-        context, *m_pipeline, m_blobUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context), *m_pipeline, m_blobUrl, protocolLayerOptions);
   }
 
 }}} // namespace Azure::Storage::Blobs
diff --git a/sdk/storage/azure-storage-blobs/src/page_blob_client.cpp b/sdk/storage/azure-storage-blobs/src/page_blob_client.cpp
index 47d0e233b..b7a2e78e2 100644
--- a/sdk/storage/azure-storage-blobs/src/page_blob_client.cpp
+++ b/sdk/storage/azure-storage-blobs/src/page_blob_client.cpp
@@ -7,6 +7,7 @@
 #include <azure/storage/common/constants.hpp>
 #include <azure/storage/common/file_io.hpp>
 #include <azure/storage/common/storage_common.hpp>
+#include <azure/storage/common/storage_switch_to_secondary_policy.hpp>
 
 namespace Azure { namespace Storage { namespace Blobs {
 
@@ -248,7 +249,7 @@ namespace Azure { namespace Storage { namespace Blobs {
     protocolLayerOptions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
     protocolLayerOptions.IfTags = options.AccessConditions.TagConditions;
     return Details::BlobRestClient::PageBlob::GetPageRanges(
-        context, *m_pipeline, m_blobUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context), *m_pipeline, m_blobUrl, protocolLayerOptions);
   }
 
   Azure::Core::Response<Models::GetPageBlobPageRangesResult> PageBlobClient::GetPageRangesDiff(
@@ -266,7 +267,7 @@ namespace Azure { namespace Storage { namespace Blobs {
     protocolLayerOptions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
     protocolLayerOptions.IfTags = options.AccessConditions.TagConditions;
     return Details::BlobRestClient::PageBlob::GetPageRanges(
-        context, *m_pipeline, m_blobUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context), *m_pipeline, m_blobUrl, protocolLayerOptions);
   }
 
   Azure::Core::Response<Models::GetPageBlobPageRangesResult>
@@ -285,7 +286,7 @@ namespace Azure { namespace Storage { namespace Blobs {
     protocolLayerOptions.IfNoneMatch = options.AccessConditions.IfNoneMatch;
     protocolLayerOptions.IfTags = options.AccessConditions.TagConditions;
     return Details::BlobRestClient::PageBlob::GetPageRanges(
-        context, *m_pipeline, m_blobUrl, protocolLayerOptions);
+        Storage::Details::WithReplicaStatus(context), *m_pipeline, m_blobUrl, protocolLayerOptions);
   }
 
   StartCopyBlobOperation PageBlobClient::StartCopyIncremental(
diff --git a/sdk/storage/azure-storage-blobs/test/ut/storage_retry_policy_test.cpp b/sdk/storage/azure-storage-blobs/test/ut/storage_retry_policy_test.cpp
index de588bf7a..117410585 100644
--- a/sdk/storage/azure-storage-blobs/test/ut/storage_retry_policy_test.cpp
+++ b/sdk/storage/azure-storage-blobs/test/ut/storage_retry_policy_test.cpp
@@ -279,7 +279,7 @@ namespace Azure { namespace Storage { namespace Test {
     Blobs::BlobClientOptions clientOptions;
     clientOptions.PerRetryPolicies.emplace_back(std::move(transportPolicyPtr));
     int64_t delayMs = 1000;
-    clientOptions.RetryOptions.RetryDelay = std::chrono::milliseconds(delayMs);
+    clientOptions.Retry.RetryDelay = std::chrono::milliseconds(delayMs);
     auto blobClient = Azure::Storage::Blobs::BlobClient::CreateFromConnectionString(
         StandardStorageConnectionString(), RandomString(), RandomString(), clientOptions);
     auto timeBegin = std::chrono::steady_clock::now();
@@ -293,7 +293,7 @@ namespace Azure { namespace Storage { namespace Test {
     int64_t elapsedTime
         = std::chrono::duration_cast<std::chrono::milliseconds>(timeEnd - timeBegin).count();
     EXPECT_GE(elapsedTime, delayMs * 0.5);
-    EXPECT_LE(elapsedTime, delayMs * 2);
+    EXPECT_LE(elapsedTime, delayMs * 4);
   }
 
   TEST(StorageRetryPolicyTest, Failover)
@@ -315,7 +315,7 @@ namespace Azure { namespace Storage { namespace Test {
 
     Blobs::BlobClientOptions clientOptions;
     clientOptions.PerRetryPolicies.emplace_back(std::move(transportPolicyPtr));
-    clientOptions.RetryOptions.RetryDelay = std::chrono::milliseconds(0);
+    clientOptions.Retry.RetryDelay = std::chrono::milliseconds(0);
     {
       std::string primaryUrl
           = Azure::Storage::Blobs::BlobClient::CreateFromConnectionString(
@@ -323,7 +323,7 @@ namespace Azure { namespace Storage { namespace Test {
                 .GetUrl();
       std::string secondaryUrl = InferSecondaryUrl(primaryUrl);
       std::string secondaryHost = Core::Http::Url(secondaryUrl).GetHost();
-      clientOptions.RetryOptions.SecondaryHostForRetryReads = secondaryHost;
+      clientOptions.SecondaryHostForRetryReads = secondaryHost;
     }
     auto blobClient = Azure::Storage::Blobs::BlobClient::CreateFromConnectionString(
         StandardStorageConnectionString(), RandomString(), RandomString(), clientOptions);
@@ -363,8 +363,8 @@ namespace Azure { namespace Storage { namespace Test {
 
     Blobs::BlobClientOptions clientOptions;
     clientOptions.PerRetryPolicies.emplace_back(std::move(transportPolicyPtr));
-    clientOptions.RetryOptions.MaxRetries = 3;
-    clientOptions.RetryOptions.RetryDelay = std::chrono::milliseconds(0);
+    clientOptions.Retry.MaxRetries = 3;
+    clientOptions.Retry.RetryDelay = std::chrono::milliseconds(0);
     {
       std::string primaryUrl
           = Azure::Storage::Blobs::BlobClient::CreateFromConnectionString(
@@ -372,7 +372,7 @@ namespace Azure { namespace Storage { namespace Test {
                 .GetUrl();
       std::string secondaryUrl = InferSecondaryUrl(primaryUrl);
       std::string secondaryHost = Core::Http::Url(secondaryUrl).GetHost();
-      clientOptions.RetryOptions.SecondaryHostForRetryReads = secondaryHost;
+      clientOptions.SecondaryHostForRetryReads = secondaryHost;
     }
     auto blobClient = Azure::Storage::Blobs::BlobClient::CreateFromConnectionString(
         StandardStorageConnectionString(), RandomString(), RandomString(), clientOptions);
@@ -418,8 +418,8 @@ namespace Azure { namespace Storage { namespace Test {
 
     Blobs::BlobClientOptions clientOptions;
     clientOptions.PerRetryPolicies.emplace_back(std::move(transportPolicyPtr));
-    clientOptions.RetryOptions.MaxRetries = 3;
-    clientOptions.RetryOptions.RetryDelay = std::chrono::milliseconds(0);
+    clientOptions.Retry.MaxRetries = 3;
+    clientOptions.Retry.RetryDelay = std::chrono::milliseconds(0);
     {
       std::string primaryUrl
           = Azure::Storage::Blobs::BlobClient::CreateFromConnectionString(
@@ -427,7 +427,7 @@ namespace Azure { namespace Storage { namespace Test {
                 .GetUrl();
       std::string secondaryUrl = InferSecondaryUrl(primaryUrl);
       std::string secondaryHost = Core::Http::Url(secondaryUrl).GetHost();
-      clientOptions.RetryOptions.SecondaryHostForRetryReads = secondaryHost;
+      clientOptions.SecondaryHostForRetryReads = secondaryHost;
     }
     auto blobClient = Azure::Storage::Blobs::BlobClient::CreateFromConnectionString(
         StandardStorageConnectionString(), RandomString(), RandomString(), clientOptions);
diff --git a/sdk/storage/azure-storage-common/CMakeLists.txt b/sdk/storage/azure-storage-common/CMakeLists.txt
index f35763afa..c87f25adc 100644
--- a/sdk/storage/azure-storage-common/CMakeLists.txt
+++ b/sdk/storage/azure-storage-common/CMakeLists.txt
@@ -44,7 +44,7 @@ set(
     inc/azure/storage/common/storage_credential.hpp
     inc/azure/storage/common/storage_exception.hpp
     inc/azure/storage/common/storage_per_retry_policy.hpp
-    inc/azure/storage/common/storage_retry_policy.hpp
+    inc/azure/storage/common/storage_switch_to_secondary_policy.hpp
     inc/azure/storage/common/version.hpp
     inc/azure/storage/common/xml_wrapper.hpp
 )
@@ -60,7 +60,7 @@ set(
     src/storage_credential.cpp
     src/storage_exception.cpp
     src/storage_per_retry_policy.cpp
-    src/storage_retry_policy.cpp
+    src/storage_switch_to_secondary_policy.cpp
     src/xml_wrapper.cpp
 )
 
diff --git a/sdk/storage/azure-storage-common/inc/azure/storage/common/storage_common.hpp b/sdk/storage/azure-storage-common/inc/azure/storage/common/storage_common.hpp
index 3950b9c23..b32ffe92a 100644
--- a/sdk/storage/azure-storage-common/inc/azure/storage/common/storage_common.hpp
+++ b/sdk/storage/azure-storage-common/inc/azure/storage/common/storage_common.hpp
@@ -15,7 +15,6 @@
 
 #include "azure/storage/common/constants.hpp"
 #include "azure/storage/common/storage_per_retry_policy.hpp"
-#include "azure/storage/common/storage_retry_policy.hpp"
 
 namespace Azure { namespace Storage {
 
@@ -57,58 +56,4 @@ namespace Azure { namespace Storage {
   } // namespace Details
   using Metadata = Azure::Core::CaseInsensitiveMap;
 
-  namespace Details {
-
-    /*
-     * Policies order:
-     * Shared built-in per-operation policies
-     * Service-specific built-in per-operation policies
-     * Customer-defined per-operation policies
-     * Retry policy
-     * Shared built-in per-retry policies
-     * Service-specific built-in per-retry policies
-     * Customer-defined per-retry policies
-     * Authentication policy
-     * Transport policy
-     */
-
-    template <class T>
-    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> ConstructPolicies(
-        std::unique_ptr<Azure::Core::Http::HttpPolicy> serviceBuiltinPerOperationPolicy,
-        std::unique_ptr<Azure::Core::Http::HttpPolicy> authenticationPolicy,
-        T&& clientOptions)
-    {
-      std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> policies;
-      {
-        Azure::Core::Http::Internal::ValueOptions options;
-        options.HeaderValues[Details::HttpHeaderXMsVersion] = clientOptions.ApiVersion;
-        policies.emplace_back(std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(options));
-      }
-      policies.emplace_back(std::make_unique<Azure::Core::Http::RequestIdPolicy>());
-      if (serviceBuiltinPerOperationPolicy)
-      {
-        policies.emplace_back(std::move(serviceBuiltinPerOperationPolicy));
-      }
-      for (const auto& p : clientOptions.PerOperationPolicies)
-      {
-        policies.emplace_back(p->Clone());
-      }
-      policies.emplace_back(std::make_unique<Storage::Details::StorageRetryPolicy>(
-          std::forward<T>(clientOptions).RetryOptions));
-      policies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
-      for (const auto& p : clientOptions.PerRetryPolicies)
-      {
-        policies.emplace_back(p->Clone());
-      }
-      if (authenticationPolicy)
-      {
-        policies.emplace_back(std::move(authenticationPolicy));
-      }
-      policies.emplace_back(std::make_unique<Azure::Core::Http::TransportPolicy>(
-          std::forward<T>(clientOptions).TransportOptions));
-      return policies;
-    }
-
-  } // namespace Details
-
 }} // namespace Azure::Storage
diff --git a/sdk/storage/azure-storage-common/inc/azure/storage/common/storage_retry_policy.hpp b/sdk/storage/azure-storage-common/inc/azure/storage/common/storage_retry_policy.hpp
deleted file mode 100644
index 80a31f940..000000000
--- a/sdk/storage/azure-storage-common/inc/azure/storage/common/storage_retry_policy.hpp
+++ /dev/null
@@ -1,62 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// SPDX-License-Identifier: MIT
-
-#pragma once
-
-#include <memory>
-#include <string>
-
-#include <azure/core/http/policy.hpp>
-
-namespace Azure { namespace Storage {
-
-  /**
-   * StorageRetryWithSecondaryOptions configures whether the retry policy should retry a read
-   * operation against another host.
-   */
-  struct StorageRetryWithSecondaryOptions : public Core::Http::RetryOptions
-  {
-    /**
-     * SecondaryHostForRetryReads specifies whether the retry policy should retry a read
-     * operation against another host. If SecondaryHostForRetryReads is "" (the default) then
-     * operations are not retried against another host. NOTE: Before setting this field, make sure
-     * you understand the issues around reading stale & potentially-inconsistent data at this
-     * webpage: https://docs.microsoft.com/en-us/azure/storage/common/geo-redundant-design.
-     */
-    std::string SecondaryHostForRetryReads;
-  };
-
-  namespace Details {
-
-    class StorageRetryPolicy : public Azure::Core::Http::HttpPolicy {
-    public:
-      explicit StorageRetryPolicy(const Core::Http::RetryOptions& options)
-      {
-        m_options.MaxRetries = options.MaxRetries;
-        m_options.RetryDelay = options.RetryDelay;
-        m_options.MaxRetryDelay = options.MaxRetryDelay;
-        m_options.StatusCodes = options.StatusCodes;
-      }
-
-      explicit StorageRetryPolicy(const StorageRetryWithSecondaryOptions& options)
-          : m_options(options)
-      {
-      }
-
-      std::unique_ptr<Azure::Core::Http::HttpPolicy> Clone() const override
-      {
-        return std::make_unique<StorageRetryPolicy>(*this);
-      }
-
-      std::unique_ptr<Azure::Core::Http::RawResponse> Send(
-          Azure::Core::Http::Request& request,
-          Azure::Core::Http::NextHttpPolicy nextHttpPolicy,
-          const Azure::Core::Context& ctx) const override;
-
-    private:
-      StorageRetryWithSecondaryOptions m_options;
-    };
-
-  } // namespace Details
-
-}} // namespace Azure::Storage
diff --git a/sdk/storage/azure-storage-common/inc/azure/storage/common/storage_switch_to_secondary_policy.hpp b/sdk/storage/azure-storage-common/inc/azure/storage/common/storage_switch_to_secondary_policy.hpp
new file mode 100644
index 000000000..3c69a63fd
--- /dev/null
+++ b/sdk/storage/azure-storage-common/inc/azure/storage/common/storage_switch_to_secondary_policy.hpp
@@ -0,0 +1,49 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include <azure/core/http/policy.hpp>
+
+namespace Azure { namespace Storage { namespace Details {
+
+  static constexpr const char* SecondaryHostReplicaStatusKey
+      = "AzureSdkStorageSecondaryHostReplicaStatusKey";
+
+  struct SecondaryHostReplicaStatus : public Azure::Core::ValueBase
+  {
+    bool replicated = true;
+  };
+
+  inline Azure::Core::Context WithReplicaStatus(const Azure::Core::Context& context)
+  {
+    return context.WithValue(
+        SecondaryHostReplicaStatusKey, std::make_unique<SecondaryHostReplicaStatus>());
+  }
+
+  class StorageSwitchToSecondaryPolicy : public Azure::Core::Http::HttpPolicy {
+  public:
+    explicit StorageSwitchToSecondaryPolicy(std::string primaryHost, std::string secondaryHost)
+        : m_primaryHost(std::move(primaryHost)), m_secondaryHost(std::move(secondaryHost))
+    {
+    }
+
+    std::unique_ptr<Azure::Core::Http::HttpPolicy> Clone() const override
+    {
+      return std::make_unique<StorageSwitchToSecondaryPolicy>(*this);
+    }
+
+    std::unique_ptr<Azure::Core::Http::RawResponse> Send(
+        Azure::Core::Http::Request& request,
+        Azure::Core::Http::NextHttpPolicy nextHttpPolicy,
+        const Azure::Core::Context& ctx) const override;
+
+  private:
+    std::string m_primaryHost;
+    std::string m_secondaryHost;
+  };
+
+}}} // namespace Azure::Storage::Details
diff --git a/sdk/storage/azure-storage-common/src/storage_retry_policy.cpp b/sdk/storage/azure-storage-common/src/storage_retry_policy.cpp
deleted file mode 100644
index 5920581ed..000000000
--- a/sdk/storage/azure-storage-common/src/storage_retry_policy.cpp
+++ /dev/null
@@ -1,134 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// SPDX-License-Identifier: MIT
-
-#include "azure/storage/common/storage_retry_policy.hpp"
-
-#include <thread>
-
-#include "azure/storage/common/constants.hpp"
-
-namespace Azure { namespace Storage { namespace Details {
-
-  std::unique_ptr<Azure::Core::Http::RawResponse> StorageRetryPolicy::Send(
-      Azure::Core::Http::Request& request,
-      Azure::Core::Http::NextHttpPolicy nextHttpPolicy,
-      const Azure::Core::Context& ctx) const
-  {
-    bool considerSecondary = (request.GetMethod() == Azure::Core::Http::HttpMethod::Get
-                              || request.GetMethod() == Azure::Core::Http::HttpMethod::Head)
-        && !m_options.SecondaryHostForRetryReads.empty();
-
-    std::string primaryHost = request.GetUrl().GetHost();
-    const std::string& secondaryHost = m_options.SecondaryHostForRetryReads;
-
-    bool isUsingSecondary = false;
-
-    auto switchHost
-        = [&considerSecondary, &isUsingSecondary, &request, &primaryHost, &secondaryHost]() {
-            if (considerSecondary)
-            {
-              isUsingSecondary = !isUsingSecondary;
-            }
-            else
-            {
-              isUsingSecondary = false;
-            }
-            if (isUsingSecondary)
-            {
-              request.GetUrl().SetHost(secondaryHost);
-            }
-            else
-            {
-              request.GetUrl().SetHost(primaryHost);
-            }
-          };
-
-    std::unique_ptr<Azure::Core::Http::RawResponse> pResponse;
-    for (int i = 0; i <= m_options.MaxRetries; ++i)
-    {
-      bool lastAttempt = i == m_options.MaxRetries;
-      try
-      {
-        auto response = nextHttpPolicy.Send(request, ctx);
-
-        bool shouldRetry = false;
-
-        if (isUsingSecondary)
-        {
-          if (response->GetStatusCode() == Azure::Core::Http::HttpStatusCode::NotFound
-              || response->GetStatusCode() == Core::Http::HttpStatusCode::PreconditionFailed)
-          {
-            considerSecondary = false;
-            // disgard this response
-            shouldRetry = true;
-          }
-        }
-
-        shouldRetry |= response
-            && std::find(
-                   m_options.StatusCodes.begin(),
-                   m_options.StatusCodes.end(),
-                   response->GetStatusCode())
-                != m_options.StatusCodes.end();
-
-        pResponse = std::move(response);
-
-        if (!shouldRetry)
-        {
-          break;
-        }
-      }
-      catch (Azure::Core::RequestFailedException const&)
-      {
-        if (lastAttempt)
-        {
-          throw;
-        }
-      }
-
-      if (!lastAttempt)
-      {
-        if (auto bodyStream = request.GetBodyStream())
-        {
-          bodyStream->Rewind();
-        }
-
-        switchHost();
-
-        const int64_t baseRetryDelayMs = m_options.RetryDelay.count();
-        const int64_t maxRetryDelayMs = m_options.MaxRetryDelay.count();
-        int64_t retryDelayMs = maxRetryDelayMs;
-        if (static_cast<std::size_t>(i) < sizeof(int64_t) * 8)
-        {
-          const int64_t factor = 1LL << i;
-          retryDelayMs = baseRetryDelayMs * factor;
-          if (baseRetryDelayMs != 0 && retryDelayMs / baseRetryDelayMs != factor)
-          {
-            retryDelayMs = maxRetryDelayMs;
-          }
-          else
-          {
-            static thread_local std::random_device rd;
-            std::mt19937_64 gen(rd());
-            std::uniform_real_distribution<> dist(0.8, 1.3);
-
-            retryDelayMs = static_cast<decltype(retryDelayMs)>(retryDelayMs * dist(gen));
-            if (retryDelayMs < 0 || retryDelayMs > maxRetryDelayMs)
-            {
-              retryDelayMs = maxRetryDelayMs;
-            }
-          }
-        }
-
-        if (retryDelayMs != 0)
-        {
-          ctx.ThrowIfCancelled();
-          std::this_thread::sleep_for(std::chrono::milliseconds(retryDelayMs));
-        }
-      }
-    }
-
-    return pResponse;
-  }
-
-}}} // namespace Azure::Storage::Details
diff --git a/sdk/storage/azure-storage-common/src/storage_switch_to_secondary_policy.cpp b/sdk/storage/azure-storage-common/src/storage_switch_to_secondary_policy.cpp
new file mode 100644
index 000000000..a0c0565da
--- /dev/null
+++ b/sdk/storage/azure-storage-common/src/storage_switch_to_secondary_policy.cpp
@@ -0,0 +1,53 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+#include <azure/storage/common/storage_switch_to_secondary_policy.hpp>
+
+namespace Azure { namespace Storage { namespace Details {
+
+  std::unique_ptr<Azure::Core::Http::RawResponse> StorageSwitchToSecondaryPolicy::Send(
+      Azure::Core::Http::Request& request,
+      Azure::Core::Http::NextHttpPolicy nextHttpPolicy,
+      const Azure::Core::Context& ctx) const
+  {
+    SecondaryHostReplicaStatus* replicaStatus = nullptr;
+    if (ctx.HasKey(SecondaryHostReplicaStatusKey))
+    {
+      replicaStatus = dynamic_cast<SecondaryHostReplicaStatus*>(
+          ctx[SecondaryHostReplicaStatusKey].Get<std::unique_ptr<Azure::Core::ValueBase>>().get());
+    }
+
+    bool considerSecondary = (request.GetMethod() == Azure::Core::Http::HttpMethod::Get
+                              || request.GetMethod() == Azure::Core::Http::HttpMethod::Head)
+        && !m_secondaryHost.empty() && replicaStatus && replicaStatus->replicated;
+
+    if (considerSecondary && Azure::Core::Http::RetryPolicy::GetRetryNumber(ctx) > 0)
+    {
+      // switch host
+      if (request.GetUrl().GetHost() == m_primaryHost)
+      {
+        request.GetUrl().SetHost(m_secondaryHost);
+      }
+      else
+      {
+        request.GetUrl().SetHost(m_primaryHost);
+      }
+    }
+
+    auto response = nextHttpPolicy.Send(request, ctx);
+
+    if (considerSecondary
+        && (response->GetStatusCode() == Azure::Core::Http::HttpStatusCode::NotFound
+            || response->GetStatusCode() == Core::Http::HttpStatusCode::PreconditionFailed)
+        && request.GetUrl().GetHost() == m_secondaryHost)
+    {
+      replicaStatus->replicated = false;
+      // switch back
+      request.GetUrl().SetHost(m_primaryHost);
+      response = nextHttpPolicy.Send(request, ctx);
+    }
+
+    return response;
+  }
+
+}}} // namespace Azure::Storage::Details
diff --git a/sdk/storage/azure-storage-files-datalake/CHANGELOG.md b/sdk/storage/azure-storage-files-datalake/CHANGELOG.md
index 9f6c2e3b9..f8742f1b8 100644
--- a/sdk/storage/azure-storage-files-datalake/CHANGELOG.md
+++ b/sdk/storage/azure-storage-files-datalake/CHANGELOG.md
@@ -4,12 +4,13 @@
 
 ### New Features
 
-- Added support for customized application ID.
+- Added support for telemetry options.
 
 ### Breaking Changes
 
 - DataLake client constructors won't automatically convert blob url to dfs url anymore.
 - String conversion functions of extensible enums were renamed from `Get()` to `ToString()`.
+- Moved `SecondaryHostForRetryReads` out of retry options, now it's under `DataLakeClientOptions`.
 
 ## 12.0.0-beta.8 (2021-02-12)
 
diff --git a/sdk/storage/azure-storage-files-datalake/inc/azure/storage/files/datalake/datalake_options.hpp b/sdk/storage/azure-storage-files-datalake/inc/azure/storage/files/datalake/datalake_options.hpp
index 5ec033138..d29db265f 100644
--- a/sdk/storage/azure-storage-files-datalake/inc/azure/storage/files/datalake/datalake_options.hpp
+++ b/sdk/storage/azure-storage-files-datalake/inc/azure/storage/files/datalake/datalake_options.hpp
@@ -24,34 +24,16 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
    * @brief Client options used to initalize DataLakeServiceClient, FileSystemClient, PathClient,
    * FileClient and DirectoryClient.
    */
-  struct DataLakeClientOptions
+  struct DataLakeClientOptions : Azure::Core::Internal::ClientOptions
   {
     /**
-     * @brief Transport pipeline policies for authentication, additional HTTP headers, etc., that
-     * are applied to every request.
+     * SecondaryHostForRetryReads specifies whether the retry policy should retry a read
+     * operation against another host. If SecondaryHostForRetryReads is "" (the default) then
+     * operations are not retried against another host. NOTE: Before setting this field, make sure
+     * you understand the issues around reading stale & potentially-inconsistent data at this
+     * webpage: https://docs.microsoft.com/en-us/azure/storage/common/geo-redundant-design.
      */
-    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> PerOperationPolicies;
-
-    /**
-     * @brief Transport pipeline policies for authentication, additional HTTP headers, etc., that
-     * are applied to every retrial.
-     */
-    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> PerRetryPolicies;
-
-    /**
-     * @brief Specify the number of retries and other retry-related options.
-     */
-    StorageRetryWithSecondaryOptions RetryOptions;
-
-    /**
-     * @brief Customized HTTP client. We're going to use the default one if this is empty.
-     */
-    Azure::Core::Http::TransportOptions TransportOptions;
-
-    /**
-     * @brief The last part of the user agent for telemetry.
-     */
-    std::string ApplicationId;
+    std::string SecondaryHostForRetryReads;
 
     /**
      * API version used by this client.
diff --git a/sdk/storage/azure-storage-files-datalake/src/datalake_directory_client.cpp b/sdk/storage/azure-storage-files-datalake/src/datalake_directory_client.cpp
index 92d10ef3b..d56a18ab8 100644
--- a/sdk/storage/azure-storage-files-datalake/src/datalake_directory_client.cpp
+++ b/sdk/storage/azure-storage-files-datalake/src/datalake_directory_client.cpp
@@ -8,6 +8,7 @@
 #include <azure/storage/common/crypt.hpp>
 #include <azure/storage/common/shared_key_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
+#include <azure/storage/common/storage_switch_to_secondary_policy.hpp>
 
 #include "azure/storage/files/datalake/datalake_file_client.hpp"
 #include "azure/storage/files/datalake/datalake_utilities.hpp"
@@ -209,7 +210,10 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
     if (firstSlashPos == 0 || (firstSlashPos == currentPath.size() + 1U))
     {
       return Details::DataLakeRestClient::FileSystem::ListPaths(
-          m_pathUrl, *m_pipeline, context, protocolLayerOptions);
+          m_pathUrl,
+          *m_pipeline,
+          Storage::Details::WithReplicaStatus(context),
+          protocolLayerOptions);
     }
     else
     {
@@ -219,7 +223,10 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
       fileSystemUrl.SetPath(currentPath.substr(
           0U, currentPath.size() - protocolLayerOptions.Directory.GetValue().size() - 1U));
       return Details::DataLakeRestClient::FileSystem::ListPaths(
-          fileSystemUrl, *m_pipeline, context, protocolLayerOptions);
+          fileSystemUrl,
+          *m_pipeline,
+          Storage::Details::WithReplicaStatus(context),
+          protocolLayerOptions);
     }
   }
 
diff --git a/sdk/storage/azure-storage-files-datalake/src/datalake_file_system_client.cpp b/sdk/storage/azure-storage-files-datalake/src/datalake_file_system_client.cpp
index 3f47c1cc0..4a96b5e60 100644
--- a/sdk/storage/azure-storage-files-datalake/src/datalake_file_system_client.cpp
+++ b/sdk/storage/azure-storage-files-datalake/src/datalake_file_system_client.cpp
@@ -9,6 +9,8 @@
 #include <azure/storage/common/crypt.hpp>
 #include <azure/storage/common/shared_key_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
+#include <azure/storage/common/storage_per_retry_policy.hpp>
+#include <azure/storage/common/storage_switch_to_secondary_policy.hpp>
 
 #include "azure/storage/files/datalake/datalake_constants.hpp"
 #include "azure/storage/files/datalake/datalake_directory_client.hpp"
@@ -48,16 +50,28 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
                                             credential,
                                             Details::GetBlobClientOptions(options))
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    DataLakeClientOptions newOptions = options;
+    newOptions.PerRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::SharedKeyPolicy>(credential));
+
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_fileSystemUrl.GetHost(), newOptions.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = newOptions.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Storage::Details::SharedKeyPolicy>(credential),
-            options));
+        newOptions,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   DataLakeFileSystemClient::DataLakeFileSystemClient(
@@ -69,19 +83,31 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
                                             credential,
                                             Details::GetBlobClientOptions(options))
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
-    Azure::Core::Http::TokenRequestOptions tokenOptions;
-    tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_fileSystemUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::TokenRequestOptions tokenOptions;
+      tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+      perRetryPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
+              credential, tokenOptions));
+    }
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
-                credential, tokenOptions),
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   DataLakeFileSystemClient::DataLakeFileSystemClient(
@@ -91,16 +117,24 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
                                             Details::GetBlobUrlFromUrl(fileSystemUrl),
                                             Details::GetBlobClientOptions(options))
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_fileSystemUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            nullptr,
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   DataLakeFileClient DataLakeFileSystemClient::GetFileClient(const std::string& fileName) const
@@ -263,7 +297,10 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
     protocolLayerOptions.MaxResults = options.PageSizeHint;
     protocolLayerOptions.RecursiveRequired = recursive;
     return Details::DataLakeRestClient::FileSystem::ListPaths(
-        m_fileSystemUrl, *m_pipeline, context, protocolLayerOptions);
+        m_fileSystemUrl,
+        *m_pipeline,
+        Storage::Details::WithReplicaStatus(context),
+        protocolLayerOptions);
   }
 
   Azure::Core::Response<Models::GetDataLakeFileSystemAccessPolicyResult>
diff --git a/sdk/storage/azure-storage-files-datalake/src/datalake_path_client.cpp b/sdk/storage/azure-storage-files-datalake/src/datalake_path_client.cpp
index 2b4d2932d..aed7a7ed3 100644
--- a/sdk/storage/azure-storage-files-datalake/src/datalake_path_client.cpp
+++ b/sdk/storage/azure-storage-files-datalake/src/datalake_path_client.cpp
@@ -8,6 +8,8 @@
 #include <azure/storage/common/crypt.hpp>
 #include <azure/storage/common/shared_key_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
+#include <azure/storage/common/storage_per_retry_policy.hpp>
+#include <azure/storage/common/storage_switch_to_secondary_policy.hpp>
 
 #include "azure/storage/files/datalake/datalake_constants.hpp"
 #include "azure/storage/files/datalake/datalake_utilities.hpp"
@@ -86,16 +88,28 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
                                 credential,
                                 Details::GetBlobClientOptions(options))
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    DataLakeClientOptions newOptions = options;
+    newOptions.PerRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::SharedKeyPolicy>(credential));
+
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_pathUrl.GetHost(), newOptions.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = newOptions.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Storage::Details::SharedKeyPolicy>(credential),
-            options));
+        newOptions,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   DataLakePathClient::DataLakePathClient(
@@ -107,19 +121,31 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
                                 credential,
                                 Details::GetBlobClientOptions(options))
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
-    Azure::Core::Http::TokenRequestOptions tokenOptions;
-    tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_pathUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::TokenRequestOptions tokenOptions;
+      tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+      perRetryPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
+              credential, tokenOptions));
+    }
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
-                credential, tokenOptions),
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   DataLakePathClient::DataLakePathClient(
@@ -128,16 +154,24 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
       : m_pathUrl(pathUrl),
         m_blobClient(Details::GetBlobUrlFromUrl(pathUrl), Details::GetBlobClientOptions(options))
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_pathUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            nullptr,
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   Azure::Core::Response<Models::SetDataLakePathAccessControlListResult>
@@ -371,7 +405,7 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
     protocolLayerOptions.IfModifiedSince = options.AccessConditions.IfModifiedSince;
     protocolLayerOptions.IfUnmodifiedSince = options.AccessConditions.IfUnmodifiedSince;
     auto result = Details::DataLakeRestClient::Path::GetProperties(
-        m_pathUrl, *m_pipeline, context, protocolLayerOptions);
+        m_pathUrl, *m_pipeline, Storage::Details::WithReplicaStatus(context), protocolLayerOptions);
     Azure::Core::Nullable<std::vector<Models::Acl>> acl;
     if (result->Acl.HasValue())
     {
diff --git a/sdk/storage/azure-storage-files-datalake/src/datalake_service_client.cpp b/sdk/storage/azure-storage-files-datalake/src/datalake_service_client.cpp
index 2e886d4ab..4424cfd98 100644
--- a/sdk/storage/azure-storage-files-datalake/src/datalake_service_client.cpp
+++ b/sdk/storage/azure-storage-files-datalake/src/datalake_service_client.cpp
@@ -9,6 +9,8 @@
 #include <azure/storage/common/shared_key_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
 #include <azure/storage/common/storage_credential.hpp>
+#include <azure/storage/common/storage_per_retry_policy.hpp>
+#include <azure/storage/common/storage_switch_to_secondary_policy.hpp>
 
 #include "azure/storage/files/datalake/datalake_file_system_client.hpp"
 #include "azure/storage/files/datalake/datalake_utilities.hpp"
@@ -89,16 +91,28 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
                                       credential,
                                       Details::GetBlobClientOptions(options))
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    DataLakeClientOptions newOptions = options;
+    newOptions.PerRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::SharedKeyPolicy>(credential));
+
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_serviceUrl.GetHost(), newOptions.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = newOptions.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Storage::Details::SharedKeyPolicy>(credential),
-            options));
+        newOptions,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   DataLakeServiceClient::DataLakeServiceClient(
@@ -110,19 +124,31 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
                                       credential,
                                       Details::GetBlobClientOptions(options))
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
-    Azure::Core::Http::TokenRequestOptions tokenOptions;
-    tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_serviceUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::TokenRequestOptions tokenOptions;
+      tokenOptions.Scopes.emplace_back(Storage::Details::StorageScope);
+      perRetryPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
+              credential, tokenOptions));
+    }
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Azure::Core::Http::BearerTokenAuthenticationPolicy>(
-                credential, tokenOptions),
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   DataLakeServiceClient::DataLakeServiceClient(
@@ -132,16 +158,24 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake {
                                       Details::GetBlobUrlFromUrl(serviceUrl),
                                       Details::GetBlobClientOptions(options))
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::StorageSwitchToSecondaryPolicy>(
+            m_serviceUrl.GetHost(), options.SecondaryHostForRetryReads));
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            nullptr,
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   DataLakeFileSystemClient DataLakeServiceClient::GetFileSystemClient(
diff --git a/sdk/storage/azure-storage-files-datalake/src/datalake_utilities.cpp b/sdk/storage/azure-storage-files-datalake/src/datalake_utilities.cpp
index 338d3f526..ac35f8e4d 100644
--- a/sdk/storage/azure-storage-files-datalake/src/datalake_utilities.cpp
+++ b/sdk/storage/azure-storage-files-datalake/src/datalake_utilities.cpp
@@ -76,19 +76,9 @@ namespace Azure { namespace Storage { namespace Files { namespace DataLake { nam
   Blobs::BlobClientOptions GetBlobClientOptions(const DataLakeClientOptions& options)
   {
     Blobs::BlobClientOptions blobOptions;
-    for (const auto& p : options.PerOperationPolicies)
-    {
-      blobOptions.PerOperationPolicies.emplace_back(p->Clone());
-    }
-    for (const auto& p : options.PerRetryPolicies)
-    {
-      blobOptions.PerRetryPolicies.emplace_back(p->Clone());
-    }
-    blobOptions.RetryOptions = options.RetryOptions;
-    blobOptions.RetryOptions.SecondaryHostForRetryReads
-        = Details::GetBlobUrlFromUrl(options.RetryOptions.SecondaryHostForRetryReads);
-    blobOptions.TransportOptions = options.TransportOptions;
-    blobOptions.ApplicationId = options.ApplicationId;
+    *(static_cast<Azure::Core::Internal::ClientOptions*>(&blobOptions)) = options;
+    blobOptions.SecondaryHostForRetryReads
+        = Details::GetBlobUrlFromUrl(options.SecondaryHostForRetryReads);
     blobOptions.ApiVersion = options.ApiVersion;
     return blobOptions;
   }
diff --git a/sdk/storage/azure-storage-files-shares/CHANGELOG.md b/sdk/storage/azure-storage-files-shares/CHANGELOG.md
index c0770ae5e..e999fe43e 100644
--- a/sdk/storage/azure-storage-files-shares/CHANGELOG.md
+++ b/sdk/storage/azure-storage-files-shares/CHANGELOG.md
@@ -4,7 +4,7 @@
 
 ### New Features
 
-- Added support for customized application ID.
+- Added support for telemetry options.
 
 ### Breaking Changes
 
diff --git a/sdk/storage/azure-storage-files-shares/inc/azure/storage/files/shares/share_options.hpp b/sdk/storage/azure-storage-files-shares/inc/azure/storage/files/shares/share_options.hpp
index f29dec6a1..ba59160fe 100644
--- a/sdk/storage/azure-storage-files-shares/inc/azure/storage/files/shares/share_options.hpp
+++ b/sdk/storage/azure-storage-files-shares/inc/azure/storage/files/shares/share_options.hpp
@@ -7,9 +7,9 @@
 #include <string>
 #include <vector>
 
+#include <azure/core/internal/client_options.hpp>
 #include <azure/core/nullable.hpp>
 #include <azure/storage/common/access_conditions.hpp>
-#include <azure/storage/common/storage_retry_policy.hpp>
 
 #include "azure/storage/files/shares/protocol/share_rest_client.hpp"
 #include "azure/storage/files/shares/share_responses.hpp"
@@ -20,35 +20,8 @@ namespace Azure { namespace Storage { namespace Files { namespace Shares {
    * @brief Client options used to initalize ShareServiceClient, ShareClient, ShareFileClient and
    * ShareDirectoryClient.
    */
-  struct ShareClientOptions
+  struct ShareClientOptions : Azure::Core::Internal::ClientOptions
   {
-    /**
-     * @brief Transport pipeline policies for authentication, additional HTTP headers, etc., that
-     * are applied to every request.
-     */
-    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> PerOperationPolicies;
-
-    /**
-     * @brief Transport pipeline policies for authentication, additional HTTP headers, etc., that
-     * are applied to every retrial.
-     */
-    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> PerRetryPolicies;
-
-    /**
-     * @brief Specify the number of retries and other retry-related options.
-     */
-    Core::Http::RetryOptions RetryOptions;
-
-    /**
-     * @brief Customized HTTP client. We're going to use the default one if this is empty.
-     */
-    Azure::Core::Http::TransportOptions TransportOptions;
-
-    /**
-     * @brief The last part of the user agent for telemetry.
-     */
-    std::string ApplicationId;
-
     /**
      * API version used by this client.
      */
diff --git a/sdk/storage/azure-storage-files-shares/src/share_client.cpp b/sdk/storage/azure-storage-files-shares/src/share_client.cpp
index 815ee91f9..1c79d7895 100644
--- a/sdk/storage/azure-storage-files-shares/src/share_client.cpp
+++ b/sdk/storage/azure-storage-files-shares/src/share_client.cpp
@@ -9,6 +9,7 @@
 #include <azure/storage/common/crypt.hpp>
 #include <azure/storage/common/shared_key_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
+#include <azure/storage/common/storage_per_retry_policy.hpp>
 
 #include "azure/storage/files/shares/share_directory_client.hpp"
 #include "azure/storage/files/shares/share_file_client.hpp"
@@ -41,31 +42,45 @@ namespace Azure { namespace Storage { namespace Files { namespace Shares {
       const ShareClientOptions& options)
       : m_shareUrl(shareUrl)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    ShareClientOptions newOptions = options;
+    newOptions.PerRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::SharedKeyPolicy>(credential));
+
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = newOptions.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Storage::Details::SharedKeyPolicy>(credential),
-            options));
+        newOptions,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   ShareClient::ShareClient(const std::string& shareUrl, const ShareClientOptions& options)
       : m_shareUrl(shareUrl)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            nullptr,
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   ShareDirectoryClient ShareClient::GetRootDirectoryClient() const
diff --git a/sdk/storage/azure-storage-files-shares/src/share_directory_client.cpp b/sdk/storage/azure-storage-files-shares/src/share_directory_client.cpp
index c3528e289..f3c86509b 100644
--- a/sdk/storage/azure-storage-files-shares/src/share_directory_client.cpp
+++ b/sdk/storage/azure-storage-files-shares/src/share_directory_client.cpp
@@ -9,6 +9,7 @@
 #include <azure/storage/common/crypt.hpp>
 #include <azure/storage/common/shared_key_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
+#include <azure/storage/common/storage_per_retry_policy.hpp>
 
 #include "azure/storage/files/shares/share_file_client.hpp"
 #include "azure/storage/files/shares/version.hpp"
@@ -43,16 +44,25 @@ namespace Azure { namespace Storage { namespace Files { namespace Shares {
       const ShareClientOptions& options)
       : m_shareDirectoryUrl(shareDirectoryUrl)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    ShareClientOptions newOptions = options;
+    newOptions.PerRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::SharedKeyPolicy>(credential));
+
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = newOptions.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Storage::Details::SharedKeyPolicy>(credential),
-            options));
+        newOptions,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   ShareDirectoryClient::ShareDirectoryClient(
@@ -60,16 +70,21 @@ namespace Azure { namespace Storage { namespace Files { namespace Shares {
       const ShareClientOptions& options)
       : m_shareDirectoryUrl(shareDirectoryUrl)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            nullptr,
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   ShareDirectoryClient ShareDirectoryClient::GetSubdirectoryClient(
diff --git a/sdk/storage/azure-storage-files-shares/src/share_file_client.cpp b/sdk/storage/azure-storage-files-shares/src/share_file_client.cpp
index d04c3b890..8bd20de46 100644
--- a/sdk/storage/azure-storage-files-shares/src/share_file_client.cpp
+++ b/sdk/storage/azure-storage-files-shares/src/share_file_client.cpp
@@ -13,6 +13,7 @@
 #include <azure/storage/common/reliable_stream.hpp>
 #include <azure/storage/common/shared_key_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
+#include <azure/storage/common/storage_per_retry_policy.hpp>
 
 #include "azure/storage/files/shares/share_constants.hpp"
 #include "azure/storage/files/shares/version.hpp"
@@ -47,16 +48,25 @@ namespace Azure { namespace Storage { namespace Files { namespace Shares {
       const ShareClientOptions& options)
       : m_shareFileUrl(shareFileUrl)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    ShareClientOptions newOptions = options;
+    newOptions.PerRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::SharedKeyPolicy>(credential));
+
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = newOptions.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Storage::Details::SharedKeyPolicy>(credential),
-            options));
+        newOptions,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   ShareFileClient::ShareFileClient(
@@ -64,16 +74,21 @@ namespace Azure { namespace Storage { namespace Files { namespace Shares {
       const ShareClientOptions& options)
       : m_shareFileUrl(shareFileUrl)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            nullptr,
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   ShareFileClient ShareFileClient::WithShareSnapshot(const std::string& shareSnapshot) const
diff --git a/sdk/storage/azure-storage-files-shares/src/share_service_client.cpp b/sdk/storage/azure-storage-files-shares/src/share_service_client.cpp
index f28e6a64f..cc1601d25 100644
--- a/sdk/storage/azure-storage-files-shares/src/share_service_client.cpp
+++ b/sdk/storage/azure-storage-files-shares/src/share_service_client.cpp
@@ -9,6 +9,7 @@
 #include <azure/storage/common/shared_key_policy.hpp>
 #include <azure/storage/common/storage_common.hpp>
 #include <azure/storage/common/storage_credential.hpp>
+#include <azure/storage/common/storage_per_retry_policy.hpp>
 
 #include "azure/storage/files/shares/share_client.hpp"
 #include "azure/storage/files/shares/version.hpp"
@@ -38,16 +39,25 @@ namespace Azure { namespace Storage { namespace Files { namespace Shares {
       const ShareClientOptions& options)
       : m_serviceUrl(serviceUrl)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    ShareClientOptions newOptions = options;
+    newOptions.PerRetryPolicies.emplace_back(
+        std::make_unique<Storage::Details::SharedKeyPolicy>(credential));
+
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = newOptions.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            std::make_unique<Storage::Details::SharedKeyPolicy>(credential),
-            options));
+        newOptions,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   ShareServiceClient::ShareServiceClient(
@@ -55,16 +65,21 @@ namespace Azure { namespace Storage { namespace Files { namespace Shares {
       const ShareClientOptions& options)
       : m_serviceUrl(serviceUrl)
   {
-    Azure::Core::Http::TelemetryOptions telemetryPolicyOptions;
-    telemetryPolicyOptions.ApplicationId = options.ApplicationId;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perRetryPolicies;
+    std::vector<std::unique_ptr<Azure::Core::Http::HttpPolicy>> perOperationPolicies;
+    perRetryPolicies.emplace_back(std::make_unique<Storage::Details::StoragePerRetryPolicy>());
+    {
+      Azure::Core::Http::Internal::ValueOptions valueOptions;
+      valueOptions.HeaderValues[Storage::Details::HttpHeaderXMsVersion] = options.ApiVersion;
+      perOperationPolicies.emplace_back(
+          std::make_unique<Azure::Core::Http::Internal::ValuePolicy>(valueOptions));
+    }
     m_pipeline = std::make_shared<Azure::Core::Internal::Http::HttpPipeline>(
-        Storage::Details::ConstructPolicies(
-            std::make_unique<Azure::Core::Http::TelemetryPolicy>(
-                Storage::Details::BlobServicePackageName,
-                Details::Version::VersionString(),
-                telemetryPolicyOptions),
-            nullptr,
-            options));
+        options,
+        Storage::Details::FileServicePackageName,
+        Details::Version::VersionString(),
+        std::move(perRetryPolicies),
+        std::move(perOperationPolicies));
   }
 
   ShareClient ShareServiceClient::GetShareClient(const std::string& shareName) const