diff --git a/api/envoy/config/route/v3/route_components.proto b/api/envoy/config/route/v3/route_components.proto
index 4a0b93f3601c2..ec27e627d346d 100644
--- a/api/envoy/config/route/v3/route_components.proto
+++ b/api/envoy/config/route/v3/route_components.proto
@@ -1350,6 +1350,11 @@ message RetryPolicy {
   // details.
   repeated RetryHostPredicate retry_host_predicate = 5;
 
+  // Retry options predicates that will be applied prior to retrying a request. These predicates
+  // allow customizing request behavior between retries.
+  // [#comment: add [#extension-category: envoy.retry_options_predicates] when there are built-in extensions]
+  repeated core.v3.TypedExtensionConfig retry_options_predicates = 12;
+
   // The maximum number of times host selection will be reattempted before giving up, at which
   // point the host that was last selected will be routed to. If unspecified, this will default to
   // retrying once.
diff --git a/docs/root/version_history/current.rst b/docs/root/version_history/current.rst
index ec8a10f5475f1..e8e93a650db5e 100644
--- a/docs/root/version_history/current.rst
+++ b/docs/root/version_history/current.rst
@@ -115,6 +115,10 @@ New Features
 * overload: add a new overload action that resets streams using a lot of memory. To enable the tracking of allocated bytes in buffers that a stream is using  we need to configure the minimum threshold for tracking via:ref:`buffer_factory_config <envoy_v3_api_field_config.overload.v3.OverloadManager.buffer_factory_config>`. We have an overload action ``Envoy::Server::OverloadActionNameValues::ResetStreams`` that takes advantage of the tracking to  reset the most expensive stream first.
 * rbac: added :ref:`destination_port_range <envoy_v3_api_field_config.rbac.v3.Permission.destination_port_range>` for matching range of destination ports.
 * route config: added :ref:`dynamic_metadata <envoy_v3_api_field_config.route.v3.RouteMatch.dynamic_metadata>` for routing based on dynamic metadata.
+* router: added retry options predicate extensions configured via
+  :ref:` <envoy_v3_api_field_config.route.v3.RetryPolicy.retry_options_predicates>`. These
+  extensions allow modification of requests between retries at the router level. There are not
+  currently any built-in extensions that implement this extension point.
 * router: added :ref:`per_try_idle_timeout <envoy_v3_api_field_config.route.v3.RetryPolicy.per_try_idle_timeout>` timeout configuration.
 * router: added an optional :ref:`override_auto_sni_header <envoy_v3_api_field_config.core.v3.UpstreamHttpProtocolOptions.override_auto_sni_header>` to support setting SNI value from an arbitrary header other than host/authority.
 * sxg_filter: added filter to transform response to SXG package to :ref:`contrib images <install_contrib>`. This can be enabled by setting :ref:`SXG <envoy_v3_api_msg_extensions.filters.http.sxg.v3alpha.SXG>` configuration.
diff --git a/envoy/router/router.h b/envoy/router/router.h
index c4288f10c3d1d..b7d8f8db2f3a9 100644
--- a/envoy/router/router.h
+++ b/envoy/router/router.h
@@ -236,6 +236,13 @@ class RetryPolicy {
    */
   virtual Upstream::RetryPrioritySharedPtr retryPriority() const PURE;
 
+  /**
+   * @return the retry options predicates for this policy. Each policy will be applied prior
+   * to retrying a request, allowing for request behavior to be customized.
+   */
+  virtual absl::Span<const Upstream::RetryOptionsPredicateConstSharedPtr>
+  retryOptionsPredicates() const PURE;
+
   /**
    * Number of times host selection should be reattempted when selecting a host
    * for a retry attempt.
diff --git a/envoy/upstream/cluster_manager.h b/envoy/upstream/cluster_manager.h
index eeb249affd58b..e24790038d241 100644
--- a/envoy/upstream/cluster_manager.h
+++ b/envoy/upstream/cluster_manager.h
@@ -412,6 +412,11 @@ class ClusterManagerFactory {
    * Returns the secret manager.
    */
   virtual Secret::SecretManager& secretManager() PURE;
+
+  /**
+   * Returns the singleton manager.
+   */
+  virtual Singleton::Manager& singletonManager() PURE;
 };
 
 /**
diff --git a/envoy/upstream/retry.h b/envoy/upstream/retry.h
index f772d54029179..9e1a8de57995f 100644
--- a/envoy/upstream/retry.h
+++ b/envoy/upstream/retry.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include "envoy/config/typed_config.h"
+#include "envoy/singleton/manager.h"
 #include "envoy/upstream/types.h"
 #include "envoy/upstream/upstream.h"
 
@@ -92,13 +93,58 @@ class RetryHostPredicate {
 
 using RetryHostPredicateSharedPtr = std::shared_ptr<RetryHostPredicate>;
 
+/**
+ * A predicate that is applied prior to retrying a request. Each predicate can customize request
+ * behavior prior to the request being retried.
+ */
+class RetryOptionsPredicate {
+public:
+  struct UpdateOptionsParameters {
+    // Stream info for the previous request attempt that is about to be retried.
+    const StreamInfo::StreamInfo& retriable_request_stream_info_;
+    // The current upstream socket options that were used for connection pool selection on the
+    // previous attempt, or the result of an updated set of options from a previously run
+    // retry options predicate.
+    Network::Socket::OptionsSharedPtr current_upstream_socket_options_;
+  };
+
+  struct UpdateOptionsReturn {
+    // New upstream socket options to apply to the next request attempt. If changed, will affect
+    // connection pool selection similar to that which was done for the initial request.
+    absl::optional<Network::Socket::OptionsSharedPtr> new_upstream_socket_options_;
+  };
+
+  virtual ~RetryOptionsPredicate() = default;
+
+  /**
+   * Update request options.
+   * @param parameters supplies the update parameters.
+   * @return the new options to apply. Each option is wrapped in an optional and is only applied
+   *         if valid.
+   */
+  virtual UpdateOptionsReturn updateOptions(const UpdateOptionsParameters& parameters) const PURE;
+};
+
+using RetryOptionsPredicateConstSharedPtr = std::shared_ptr<const RetryOptionsPredicate>;
+
+/**
+ * Context for all retry extensions.
+ */
+class RetryExtensionFactoryContext {
+public:
+  virtual ~RetryExtensionFactoryContext() = default;
+
+  /**
+   * @return Singleton::Manager& the server-wide singleton manager.
+   */
+  virtual Singleton::Manager& singletonManager() PURE;
+};
+
 /**
  * Factory for RetryPriority.
  */
 class RetryPriorityFactory : public Config::TypedFactory {
 public:
-  ~RetryPriorityFactory() override = default;
-
   virtual RetryPrioritySharedPtr
   createRetryPriority(const Protobuf::Message& config,
                       ProtobufMessage::ValidationVisitor& validation_visitor,
@@ -112,13 +158,23 @@ class RetryPriorityFactory : public Config::TypedFactory {
  */
 class RetryHostPredicateFactory : public Config::TypedFactory {
 public:
-  ~RetryHostPredicateFactory() override = default;
-
   virtual RetryHostPredicateSharedPtr createHostPredicate(const Protobuf::Message& config,
                                                           uint32_t retry_count) PURE;
 
   std::string category() const override { return "envoy.retry_host_predicates"; }
 };
 
+/**
+ * Factory for RetryOptionsPredicate.
+ */
+class RetryOptionsPredicateFactory : public Config::TypedFactory {
+public:
+  virtual RetryOptionsPredicateConstSharedPtr
+  createOptionsPredicate(const Protobuf::Message& config,
+                         RetryExtensionFactoryContext& context) PURE;
+
+  std::string category() const override { return "envoy.retry_options_predicates"; }
+};
+
 } // namespace Upstream
 } // namespace Envoy
diff --git a/source/common/http/async_client_impl.cc b/source/common/http/async_client_impl.cc
index f913103750301..2c1da999f255b 100644
--- a/source/common/http/async_client_impl.cc
+++ b/source/common/http/async_client_impl.cc
@@ -42,7 +42,7 @@ AsyncClientImpl::AsyncClientImpl(Upstream::ClusterInfoConstSharedPtr cluster,
       config_(http_context.asyncClientStatPrefix(), local_info, stats_store, cm, runtime, random,
               std::move(shadow_writer), true, false, false, false, false, {},
               dispatcher.timeSource(), http_context, router_context),
-      dispatcher_(dispatcher) {}
+      dispatcher_(dispatcher), singleton_manager_(cm.clusterManagerFactory().singletonManager()) {}
 
 AsyncClientImpl::~AsyncClientImpl() {
   while (!active_streams_.empty()) {
@@ -81,8 +81,8 @@ AsyncStreamImpl::AsyncStreamImpl(AsyncClientImpl& parent, AsyncClient::StreamCal
       router_(parent.config_),
       stream_info_(Protocol::Http11, parent.dispatcher().timeSource(), nullptr),
       tracing_config_(Tracing::EgressConfig::get()),
-      route_(std::make_shared<RouteImpl>(parent_.cluster_->name(), options.timeout,
-                                         options.hash_policy, options.retry_policy)),
+      route_(std::make_shared<RouteImpl>(parent_, options.timeout, options.hash_policy,
+                                         options.retry_policy)),
       send_xff_(options.send_xff) {
 
   stream_info_.dynamicMetadata().MergeFrom(options.metadata);
diff --git a/source/common/http/async_client_impl.h b/source/common/http/async_client_impl.h
index 27ea7b3abea3d..608813cc01722 100644
--- a/source/common/http/async_client_impl.h
+++ b/source/common/http/async_client_impl.h
@@ -40,6 +40,7 @@
 #include "source/common/router/router.h"
 #include "source/common/stream_info/stream_info_impl.h"
 #include "source/common/tracing/http_tracer_impl.h"
+#include "source/common/upstream/retry_factory.h"
 
 namespace Envoy {
 namespace Http {
@@ -67,6 +68,7 @@ class AsyncClientImpl final : public AsyncClient {
   Router::FilterConfig config_;
   Event::Dispatcher& dispatcher_;
   std::list<std::unique_ptr<AsyncStreamImpl>> active_streams_;
+  Singleton::Manager& singleton_manager_;
 
   friend class AsyncStreamImpl;
   friend class AsyncRequestImpl;
@@ -124,48 +126,6 @@ class AsyncStreamImpl : public AsyncClient::Stream,
         rate_limit_policy_entry_;
   };
 
-  struct NullRetryPolicy : public Router::RetryPolicy {
-    // Router::RetryPolicy
-    std::chrono::milliseconds perTryTimeout() const override {
-      return std::chrono::milliseconds(0);
-    }
-    std::chrono::milliseconds perTryIdleTimeout() const override {
-      return std::chrono::milliseconds(0);
-    }
-    std::vector<Upstream::RetryHostPredicateSharedPtr> retryHostPredicates() const override {
-      return {};
-    }
-    Upstream::RetryPrioritySharedPtr retryPriority() const override { return {}; }
-
-    uint32_t hostSelectionMaxAttempts() const override { return 1; }
-    uint32_t numRetries() const override { return 1; }
-    uint32_t retryOn() const override { return 0; }
-    const std::vector<uint32_t>& retriableStatusCodes() const override {
-      return retriable_status_codes_;
-    }
-    const std::vector<Http::HeaderMatcherSharedPtr>& retriableHeaders() const override {
-      return retriable_headers_;
-    }
-    const std::vector<Http::HeaderMatcherSharedPtr>& retriableRequestHeaders() const override {
-      return retriable_request_headers_;
-    }
-    absl::optional<std::chrono::milliseconds> baseInterval() const override {
-      return absl::nullopt;
-    }
-    absl::optional<std::chrono::milliseconds> maxInterval() const override { return absl::nullopt; }
-    const std::vector<Router::ResetHeaderParserSharedPtr>& resetHeaders() const override {
-      return reset_headers_;
-    }
-    std::chrono::milliseconds resetMaxInterval() const override {
-      return std::chrono::milliseconds(300000);
-    }
-
-    const std::vector<uint32_t> retriable_status_codes_{};
-    const std::vector<Http::HeaderMatcherSharedPtr> retriable_headers_{};
-    const std::vector<Http::HeaderMatcherSharedPtr> retriable_request_headers_{};
-    const std::vector<Router::ResetHeaderParserSharedPtr> reset_headers_{};
-  };
-
   struct NullConfig : public Router::Config {
     Router::RouteConstSharedPtr route(const Http::RequestHeaderMap&, const StreamInfo::StreamInfo&,
                                       uint64_t) const override {
@@ -211,20 +171,21 @@ class AsyncStreamImpl : public AsyncClient::Stream,
 
   struct RouteEntryImpl : public Router::RouteEntry {
     RouteEntryImpl(
-        const std::string& cluster_name, const absl::optional<std::chrono::milliseconds>& timeout,
+        AsyncClientImpl& parent, const absl::optional<std::chrono::milliseconds>& timeout,
         const Protobuf::RepeatedPtrField<envoy::config::route::v3::RouteAction::HashPolicy>&
             hash_policy,
         const absl::optional<envoy::config::route::v3::RetryPolicy>& retry_policy)
-        : cluster_name_(cluster_name), timeout_(timeout) {
+        : cluster_name_(parent.cluster_->name()), timeout_(timeout) {
       if (!hash_policy.empty()) {
         hash_policy_ = std::make_unique<HashPolicyImpl>(hash_policy);
       }
       if (retry_policy.has_value()) {
         // ProtobufMessage::getStrictValidationVisitor() ?  how often do we do this?
+        Upstream::RetryExtensionFactoryContextImpl factory_context(parent.singleton_manager_);
         retry_policy_ = std::make_unique<Router::RetryPolicyImpl>(
-            retry_policy.value(), ProtobufMessage::getNullValidationVisitor());
+            retry_policy.value(), ProtobufMessage::getNullValidationVisitor(), factory_context);
       } else {
-        retry_policy_ = std::make_unique<NullRetryPolicy>();
+        retry_policy_ = std::make_unique<Router::RetryPolicyImpl>();
       }
     }
 
@@ -330,12 +291,11 @@ class AsyncStreamImpl : public AsyncClient::Stream,
   };
 
   struct RouteImpl : public Router::Route {
-    RouteImpl(const std::string& cluster_name,
-              const absl::optional<std::chrono::milliseconds>& timeout,
+    RouteImpl(AsyncClientImpl& parent, const absl::optional<std::chrono::milliseconds>& timeout,
               const Protobuf::RepeatedPtrField<envoy::config::route::v3::RouteAction::HashPolicy>&
                   hash_policy,
               const absl::optional<envoy::config::route::v3::RetryPolicy>& retry_policy)
-        : route_entry_(cluster_name, timeout, hash_policy, retry_policy), typed_metadata_({}) {}
+        : route_entry_(parent, timeout, hash_policy, retry_policy), typed_metadata_({}) {}
 
     // Router::Route
     const Router::DirectResponseEntry* directResponseEntry() const override { return nullptr; }
diff --git a/source/common/network/BUILD b/source/common/network/BUILD
index 6d85e13184c8a..114142e7040a8 100644
--- a/source/common/network/BUILD
+++ b/source/common/network/BUILD
@@ -343,6 +343,7 @@ envoy_cc_library(
         "//source/common/api:os_sys_calls_lib",
         "//source/common/common:assert_lib",
         "//source/common/common:minimal_logger_lib",
+        "//source/common/common:scalar_to_byte_vector_lib",
         "//source/common/common:utility_lib",
         "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
     ],
diff --git a/source/common/network/addr_family_aware_socket_option_impl.h b/source/common/network/addr_family_aware_socket_option_impl.h
index 749788a4d9fa4..85a1075371eac 100644
--- a/source/common/network/addr_family_aware_socket_option_impl.h
+++ b/source/common/network/addr_family_aware_socket_option_impl.h
@@ -24,9 +24,11 @@ class AddrFamilyAwareSocketOptionImpl : public Socket::Option,
   // Socket::Option
   bool setOption(Socket& socket,
                  envoy::config::core::v3::SocketOption::SocketState state) const override;
-  // The common socket options don't require a hash key.
-  void hashKey(std::vector<uint8_t>&) const override {}
-
+  void hashKey(std::vector<uint8_t>& hash_key) const override {
+    // Add both sub-options to the hash.
+    ipv4_option_->hashKey(hash_key);
+    ipv6_option_->hashKey(hash_key);
+  }
   absl::optional<Details>
   getOptionDetails(const Socket& socket,
                    envoy::config::core::v3::SocketOption::SocketState state) const override;
diff --git a/source/common/network/socket_option_impl.cc b/source/common/network/socket_option_impl.cc
index 929979c8fa128..ba9ff7362dc6d 100644
--- a/source/common/network/socket_option_impl.cc
+++ b/source/common/network/socket_option_impl.cc
@@ -5,6 +5,7 @@
 
 #include "source/common/api/os_sys_calls_impl.h"
 #include "source/common/common/assert.h"
+#include "source/common/common/scalar_to_byte_vector.h"
 #include "source/common/common/utility.h"
 #include "source/common/network/address_impl.h"
 
@@ -32,6 +33,14 @@ bool SocketOptionImpl::setOption(Socket& socket,
   return true;
 }
 
+void SocketOptionImpl::hashKey(std::vector<uint8_t>& hash_key) const {
+  if (optname_.hasValue()) {
+    pushScalarToByteVector(optname_.level(), hash_key);
+    pushScalarToByteVector(optname_.option(), hash_key);
+    hash_key.insert(hash_key.end(), value_.begin(), value_.end());
+  }
+}
+
 absl::optional<Socket::Option::Details>
 SocketOptionImpl::getOptionDetails(const Socket&,
                                    envoy::config::core::v3::SocketOption::SocketState state) const {
diff --git a/source/common/network/socket_option_impl.h b/source/common/network/socket_option_impl.h
index 4c47dde2c08a7..fd42517c7bd90 100644
--- a/source/common/network/socket_option_impl.h
+++ b/source/common/network/socket_option_impl.h
@@ -134,10 +134,7 @@ class SocketOptionImpl : public Socket::Option, Logger::Loggable<Logger::Id::con
   // Socket::Option
   bool setOption(Socket& socket,
                  envoy::config::core::v3::SocketOption::SocketState state) const override;
-
-  // The common socket options don't require a hash key.
-  void hashKey(std::vector<uint8_t>&) const override {}
-
+  void hashKey(std::vector<uint8_t>& hash_key) const override;
   absl::optional<Details>
   getOptionDetails(const Socket& socket,
                    envoy::config::core::v3::SocketOption::SocketState state) const override;
diff --git a/source/common/network/win32_redirect_records_option_impl.h b/source/common/network/win32_redirect_records_option_impl.h
index 3fbe5f6fc5855..efa88048d9705 100644
--- a/source/common/network/win32_redirect_records_option_impl.h
+++ b/source/common/network/win32_redirect_records_option_impl.h
@@ -20,8 +20,6 @@ class Win32RedirectRecordsOptionImpl : public Socket::Option,
   // Socket::Option
   bool setOption(Socket& socket,
                  envoy::config::core::v3::SocketOption::SocketState state) const override;
-
-  // The common socket options don't require a hash key.
   void hashKey(std::vector<uint8_t>&) const override;
 
   absl::optional<Details>
diff --git a/source/common/router/BUILD b/source/common/router/BUILD
index ffea1d8aa65e4..4e0bbe9145706 100644
--- a/source/common/router/BUILD
+++ b/source/common/router/BUILD
@@ -63,6 +63,7 @@ envoy_cc_library(
         "//source/common/http:utility_lib",
         "//source/common/protobuf:utility_lib",
         "//source/common/tracing:http_tracer_lib",
+        "//source/common/upstream:retry_factory_lib",
         "//source/extensions/filters/http/common:utility_lib",
         "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
         "@envoy_api//envoy/config/route/v3:pkg_cc_proto",
diff --git a/source/common/router/config_impl.cc b/source/common/router/config_impl.cc
index e71f5171a427c..adf6b70edeaba 100644
--- a/source/common/router/config_impl.cc
+++ b/source/common/router/config_impl.cc
@@ -37,6 +37,7 @@
 #include "source/common/router/retry_state_impl.h"
 #include "source/common/runtime/runtime_features.h"
 #include "source/common/tracing/http_tracer_impl.h"
+#include "source/common/upstream/retry_factory.h"
 #include "source/extensions/filters/http/common/utility.h"
 
 #include "absl/strings/match.h"
@@ -87,7 +88,8 @@ HedgePolicyImpl::HedgePolicyImpl(const envoy::config::route::v3::HedgePolicy& he
 HedgePolicyImpl::HedgePolicyImpl() : initial_requests_(1), hedge_on_per_try_timeout_(false) {}
 
 RetryPolicyImpl::RetryPolicyImpl(const envoy::config::route::v3::RetryPolicy& retry_policy,
-                                 ProtobufMessage::ValidationVisitor& validation_visitor)
+                                 ProtobufMessage::ValidationVisitor& validation_visitor,
+                                 Upstream::RetryExtensionFactoryContext& factory_context)
     : retriable_headers_(
           Http::HeaderUtility::buildHeaderMatcherVector(retry_policy.retriable_headers())),
       retriable_request_headers_(
@@ -118,6 +120,16 @@ RetryPolicyImpl::RetryPolicyImpl(const envoy::config::route::v3::RetryPolicy& re
                                      retry_priority, validation_visitor, factory));
   }
 
+  for (const auto& options_predicate : retry_policy.retry_options_predicates()) {
+    auto& factory =
+        Envoy::Config::Utility::getAndCheckFactory<Upstream::RetryOptionsPredicateFactory>(
+            options_predicate);
+    retry_options_predicates_.emplace_back(
+        factory.createOptionsPredicate(*Envoy::Config::Utility::translateToFactoryConfig(
+                                           options_predicate, validation_visitor, factory),
+                                       factory_context));
+  }
+
   auto host_selection_attempts = retry_policy.host_selection_retry_max_attempts();
   if (host_selection_attempts) {
     host_selection_attempts_ = host_selection_attempts;
@@ -350,7 +362,8 @@ RouteEntryImplBase::RouteEntryImplBase(const VirtualHostImpl& vhost,
       prefix_rewrite_redirect_(route.redirect().prefix_rewrite()),
       strip_query_(route.redirect().strip_query()),
       hedge_policy_(buildHedgePolicy(vhost.hedgePolicy(), route.route())),
-      retry_policy_(buildRetryPolicy(vhost.retryPolicy(), route.route(), validator)),
+      retry_policy_(
+          buildRetryPolicy(vhost.retryPolicy(), route.route(), validator, factory_context)),
       internal_redirect_policy_(
           buildInternalRedirectPolicy(route.route(), validator, route.name())),
       rate_limit_policy_(route.route().rate_limits(), validator),
@@ -893,15 +906,18 @@ HedgePolicyImpl RouteEntryImplBase::buildHedgePolicy(
 RetryPolicyImpl RouteEntryImplBase::buildRetryPolicy(
     const absl::optional<envoy::config::route::v3::RetryPolicy>& vhost_retry_policy,
     const envoy::config::route::v3::RouteAction& route_config,
-    ProtobufMessage::ValidationVisitor& validation_visitor) const {
+    ProtobufMessage::ValidationVisitor& validation_visitor,
+    Server::Configuration::ServerFactoryContext& factory_context) const {
+  Upstream::RetryExtensionFactoryContextImpl retry_factory_context(
+      factory_context.singletonManager());
   // Route specific policy wins, if available.
   if (route_config.has_retry_policy()) {
-    return RetryPolicyImpl(route_config.retry_policy(), validation_visitor);
+    return RetryPolicyImpl(route_config.retry_policy(), validation_visitor, retry_factory_context);
   }
 
   // If not, we fallback to the virtual host policy if there is one.
   if (vhost_retry_policy) {
-    return RetryPolicyImpl(vhost_retry_policy.value(), validation_visitor);
+    return RetryPolicyImpl(vhost_retry_policy.value(), validation_visitor, retry_factory_context);
   }
 
   // Otherwise, an empty policy will do.
diff --git a/source/common/router/config_impl.h b/source/common/router/config_impl.h
index cc7a9a3dc5497..6688b14a548cd 100644
--- a/source/common/router/config_impl.h
+++ b/source/common/router/config_impl.h
@@ -292,7 +292,8 @@ class RetryPolicyImpl : public RetryPolicy {
 
 public:
   RetryPolicyImpl(const envoy::config::route::v3::RetryPolicy& retry_policy,
-                  ProtobufMessage::ValidationVisitor& validation_visitor);
+                  ProtobufMessage::ValidationVisitor& validation_visitor,
+                  Upstream::RetryExtensionFactoryContext& factory_context);
   RetryPolicyImpl() = default;
 
   // Router::RetryPolicy
@@ -302,6 +303,10 @@ class RetryPolicyImpl : public RetryPolicy {
   uint32_t retryOn() const override { return retry_on_; }
   std::vector<Upstream::RetryHostPredicateSharedPtr> retryHostPredicates() const override;
   Upstream::RetryPrioritySharedPtr retryPriority() const override;
+  absl::Span<const Upstream::RetryOptionsPredicateConstSharedPtr>
+  retryOptionsPredicates() const override {
+    return retry_options_predicates_;
+  }
   uint32_t hostSelectionMaxAttempts() const override { return host_selection_attempts_; }
   const std::vector<uint32_t>& retriableStatusCodes() const override {
     return retriable_status_codes_;
@@ -344,6 +349,7 @@ class RetryPolicyImpl : public RetryPolicy {
   std::vector<ResetHeaderParserSharedPtr> reset_headers_{};
   std::chrono::milliseconds reset_max_interval_{300000};
   ProtobufMessage::ValidationVisitor* validation_visitor_{};
+  std::vector<Upstream::RetryOptionsPredicateConstSharedPtr> retry_options_predicates_;
 };
 
 /**
@@ -849,7 +855,8 @@ class RouteEntryImplBase : public RouteEntry,
   RetryPolicyImpl
   buildRetryPolicy(const absl::optional<envoy::config::route::v3::RetryPolicy>& vhost_retry_policy,
                    const envoy::config::route::v3::RouteAction& route_config,
-                   ProtobufMessage::ValidationVisitor& validation_visitor) const;
+                   ProtobufMessage::ValidationVisitor& validation_visitor,
+                   Server::Configuration::ServerFactoryContext& factory_context) const;
 
   InternalRedirectPolicyImpl
   buildInternalRedirectPolicy(const envoy::config::route::v3::RouteAction& route_config,
diff --git a/source/common/router/router.cc b/source/common/router/router.cc
index 8e3178738e411..054b6a6858ecd 100644
--- a/source/common/router/router.cc
+++ b/source/common/router/router.cc
@@ -953,6 +953,7 @@ void Filter::onSoftPerTryTimeout(UpstreamRequest& upstream_request) {
         retry_state_->shouldHedgeRetryPerTryTimeout([this]() -> void { doRetry(); });
 
     if (retry_status == RetryStatus::Yes) {
+      runRetryOptionsPredicates(upstream_request);
       pending_retries_++;
 
       // Don't increment upstream_host->stats().rq_error_ here, we'll do that
@@ -1103,6 +1104,7 @@ bool Filter::maybeRetryReset(Http::StreamResetReason reset_reason,
   const RetryStatus retry_status =
       retry_state_->shouldRetryReset(reset_reason, [this]() -> void { doRetry(); });
   if (retry_status == RetryStatus::Yes) {
+    runRetryOptionsPredicates(upstream_request);
     pending_retries_++;
 
     if (upstream_request.upstreamHost()) {
@@ -1320,6 +1322,7 @@ void Filter::onUpstreamHeaders(uint64_t response_code, Http::ResponseHeaderMapPt
       const RetryStatus retry_status =
           retry_state_->shouldRetryHeaders(*headers, [this]() -> void { doRetry(); });
       if (retry_status == RetryStatus::Yes) {
+        runRetryOptionsPredicates(upstream_request);
         pending_retries_++;
         upstream_request.upstreamHost()->stats().rq_error_.inc();
         Http::CodeStats& code_stats = httpContext().codeStats();
@@ -1651,6 +1654,17 @@ bool Filter::convertRequestHeadersForInternalRedirect(Http::RequestHeaderMap& do
   return true;
 }
 
+void Filter::runRetryOptionsPredicates(UpstreamRequest& retriable_request) {
+  for (const auto& options_predicate : route_entry_->retryPolicy().retryOptionsPredicates()) {
+    const Upstream::RetryOptionsPredicate::UpdateOptionsParameters parameters{
+        retriable_request.streamInfo(), upstreamSocketOptions()};
+    auto ret = options_predicate->updateOptions(parameters);
+    if (ret.new_upstream_socket_options_.has_value()) {
+      upstream_options_ = ret.new_upstream_socket_options_.value();
+    }
+  }
+}
+
 void Filter::doRetry() {
   ENVOY_STREAM_LOG(debug, "performing retry", *callbacks_);
 
diff --git a/source/common/router/router.h b/source/common/router/router.h
index 2e3a21eacda90..e9d0234aed5d0 100644
--- a/source/common/router/router.h
+++ b/source/common/router/router.h
@@ -524,6 +524,7 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   void updateOutlierDetection(Upstream::Outlier::Result result, UpstreamRequest& upstream_request,
                               absl::optional<uint64_t> code);
   void doRetry();
+  void runRetryOptionsPredicates(UpstreamRequest& retriable_request);
   // Called immediately after a non-5xx header is received from upstream, performs stats accounting
   // and handle difference between gRPC and non-gRPC requests.
   void handleNon5xxResponseHeaders(absl::optional<Grpc::Status::GrpcStatus> grpc_status,
diff --git a/source/common/upstream/BUILD b/source/common/upstream/BUILD
index cca323285187e..eeed927f16837 100644
--- a/source/common/upstream/BUILD
+++ b/source/common/upstream/BUILD
@@ -106,6 +106,14 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "retry_factory_lib",
+    hdrs = ["retry_factory.h"],
+    deps = [
+        "//envoy/upstream:retry_interface",
+    ],
+)
+
 envoy_cc_library(
     name = "conn_pool_map",
     hdrs = ["conn_pool_map.h"],
diff --git a/source/common/upstream/cluster_manager_impl.h b/source/common/upstream/cluster_manager_impl.h
index ce5480b1ea920..f96b417d9ae61 100644
--- a/source/common/upstream/cluster_manager_impl.h
+++ b/source/common/upstream/cluster_manager_impl.h
@@ -88,6 +88,7 @@ class ProdClusterManagerFactory : public ClusterManagerFactory {
                       const xds::core::v3::ResourceLocator* cds_resources_locator,
                       ClusterManager& cm) override;
   Secret::SecretManager& secretManager() override { return secret_manager_; }
+  Singleton::Manager& singletonManager() override { return singleton_manager_; }
 
 protected:
   Event::Dispatcher& main_thread_dispatcher_;
diff --git a/source/common/upstream/retry_factory.h b/source/common/upstream/retry_factory.h
new file mode 100644
index 0000000000000..7c335116cb663
--- /dev/null
+++ b/source/common/upstream/retry_factory.h
@@ -0,0 +1,21 @@
+#pragma once
+
+#include "envoy/upstream/retry.h"
+
+namespace Envoy {
+namespace Upstream {
+
+class RetryExtensionFactoryContextImpl : public Upstream::RetryExtensionFactoryContext {
+public:
+  RetryExtensionFactoryContextImpl(Singleton::Manager& singleton_manager)
+      : singleton_manager_(singleton_manager) {}
+
+  // Upstream::RetryOptionsPredicateFactoryContext
+  Singleton::Manager& singletonManager() override { return singleton_manager_; }
+
+private:
+  Singleton::Manager& singleton_manager_;
+};
+
+} // namespace Upstream
+} // namespace Envoy
diff --git a/test/common/http/async_client_impl_test.cc b/test/common/http/async_client_impl_test.cc
index 5728e0fa31ada..28d78009c2fdb 100644
--- a/test/common/http/async_client_impl_test.cc
+++ b/test/common/http/async_client_impl_test.cc
@@ -1544,10 +1544,10 @@ TEST_F(AsyncClientImplTest, DumpState) {
 } // namespace
 
 // Must not be in anonymous namespace for friend to work.
-class AsyncClientImplUnitTest : public testing::Test {
+class AsyncClientImplUnitTest : public AsyncClientImplTest {
 public:
   std::unique_ptr<AsyncStreamImpl::RouteImpl> route_impl_{new AsyncStreamImpl::RouteImpl(
-      "foo", absl::nullopt,
+      client_, absl::nullopt,
       Protobuf::RepeatedPtrField<envoy::config::route::v3::RouteAction::HashPolicy>(),
       absl::nullopt)};
   AsyncStreamImpl::NullVirtualHost vhost_;
@@ -1559,7 +1559,7 @@ class AsyncClientImplUnitTest : public testing::Test {
     TestUtility::loadFromYaml(yaml_config, retry_policy);
 
     route_impl_ = std::make_unique<AsyncStreamImpl::RouteImpl>(
-        "foo", absl::nullopt,
+        client_, absl::nullopt,
         Protobuf::RepeatedPtrField<envoy::config::route::v3::RouteAction::HashPolicy>(),
         std::move(retry_policy));
   }
@@ -1567,7 +1567,6 @@ class AsyncClientImplUnitTest : public testing::Test {
 
 // Test the extended fake route that AsyncClient uses.
 TEST_F(AsyncClientImplUnitTest, NullRouteImplInitTest) {
-
   auto& route_entry = *(route_impl_->routeEntry());
 
   EXPECT_EQ(nullptr, route_impl_->decorator());
@@ -1598,7 +1597,6 @@ TEST_F(AsyncClientImplUnitTest, NullRouteImplInitTest) {
 }
 
 TEST_F(AsyncClientImplUnitTest, RouteImplInitTestWithRetryPolicy) {
-
   const std::string yaml = R"EOF(
 per_try_timeout: 30s
 num_retries: 10
diff --git a/test/common/network/socket_option_impl_test.cc b/test/common/network/socket_option_impl_test.cc
index c2736caed50d9..6979b7b7f3fb9 100644
--- a/test/common/network/socket_option_impl_test.cc
+++ b/test/common/network/socket_option_impl_test.cc
@@ -41,6 +41,10 @@ TEST_F(SocketOptionImplTest, HasName) {
   EXPECT_LOG_CONTAINS(
       "warning", "Setting SOL_SOCKET/SO_SNDBUF option on socket failed",
       socket_option.setOption(socket_, envoy::config::core::v3::SocketOption::STATE_PREBIND));
+
+  std::vector<uint8_t> hash_key;
+  socket_option.hashKey(hash_key);
+  EXPECT_FALSE(hash_key.empty());
 }
 
 TEST_F(SocketOptionImplTest, SetOptionSuccessTrue) {
diff --git a/test/common/router/config_impl_test.cc b/test/common/router/config_impl_test.cc
index ccf64a29195ff..9961f532b698c 100644
--- a/test/common/router/config_impl_test.cc
+++ b/test/common/router/config_impl_test.cc
@@ -3401,25 +3401,56 @@ TEST_F(RouteMatcherTest, Retry) {
                 .retryOn());
 }
 
+class TestRetryOptionsPredicateFactory : public Upstream::RetryOptionsPredicateFactory {
+public:
+  Upstream::RetryOptionsPredicateConstSharedPtr
+  createOptionsPredicate(const Protobuf::Message&,
+                         Upstream::RetryExtensionFactoryContext&) override {
+    return nullptr;
+  }
+
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    // Using Struct instead of a custom empty config proto. This is only allowed in tests.
+    return ProtobufTypes::MessagePtr{new Envoy::ProtobufWkt::Struct()};
+  }
+
+  std::string name() const override { return "test_retry_options_predicate_factory"; }
+};
+
 TEST_F(RouteMatcherTest, RetryVirtualHostLevel) {
   const std::string yaml = R"EOF(
 virtual_hosts:
 - domains: [www.lyft.com]
   per_request_buffer_limit_bytes: 8
   name: www
-  retry_policy: {num_retries: 3, per_try_timeout: 1s, retry_on: '5xx,gateway-error,connect-failure,reset'}
+  retry_policy:
+    num_retries: 3
+    per_try_timeout: 1s
+    retry_on: '5xx,gateway-error,connect-failure,reset'
+    retry_options_predicates:
+    - name: test_retry_options_predicate_factory
+      typed_config:
+        "@type": type.googleapis.com/google.protobuf.Struct
   routes:
   - match: {prefix: /foo}
     per_request_buffer_limit_bytes: 7
     route:
       cluster: www
-      retry_policy: {retry_on: connect-failure}
+      retry_policy:
+        retry_on: connect-failure
+        retry_options_predicates:
+        - name: test_retry_options_predicate_factory
+          typed_config:
+            "@type": type.googleapis.com/google.protobuf.Struct
   - match: {prefix: /bar}
     route: {cluster: www}
   - match: {prefix: /}
     route: {cluster: www}
   )EOF";
 
+  TestRetryOptionsPredicateFactory factory;
+  Registry::InjectFactory<Upstream::RetryOptionsPredicateFactory> registered(factory);
+
   factory_context_.cluster_manager_.initializeClusters({"www"}, {});
   TestConfigImpl config(parseRouteConfigurationFromYaml(yaml), factory_context_, true);
 
@@ -3441,6 +3472,11 @@ TEST_F(RouteMatcherTest, RetryVirtualHostLevel) {
   EXPECT_EQ(7U, config.route(genHeaders("www.lyft.com", "/foo", "GET"), 0)
                     ->routeEntry()
                     ->retryShadowBufferLimit());
+  EXPECT_EQ(1U, config.route(genHeaders("www.lyft.com", "/foo", "GET"), 0)
+                    ->routeEntry()
+                    ->retryPolicy()
+                    .retryOptionsPredicates()
+                    .size());
 
   // Virtual Host level retry policy kicks in.
   EXPECT_EQ(std::chrono::milliseconds(1000),
@@ -3476,6 +3512,11 @@ TEST_F(RouteMatcherTest, RetryVirtualHostLevel) {
   EXPECT_EQ(8U, config.route(genHeaders("www.lyft.com", "/", "GET"), 0)
                     ->routeEntry()
                     ->retryShadowBufferLimit());
+  EXPECT_EQ(1U, config.route(genHeaders("www.lyft.com", "/", "GET"), 0)
+                    ->routeEntry()
+                    ->retryPolicy()
+                    .retryOptionsPredicates()
+                    .size());
 }
 
 TEST_F(RouteMatcherTest, GrpcRetry) {
diff --git a/test/common/router/router_test.cc b/test/common/router/router_test.cc
index 9bf90e943e9a6..1ff5661d20e03 100644
--- a/test/common/router/router_test.cc
+++ b/test/common/router/router_test.cc
@@ -911,7 +911,18 @@ TEST_F(RouterTest, EnvoyAttemptCountInRequestNotOverwritten) {
       /* expected_count */ 123);
 }
 
+class MockRetryOptionsPredicate : public Upstream::RetryOptionsPredicate {
+public:
+  MOCK_METHOD(UpdateOptionsReturn, updateOptions, (const UpdateOptionsParameters& parameters),
+              (const));
+};
+
+// Also verify retry options predicates work.
 TEST_F(RouterTest, EnvoyAttemptCountInRequestUpdatedInRetries) {
+  auto retry_options_predicate = std::make_shared<MockRetryOptionsPredicate>();
+  callbacks_.route_->route_entry_.retry_policy_.retry_options_predicates_.emplace_back(
+      retry_options_predicate);
+
   setIncludeAttemptCountInRequest(true);
 
   NiceMock<Http::MockRequestEncoder> encoder1;
@@ -938,13 +949,21 @@ TEST_F(RouterTest, EnvoyAttemptCountInRequestUpdatedInRetries) {
 
   // 5xx response.
   router_.retry_state_->expectHeadersRetry();
+  Upstream::RetryOptionsPredicate::UpdateOptionsReturn update_options_return{
+      std::make_shared<Network::Socket::Options>()};
+  EXPECT_CALL(*retry_options_predicate, updateOptions(_)).WillOnce(Return(update_options_return));
   Http::ResponseHeaderMapPtr response_headers1(
       new Http::TestResponseHeaderMapImpl{{":status", "503"}});
   EXPECT_CALL(cm_.thread_local_cluster_.conn_pool_.host_->outlier_detector_,
               putHttpResponseCode(503));
+  // NOLINTNEXTLINE(clang-analyzer-core.CallAndMessage)
   response_decoder->decodeHeaders(std::move(response_headers1), true);
   EXPECT_TRUE(verifyHostUpstreamStats(0, 1));
 
+  // Verify retry options predicate return values have been updated.
+  EXPECT_EQ(update_options_return.new_upstream_socket_options_.value(),
+            router_.upstreamSocketOptions());
+
   // We expect the 5xx response to kick off a new request.
   EXPECT_CALL(encoder1.stream_, resetStream(_)).Times(0);
   NiceMock<Http::MockRequestEncoder> encoder2;
@@ -2260,8 +2279,12 @@ TEST_F(RouterTest, UpstreamPerTryTimeoutExcludesNewStream) {
 
 // Tests that a retry is sent after the first request hits the per try timeout, but then
 // headers received in response to the first request are still used (and the 2nd request
-// canceled).
+// canceled). Also verify retry options predicates work.
 TEST_F(RouterTest, HedgedPerTryTimeoutFirstRequestSucceeds) {
+  auto retry_options_predicate = std::make_shared<MockRetryOptionsPredicate>();
+  callbacks_.route_->route_entry_.retry_policy_.retry_options_predicates_.emplace_back(
+      retry_options_predicate);
+
   enableHedgeOnPerTryTimeout();
 
   NiceMock<Http::MockRequestEncoder> encoder1;
@@ -2296,6 +2319,7 @@ TEST_F(RouterTest, HedgedPerTryTimeoutFirstRequestSucceeds) {
   NiceMock<Http::MockRequestEncoder> encoder2;
   Http::ResponseDecoder* response_decoder2 = nullptr;
   router_.retry_state_->expectHedgedPerTryTimeoutRetry();
+  EXPECT_CALL(*retry_options_predicate, updateOptions(_));
   per_try_timeout_->invokeCallback();
 
   EXPECT_CALL(cm_.thread_local_cluster_.conn_pool_, newStream(_, _))
@@ -2714,8 +2738,12 @@ TEST_F(RouterTest, BadHeadersDroppedIfPreviousRetryScheduled) {
 }
 
 // Test retrying a request, when the first attempt fails before the client
-// has sent any of the body.
+// has sent any of the body. Also verify retry options predicates work.
 TEST_F(RouterTest, RetryRequestBeforeBody) {
+  auto retry_options_predicate = std::make_shared<MockRetryOptionsPredicate>();
+  callbacks_.route_->route_entry_.retry_policy_.retry_options_predicates_.emplace_back(
+      retry_options_predicate);
+
   NiceMock<Http::MockRequestEncoder> encoder1;
   Http::ResponseDecoder* response_decoder = nullptr;
   EXPECT_CALL(cm_.thread_local_cluster_.conn_pool_, newStream(_, _))
@@ -2735,6 +2763,7 @@ TEST_F(RouterTest, RetryRequestBeforeBody) {
   router_.decodeHeaders(headers, false);
 
   router_.retry_state_->expectResetRetry();
+  EXPECT_CALL(*retry_options_predicate, updateOptions(_));
   encoder1.stream_.resetStream(Http::StreamResetReason::RemoteReset);
 
   NiceMock<Http::MockRequestEncoder> encoder2;
@@ -2766,6 +2795,7 @@ TEST_F(RouterTest, RetryRequestBeforeBody) {
       .WillOnce(Invoke([&](Http::ResponseHeaderMap& headers, bool) -> void {
         EXPECT_EQ(headers.Status()->value(), "200");
       }));
+  // NOLINTNEXTLINE(clang-analyzer-core.CallAndMessage)
   response_decoder->decodeHeaders(std::move(response_headers), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 1));
 }
diff --git a/test/common/upstream/test_cluster_manager.h b/test/common/upstream/test_cluster_manager.h
index 463f4fb2a9222..8bc91e7ac8772 100644
--- a/test/common/upstream/test_cluster_manager.h
+++ b/test/common/upstream/test_cluster_manager.h
@@ -115,6 +115,7 @@ class TestClusterManagerFactory : public ClusterManagerFactory {
   }
 
   Secret::SecretManager& secretManager() override { return secret_manager_; }
+  Singleton::Manager& singletonManager() override { return singleton_manager_; }
 
   MOCK_METHOD(ClusterManager*, clusterManagerFromProto_,
               (const envoy::config::bootstrap::v3::Bootstrap& bootstrap));
diff --git a/test/extensions/filters/http/original_src/original_src_test.cc b/test/extensions/filters/http/original_src/original_src_test.cc
index 5839baa88b9e5..a26db42297c79 100644
--- a/test/extensions/filters/http/original_src/original_src_test.cc
+++ b/test/extensions/filters/http/original_src/original_src_test.cc
@@ -110,9 +110,17 @@ TEST_F(OriginalSrcHttpTest, DecodeHeadersIpv4AddressUsesCorrectAddress) {
     option->hashKey(key);
   }
 
-  std::vector<uint8_t> expected_key = {1, 2, 3, 4};
-
-  EXPECT_EQ(key, expected_key);
+  // The first part of the hash is the address. Then come the other options. On Windows there are
+  // is only the single option. On other platforms there are more that get hashed.
+  EXPECT_EQ(key[0], 1);
+  EXPECT_EQ(key[1], 2);
+  EXPECT_EQ(key[2], 3);
+  EXPECT_EQ(key[3], 4);
+#ifndef WIN32
+  EXPECT_GT(key.size(), 4);
+#else
+  EXPECT_EQ(key.size(), 4);
+#endif
 }
 
 TEST_F(OriginalSrcHttpTest, DecodeHeadersIpv4AddressBleachesPort) {
diff --git a/test/integration/integration_test.cc b/test/integration/integration_test.cc
index 794a49adda065..cc004f0a616f9 100644
--- a/test/integration/integration_test.cc
+++ b/test/integration/integration_test.cc
@@ -9,6 +9,7 @@
 
 #include "source/common/http/header_map_impl.h"
 #include "source/common/http/headers.h"
+#include "source/common/network/socket_option_factory.h"
 #include "source/common/network/socket_option_impl.h"
 #include "source/common/network/utility.h"
 #include "source/common/protobuf/utility.h"
@@ -20,6 +21,7 @@
 #include "test/mocks/http/mocks.h"
 #include "test/test_common/network_utility.h"
 #include "test/test_common/printers.h"
+#include "test/test_common/registry.h"
 #include "test/test_common/utility.h"
 
 #include "gtest/gtest.h"
@@ -2097,6 +2099,98 @@ TEST_P(IntegrationTest, RandomPreconnect) {
   }
 }
 
+class TestRetryOptionsPredicateFactory : public Upstream::RetryOptionsPredicateFactory {
+public:
+  Upstream::RetryOptionsPredicateConstSharedPtr
+  createOptionsPredicate(const Protobuf::Message&,
+                         Upstream::RetryExtensionFactoryContext&) override {
+    return std::make_shared<TestPredicate>();
+  }
+
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    // Using Struct instead of a custom empty config proto. This is only allowed in tests.
+    return ProtobufTypes::MessagePtr{new Envoy::ProtobufWkt::Struct()};
+  }
+
+  std::string name() const override { return "test_retry_options_predicate_factory"; }
+
+private:
+  struct TestPredicate : public Upstream::RetryOptionsPredicate {
+    UpdateOptionsReturn updateOptions(const UpdateOptionsParameters&) const override {
+      UpdateOptionsReturn ret;
+      Network::TcpKeepaliveConfig tcp_keepalive_config;
+      tcp_keepalive_config.keepalive_probes_ = 1;
+      tcp_keepalive_config.keepalive_time_ = 1;
+      tcp_keepalive_config.keepalive_interval_ = 1;
+      ret.new_upstream_socket_options_ =
+          Network::SocketOptionFactory::buildTcpKeepaliveOptions(tcp_keepalive_config);
+      return ret;
+    }
+  };
+};
+
+// Verify that a test retry options predicate starts a new connection pool with a new connection.
+TEST_P(IntegrationTest, RetryOptionsPredicate) {
+  TestRetryOptionsPredicateFactory factory;
+  Registry::InjectFactory<Upstream::RetryOptionsPredicateFactory> registered(factory);
+
+  config_helper_.addConfigModifier(
+      [](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
+             hcm) {
+        auto* route_config = hcm.mutable_route_config();
+        auto* virtual_host = route_config->mutable_virtual_hosts(0);
+        auto* route = virtual_host->mutable_routes(0)->mutable_route();
+        auto* retry_policy = route->mutable_retry_policy();
+        retry_policy->set_retry_on("5xx");
+        auto* predicate = retry_policy->add_retry_options_predicates();
+        predicate->set_name("test_retry_options_predicate_factory");
+        predicate->mutable_typed_config()->set_type_url(
+            "type.googleapis.com/google.protobuf.Struct");
+      });
+
+  initialize();
+
+  Http::TestRequestHeaderMapImpl request_headers{
+      {":method", "GET"},
+      {":path", "/some/path"},
+      {":scheme", "http"},
+      {":authority", "cluster_0"},
+  };
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  auto response = codec_client_->makeHeaderOnlyRequest(request_headers);
+  AssertionResult result =
+      fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection_);
+  RELEASE_ASSERT(result, result.message());
+  result = fake_upstream_connection_->waitForNewStream(*dispatcher_, upstream_request_);
+  RELEASE_ASSERT(result, result.message());
+  result = upstream_request_->waitForEndStream(*dispatcher_);
+  RELEASE_ASSERT(result, result.message());
+
+  // Force a retry and run the predicate
+  upstream_request_->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "503"}}, true);
+
+  // Using a different socket option will cause a new connection pool to be used and a new
+  // connection.
+  FakeHttpConnectionPtr new_upstream_connection;
+  FakeStreamPtr new_upstream_request;
+  result = fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, new_upstream_connection);
+  RELEASE_ASSERT(result, result.message());
+  result = new_upstream_connection->waitForNewStream(*dispatcher_, new_upstream_request);
+  RELEASE_ASSERT(result, result.message());
+  result = new_upstream_request->waitForEndStream(*dispatcher_);
+  RELEASE_ASSERT(result, result.message());
+
+  new_upstream_request->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "200"}}, true);
+  result = response->waitForEndStream();
+  RELEASE_ASSERT(result, result.message());
+
+  result = new_upstream_connection->close();
+  RELEASE_ASSERT(result, result.message());
+  result = new_upstream_connection->waitForDisconnect();
+  RELEASE_ASSERT(result, result.message());
+}
+
 // Tests that a filter (set-route-filter) using the setRoute callback and DelegatingRoute mechanism
 // successfully overrides the cached route, and subsequently, the request's upstream cluster
 // selection.
diff --git a/test/mocks/router/mocks.h b/test/mocks/router/mocks.h
index 1bf7e452ed56b..e9ccd1de05a2e 100644
--- a/test/mocks/router/mocks.h
+++ b/test/mocks/router/mocks.h
@@ -109,6 +109,10 @@ class TestRetryPolicy : public RetryPolicy {
   uint32_t retryOn() const override { return retry_on_; }
   MOCK_METHOD(std::vector<Upstream::RetryHostPredicateSharedPtr>, retryHostPredicates, (), (const));
   MOCK_METHOD(Upstream::RetryPrioritySharedPtr, retryPriority, (), (const));
+  absl::Span<const Upstream::RetryOptionsPredicateConstSharedPtr>
+  retryOptionsPredicates() const override {
+    return retry_options_predicates_;
+  }
   uint32_t hostSelectionMaxAttempts() const override { return host_selection_max_attempts_; }
   const std::vector<uint32_t>& retriableStatusCodes() const override {
     return retriable_status_codes_;
@@ -139,6 +143,7 @@ class TestRetryPolicy : public RetryPolicy {
   absl::optional<std::chrono::milliseconds> max_interval_{};
   std::vector<ResetHeaderParserSharedPtr> reset_headers_{};
   std::chrono::milliseconds reset_max_interval_{300000};
+  std::vector<Upstream::RetryOptionsPredicateConstSharedPtr> retry_options_predicates_;
 };
 
 class MockInternalRedirectPolicy : public InternalRedirectPolicy {
diff --git a/test/mocks/upstream/BUILD b/test/mocks/upstream/BUILD
index e7fcdef38adcf..e6c41713f5458 100644
--- a/test/mocks/upstream/BUILD
+++ b/test/mocks/upstream/BUILD
@@ -224,7 +224,9 @@ envoy_cc_mock(
     hdrs = ["cluster_manager_factory.h"],
     deps = [
         "//envoy/upstream:cluster_manager_interface",
+        "//source/common/singleton:manager_impl_lib",
         "//test/mocks/secret:secret_mocks",
+        "//test/test_common:thread_factory_for_test_lib",
     ],
 )
 
diff --git a/test/mocks/upstream/cluster_manager_factory.h b/test/mocks/upstream/cluster_manager_factory.h
index a9354c97998ed..b4328b31beb12 100644
--- a/test/mocks/upstream/cluster_manager_factory.h
+++ b/test/mocks/upstream/cluster_manager_factory.h
@@ -2,7 +2,10 @@
 
 #include "envoy/upstream/cluster_manager.h"
 
+#include "source/common/singleton/manager_impl.h"
+
 #include "test/mocks/secret/mocks.h"
+#include "test/test_common/thread_factory_for_test.h"
 
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
@@ -16,6 +19,7 @@ class MockClusterManagerFactory : public ClusterManagerFactory {
   ~MockClusterManagerFactory() override;
 
   Secret::MockSecretManager& secretManager() override { return secret_manager_; };
+  Singleton::Manager& singletonManager() override { return singleton_manager_; }
 
   MOCK_METHOD(ClusterManagerPtr, clusterManagerFromProto,
               (const envoy::config::bootstrap::v3::Bootstrap& bootstrap));
@@ -44,6 +48,7 @@ class MockClusterManagerFactory : public ClusterManagerFactory {
 
 private:
   NiceMock<Secret::MockSecretManager> secret_manager_;
+  Singleton::ManagerImpl singleton_manager_{Thread::threadFactoryForTest()};
 };
 } // namespace Upstream
 } // namespace Envoy