diff --git a/api/envoy/api/v2/cluster/circuit_breaker.proto b/api/envoy/api/v2/cluster/circuit_breaker.proto
index ebee99dae163d..f219fa07b4feb 100644
--- a/api/envoy/api/v2/cluster/circuit_breaker.proto
+++ b/api/envoy/api/v2/cluster/circuit_breaker.proto
@@ -51,6 +51,13 @@ message CircuitBreakers {
     // the number of resources remaining until the circuit breakers open. If
     // not specified, the default is false.
     bool track_remaining = 6;
+
+    // The maximum number of connection pools per cluster that Envoy will concurrently support at
+    // once. If not specified, the default is unlimited. Set this for clusters which create a
+    // large number of connection pools. See
+    // :ref:`Circuit Breaking <arch_overview_circuit_break_cluster_maximum_connection_pools>` for
+    // more details.
+    google.protobuf.UInt32Value max_connection_pools = 7;
   }
 
   // If multiple :ref:`Thresholds<envoy_api_msg_cluster.CircuitBreakers.Thresholds>`
diff --git a/docs/root/configuration/cluster_manager/cluster_stats.rst b/docs/root/configuration/cluster_manager/cluster_stats.rst
index 47c2a011c3e66..370a9e1402d17 100644
--- a/docs/root/configuration/cluster_manager/cluster_stats.rst
+++ b/docs/root/configuration/cluster_manager/cluster_stats.rst
@@ -149,6 +149,7 @@ Circuit breakers statistics will be rooted at *cluster.<name>.circuit_breakers.<
   :widths: 1, 1, 2
 
   cx_open, Gauge, Whether the connection circuit breaker is closed (0) or open (1)
+  cx_pool_open, Gauge, Whether the connection pool circuit breaker is closed (0) or open (1)
   rq_pending_open, Gauge, Whether the pending requests circuit breaker is closed (0) or open (1)
   rq_open, Gauge, Whether the requests circuit breaker is closed (0) or open (1)
   rq_retry_open, Gauge, Whether the retry circuit breaker is closed (0) or open (1)
diff --git a/docs/root/intro/arch_overview/circuit_breaking.rst b/docs/root/intro/arch_overview/circuit_breaking.rst
index b2b6e31aa8c30..152284363fb51 100644
--- a/docs/root/intro/arch_overview/circuit_breaking.rst
+++ b/docs/root/intro/arch_overview/circuit_breaking.rst
@@ -9,6 +9,8 @@ mesh is that Envoy enforces circuit breaking limits at the network level as oppo
 configure and code each application independently. Envoy supports various types of fully distributed
 (not coordinated) circuit breaking:
 
+.. _arch_overview_circuit_break_cluster_maximum_connections:
+
 * **Cluster maximum connections**: The maximum number of connections that Envoy will establish to
   all hosts in an upstream cluster. In practice this is only applicable to HTTP/1.1 clusters since
   HTTP/2 uses a single connection to each host. If this circuit breaker overflows the :ref:`upstream_cx_overflow
@@ -34,6 +36,20 @@ configure and code each application independently. Envoy supports various types
   :ref:`upstream_rq_retry_overflow <config_cluster_manager_cluster_stats>` counter for the cluster
   will increment.
 
+  .. _arch_overview_circuit_break_cluster_maximum_connection_pools:
+
+* **Cluster maximum concurrent connection pools**: The maximum number of connection pools that can be
+  concurrently instantiated. Some features, such as the
+  :ref:`Original Src Listener Filter <arch_overview_ip_transparency_original_src_listener>`, can
+  create an unbounded number of connection pools. When a cluster has exhausted its concurrent
+  connection pools, it will attempt to reclaim an idle one. If it cannot, then the circuit breaker
+  will overflow. This differs from
+  :ref:`Cluster maximum connections <arch_overview_circuit_break_cluster_maximum_connections>` in that
+  connection pools never time out, whereas connections typically will. Connections automatically
+  clean up; connection pools do not. Note that in order for a connection pool to function it needs
+  at least one upstream connection, so this value should likely be no greater than
+  :ref:`Cluster maximum connections <arch_overview_circuit_break_cluster_maximum_connections>`.
+
 Each circuit breaking limit is :ref:`configurable <config_cluster_manager_cluster_circuit_breakers>`
 and tracked on a per upstream cluster and per priority basis. This allows different components of
 the distributed system to be tuned independently and have different limits. The live state of these
diff --git a/docs/root/intro/version_history.rst b/docs/root/intro/version_history.rst
index b307ddc5d1bdb..db7a1014c84d6 100644
--- a/docs/root/intro/version_history.rst
+++ b/docs/root/intro/version_history.rst
@@ -69,6 +69,7 @@ Version history
 * upstream: added :ref:`degraded health value<arch_overview_load_balancing_degraded>` which allows
   routing to certain hosts only when there are insufficient healthy hosts available.
 * upstream: add cluster factory to allow creating and registering :ref:`custom cluster type<arch_overview_service_discovery_types_custom>`.
+* upstream: added a :ref:`circuit breaker <arch_overview_circuit_break_cluster_maximum_connection_pools>` to limit the number of concurrent connection pools in use.
 * tracing: added :ref:`verbose <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.tracing>` to support logging annotations on spans.
 * upstream: added support for host weighting and :ref:`locality weighting <arch_overview_load_balancing_locality_weighted_lb>` in the :ref:`ring hash load balancer <arch_overview_load_balancing_types_ring_hash>`, and added a :ref:`maximum_ring_size<envoy_api_field_Cluster.RingHashLbConfig.maximum_ring_size>` config parameter to strictly bound the ring size.
 * zookeeper: added a ZooKeeper proxy filter that parses ZooKeeper messages (requests/responses/events).
diff --git a/include/envoy/upstream/resource_manager.h b/include/envoy/upstream/resource_manager.h
index 4bd45feed3ddd..902ce75955f68 100644
--- a/include/envoy/upstream/resource_manager.h
+++ b/include/envoy/upstream/resource_manager.h
@@ -37,6 +37,11 @@ class Resource {
    */
   virtual void dec() PURE;
 
+  /**
+   * Decrement the resource count by a specific amount.
+   */
+  virtual void decBy(uint64_t amount) PURE;
+
   /**
    * @return the current maximum allowed number of this resource.
    */
@@ -73,6 +78,11 @@ class ResourceManager {
    * @return Resource& active retries.
    */
   virtual Resource& retries() PURE;
+
+  /**
+   * @return Resource& active connection pools.
+   */
+  virtual Resource& connectionPools() PURE;
 };
 
 } // namespace Upstream
diff --git a/include/envoy/upstream/upstream.h b/include/envoy/upstream/upstream.h
index e2c2300d6e5d6..3fdb15cdeee17 100644
--- a/include/envoy/upstream/upstream.h
+++ b/include/envoy/upstream/upstream.h
@@ -537,10 +537,12 @@ class PrioritySet {
   OPEN_GAUGE      (rq_pending_open)                                                                \
   OPEN_GAUGE      (rq_open)                                                                        \
   OPEN_GAUGE      (rq_retry_open)                                                                  \
+  OPEN_GAUGE      (cx_pool_open)                                                                   \
   REMAINING_GAUGE (remaining_cx)                                                                   \
   REMAINING_GAUGE (remaining_pending)                                                              \
   REMAINING_GAUGE (remaining_rq)                                                                   \
-  REMAINING_GAUGE (remaining_retries)
+  REMAINING_GAUGE (remaining_retries)                                                              \
+  REMAINING_GAUGE (remaining_cx_pools)
 // clang-format on
 
 /**
diff --git a/source/common/upstream/BUILD b/source/common/upstream/BUILD
index bd09d3f9329b1..0629e568dd77b 100644
--- a/source/common/upstream/BUILD
+++ b/source/common/upstream/BUILD
@@ -60,8 +60,7 @@ envoy_cc_library(
         "//source/common/protobuf:utility_lib",
         "//source/common/router:shadow_writer_lib",
         "//source/common/tcp:conn_pool_lib",
-        "//source/common/upstream:conn_pool_map",
-        "//source/common/upstream:conn_pool_map_impl_lib",
+        "//source/common/upstream:priority_conn_pool_map_impl_lib",
         "//source/common/upstream:upstream_lib",
         "@envoy_api//envoy/admin/v2alpha:config_dump_cc",
         "@envoy_api//envoy/api/v2/core:base_cc",
@@ -73,6 +72,8 @@ envoy_cc_library(
     hdrs = ["conn_pool_map.h"],
     deps = [
         "//include/envoy/event:dispatcher_interface",
+        "//include/envoy/upstream:resource_manager_interface",
+        "//include/envoy/upstream:upstream_interface",
         "//source/common/common:debug_recursion_checker_lib",
     ],
 )
@@ -85,6 +86,27 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "priority_conn_pool_map",
+    hdrs = ["priority_conn_pool_map.h"],
+    deps = [
+        ":conn_pool_map",
+        "//include/envoy/event:dispatcher_interface",
+        "//include/envoy/upstream:resource_manager_interface",
+        "//include/envoy/upstream:upstream_interface",
+        "//source/common/common:debug_recursion_checker_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "priority_conn_pool_map_impl_lib",
+    hdrs = ["priority_conn_pool_map_impl.h"],
+    deps = [
+        ":conn_pool_map_impl_lib",
+        ":priority_conn_pool_map",
+    ],
+)
+
 envoy_cc_library(
     name = "edf_scheduler_lib",
     hdrs = ["edf_scheduler.h"],
diff --git a/source/common/upstream/cluster_manager_impl.cc b/source/common/upstream/cluster_manager_impl.cc
index d178871866286..8880e01ff547f 100644
--- a/source/common/upstream/cluster_manager_impl.cc
+++ b/source/common/upstream/cluster_manager_impl.cc
@@ -31,10 +31,10 @@
 #include "common/router/shadow_writer_impl.h"
 #include "common/tcp/conn_pool.h"
 #include "common/upstream/cds_api_impl.h"
-#include "common/upstream/conn_pool_map_impl.h"
 #include "common/upstream/load_balancer_impl.h"
 #include "common/upstream/maglev_lb.h"
 #include "common/upstream/original_dst_cluster.h"
+#include "common/upstream/priority_conn_pool_map_impl.h"
 #include "common/upstream/ring_hash_lb.h"
 #include "common/upstream/subset_lb.h"
 
@@ -1043,7 +1043,7 @@ ClusterManagerImpl::ThreadLocalClusterManagerImpl::getHttpConnPoolsContainer(
     if (!allocate) {
       return nullptr;
     }
-    ConnPoolsContainer container{thread_local_dispatcher_};
+    ConnPoolsContainer container{thread_local_dispatcher_, host};
     container_iter = host_http_conn_pool_map_.emplace(host, std::move(container)).first;
   }
 
@@ -1132,7 +1132,7 @@ ClusterManagerImpl::ThreadLocalClusterManagerImpl::ClusterEntry::connPool(
   }
 
   // Inherit socket options from downstream connection, if set.
-  std::vector<uint8_t> hash_key = {uint8_t(protocol), uint8_t(priority)};
+  std::vector<uint8_t> hash_key = {uint8_t(protocol)};
 
   // Use downstream connection socket options for computing connection pool hash key, if any.
   // This allows socket options to control connection pooling so that connections with
@@ -1153,16 +1153,18 @@ ClusterManagerImpl::ThreadLocalClusterManagerImpl::ClusterEntry::connPool(
 
   // Note: to simplify this, we assume that the factory is only called in the scope of this
   // function. Otherwise, we'd need to capture a few of these variables by value.
-  ConnPoolsContainer::ConnPools::OptPoolRef pool = container.pools_->getPool(hash_key, [&]() {
-    return parent_.parent_.factory_.allocateConnPool(
-        parent_.thread_local_dispatcher_, host, priority, protocol,
-        have_options ? context->downstreamConnection()->socketOptions() : nullptr);
-  });
-  // The Connection Pool tracking is a work in progress. We plan for it to eventually have the
-  // ability to fail, but until we add upper layer handling for failures, it should not. So, assert
-  // that we don't accidentally add conditions that could allow it to fail.
-  ASSERT(pool.has_value(), "Pool allocation should never fail");
-  return &(pool.value().get());
+  ConnPoolsContainer::ConnPools::OptPoolRef pool =
+      container.pools_->getPool(priority, hash_key, [&]() {
+        return parent_.parent_.factory_.allocateConnPool(
+            parent_.thread_local_dispatcher_, host, priority, protocol,
+            have_options ? context->downstreamConnection()->socketOptions() : nullptr);
+      });
+
+  if (pool.has_value()) {
+    return &(pool.value().get());
+  } else {
+    return nullptr;
+  }
 }
 
 Tcp::ConnectionPool::Instance*
diff --git a/source/common/upstream/cluster_manager_impl.h b/source/common/upstream/cluster_manager_impl.h
index 0e9e8eb52667d..9586357bc4aa1 100644
--- a/source/common/upstream/cluster_manager_impl.h
+++ b/source/common/upstream/cluster_manager_impl.h
@@ -22,8 +22,8 @@
 
 #include "common/config/grpc_mux_impl.h"
 #include "common/http/async_client_impl.h"
-#include "common/upstream/conn_pool_map.h"
 #include "common/upstream/load_stats_reporter.h"
+#include "common/upstream/priority_conn_pool_map.h"
 #include "common/upstream/upstream_impl.h"
 
 namespace Envoy {
@@ -236,10 +236,10 @@ class ClusterManagerImpl : public ClusterManager, Logger::Loggable<Logger::Id::u
    */
   struct ThreadLocalClusterManagerImpl : public ThreadLocal::ThreadLocalObject {
     struct ConnPoolsContainer {
-      ConnPoolsContainer(Event::Dispatcher& dispatcher)
-          : pools_{std::make_shared<ConnPools>(dispatcher, absl::nullopt)} {}
+      ConnPoolsContainer(Event::Dispatcher& dispatcher, const HostConstSharedPtr& host)
+          : pools_{std::make_shared<ConnPools>(dispatcher, host)} {}
 
-      typedef ConnPoolMap<std::vector<uint8_t>, Http::ConnectionPool::Instance> ConnPools;
+      typedef PriorityConnPoolMap<std::vector<uint8_t>, Http::ConnectionPool::Instance> ConnPools;
 
       // This is a shared_ptr so we can keep it alive while cleaning up.
       std::shared_ptr<ConnPools> pools_;
diff --git a/source/common/upstream/conn_pool_map.h b/source/common/upstream/conn_pool_map.h
index f0ef4cd4391d9..20127dbe401a4 100644
--- a/source/common/upstream/conn_pool_map.h
+++ b/source/common/upstream/conn_pool_map.h
@@ -4,6 +4,8 @@
 #include <vector>
 
 #include "envoy/event/dispatcher.h"
+#include "envoy/upstream/resource_manager.h"
+#include "envoy/upstream/upstream.h"
 
 #include "common/common/debug_recursion_checker.h"
 
@@ -21,7 +23,8 @@ template <typename KEY_TYPE, typename POOL_TYPE> class ConnPoolMap {
   using DrainedCb = std::function<void()>;
   using OptPoolRef = absl::optional<std::reference_wrapper<POOL_TYPE>>;
 
-  ConnPoolMap(Event::Dispatcher& dispatcher, absl::optional<uint64_t> max_size);
+  ConnPoolMap(Event::Dispatcher& dispatcher, const HostConstSharedPtr& host,
+              ResourcePriority priority);
   ~ConnPoolMap();
   /**
    * Returns an existing pool for `key`, or creates a new one using `factory`. Note that it is
@@ -60,11 +63,17 @@ template <typename KEY_TYPE, typename POOL_TYPE> class ConnPoolMap {
    */
   bool freeOnePool();
 
+  /**
+   * Cleans up the active_pools_ map and updates resource tracking
+   **/
+  void clearActivePools();
+
   absl::flat_hash_map<KEY_TYPE, std::unique_ptr<POOL_TYPE>> active_pools_;
   Event::Dispatcher& thread_local_dispatcher_;
   std::vector<DrainedCb> cached_callbacks_;
   Common::DebugRecursionChecker recursion_checker_;
-  const absl::optional<uint64_t> max_size_;
+  const HostConstSharedPtr host_;
+  const ResourcePriority priority_;
 };
 
 } // namespace Upstream
diff --git a/source/common/upstream/conn_pool_map_impl.h b/source/common/upstream/conn_pool_map_impl.h
index 598dea9a21f85..156751942535f 100644
--- a/source/common/upstream/conn_pool_map_impl.h
+++ b/source/common/upstream/conn_pool_map_impl.h
@@ -7,11 +7,17 @@ namespace Upstream {
 
 template <typename KEY_TYPE, typename POOL_TYPE>
 ConnPoolMap<KEY_TYPE, POOL_TYPE>::ConnPoolMap(Envoy::Event::Dispatcher& dispatcher,
-                                              absl::optional<uint64_t> max_size)
-    : thread_local_dispatcher_(dispatcher), max_size_(max_size) {}
-
-template <typename KEY_TYPE, typename POOL_TYPE>
-ConnPoolMap<KEY_TYPE, POOL_TYPE>::~ConnPoolMap() = default;
+                                              const HostConstSharedPtr& host,
+                                              ResourcePriority priority)
+    : thread_local_dispatcher_(dispatcher), host_(host), priority_(priority) {}
+
+template <typename KEY_TYPE, typename POOL_TYPE> ConnPoolMap<KEY_TYPE, POOL_TYPE>::~ConnPoolMap() {
+  // Clean up the pools to ensure resource tracking is kept up to date. Note that we do not call
+  // `clear()` here to avoid doing a deferred delete. This triggers some unwanted race conditions
+  // on shutdown where deleted resources end up putting stuff on the deferred delete list after the
+  // worker threads have shut down.
+  clearActivePools();
+}
 
 template <typename KEY_TYPE, typename POOL_TYPE>
 typename ConnPoolMap<KEY_TYPE, POOL_TYPE>::OptPoolRef
@@ -24,15 +30,19 @@ ConnPoolMap<KEY_TYPE, POOL_TYPE>::getPool(KEY_TYPE key, const PoolFactory& facto
   if (pool_iter != active_pools_.end()) {
     return std::ref(*(pool_iter->second));
   }
-
+  Resource& connPoolResource = host_->cluster().resourceManager(priority_).connectionPools();
   // We need a new pool. Check if we have room.
-  if (max_size_.has_value() && size() >= max_size_.value()) {
+  if (!connPoolResource.canCreate()) {
     // We're full. Try to free up a pool. If we can't, bail out.
     if (!freeOnePool()) {
+      // TODO(klarose): Add some explicit counters for failure cases here, similar to the other
+      // circuit breakers.
       return absl::nullopt;
     }
 
-    ASSERT(size() < max_size_.value(), "Freeing a pool should reduce the size to below the max.");
+    ASSERT(size() < connPoolResource.max(),
+           "Freeing a pool should reduce the size to below the max.");
+
     // TODO(klarose): Consider some simple hysteresis here. How can we prevent iterating over all
     // pools when we're at the limit every time we want to allocate a new one, even if most of the
     // pools are not busy, while balancing that with not unnecessarily freeing all pools? If we
@@ -42,6 +52,7 @@ ConnPoolMap<KEY_TYPE, POOL_TYPE>::getPool(KEY_TYPE key, const PoolFactory& facto
 
   // We have room for a new pool. Allocate one and let it know about any cached callbacks.
   auto new_pool = factory();
+  connPoolResource.inc();
   for (const auto& cb : cached_callbacks_) {
     new_pool->addDrainedCallback(cb);
   }
@@ -60,8 +71,7 @@ template <typename KEY_TYPE, typename POOL_TYPE> void ConnPoolMap<KEY_TYPE, POOL
   for (auto& pool_pair : active_pools_) {
     thread_local_dispatcher_.deferredDelete(std::move(pool_pair.second));
   }
-
-  active_pools_.clear();
+  clearActivePools();
 }
 
 template <typename KEY_TYPE, typename POOL_TYPE>
@@ -96,11 +106,17 @@ bool ConnPoolMap<KEY_TYPE, POOL_TYPE>::freeOnePool() {
   if (pool_iter != active_pools_.end()) {
     // We found one. Free it up, and let the caller know.
     active_pools_.erase(pool_iter);
+    host_->cluster().resourceManager(priority_).connectionPools().dec();
     return true;
   }
 
   return false;
 }
 
+template <typename KEY_TYPE, typename POOL_TYPE>
+void ConnPoolMap<KEY_TYPE, POOL_TYPE>::clearActivePools() {
+  host_->cluster().resourceManager(priority_).connectionPools().decBy(active_pools_.size());
+  active_pools_.clear();
+}
 } // namespace Upstream
 } // namespace Envoy
diff --git a/source/common/upstream/priority_conn_pool_map.h b/source/common/upstream/priority_conn_pool_map.h
new file mode 100644
index 0000000000000..30636728d31b4
--- /dev/null
+++ b/source/common/upstream/priority_conn_pool_map.h
@@ -0,0 +1,59 @@
+#pragma once
+
+#include "envoy/event/dispatcher.h"
+#include "envoy/upstream/resource_manager.h"
+#include "envoy/upstream/upstream.h"
+
+#include "common/upstream/conn_pool_map.h"
+
+namespace Envoy {
+namespace Upstream {
+/**
+ *  A class mapping keys to connection pools, with some recycling logic built in.
+ */
+template <typename KEY_TYPE, typename POOL_TYPE> class PriorityConnPoolMap {
+public:
+  using ConnPoolMapType = ConnPoolMap<KEY_TYPE, POOL_TYPE>;
+  using PoolFactory = typename ConnPoolMapType::PoolFactory;
+  using DrainedCb = typename ConnPoolMapType::DrainedCb;
+  using OptPoolRef = typename ConnPoolMapType::OptPoolRef;
+
+  PriorityConnPoolMap(Event::Dispatcher& dispatcher, const HostConstSharedPtr& host);
+  ~PriorityConnPoolMap();
+  /**
+   * Returns an existing pool for the given priority and `key`, or creates a new one using
+   * `factory`. Note that it is possible for this to fail if a limit on the number of pools allowed
+   * is reached.
+   * @return The pool corresponding to `key`, or `absl::nullopt`.
+   */
+  OptPoolRef getPool(ResourcePriority priority, KEY_TYPE key, const PoolFactory& factory);
+
+  /**
+   * @return the number of pools across all priorities.
+   */
+  size_t size() const;
+
+  /**
+   * Destroys all mapped pools.
+   */
+  void clear();
+
+  /**
+   * Adds a drain callback to all mapped pools. Any future mapped pools with have the callback
+   * automatically added. Be careful with the callback. If it itself calls into `this`, modifying
+   * the state of `this`, there is a good chance it will cause corruption due to the callback firing
+   * immediately.
+   */
+  void addDrainedCallback(const DrainedCb& cb);
+
+  /**
+   * Instructs each connection pool to drain its connections.
+   */
+  void drainConnections();
+
+private:
+  std::array<std::unique_ptr<ConnPoolMapType>, NumResourcePriorities> conn_pool_maps_;
+};
+
+} // namespace Upstream
+} // namespace Envoy
diff --git a/source/common/upstream/priority_conn_pool_map_impl.h b/source/common/upstream/priority_conn_pool_map_impl.h
new file mode 100644
index 0000000000000..cfe1c021393bc
--- /dev/null
+++ b/source/common/upstream/priority_conn_pool_map_impl.h
@@ -0,0 +1,61 @@
+#pragma once
+
+#include "common/upstream/conn_pool_map_impl.h"
+#include "common/upstream/priority_conn_pool_map.h"
+
+namespace Envoy {
+namespace Upstream {
+
+template <typename KEY_TYPE, typename POOL_TYPE>
+PriorityConnPoolMap<KEY_TYPE, POOL_TYPE>::PriorityConnPoolMap(Envoy::Event::Dispatcher& dispatcher,
+                                                              const HostConstSharedPtr& host) {
+  for (size_t pool_map_index = 0; pool_map_index < NumResourcePriorities; ++pool_map_index) {
+    ResourcePriority priority = static_cast<ResourcePriority>(pool_map_index);
+    conn_pool_maps_[pool_map_index].reset(new ConnPoolMapType(dispatcher, host, priority));
+  }
+}
+
+template <typename KEY_TYPE, typename POOL_TYPE>
+PriorityConnPoolMap<KEY_TYPE, POOL_TYPE>::~PriorityConnPoolMap() = default;
+
+template <typename KEY_TYPE, typename POOL_TYPE>
+typename PriorityConnPoolMap<KEY_TYPE, POOL_TYPE>::OptPoolRef
+PriorityConnPoolMap<KEY_TYPE, POOL_TYPE>::getPool(ResourcePriority priority, KEY_TYPE key,
+                                                  const PoolFactory& factory) {
+  size_t index = static_cast<size_t>(priority);
+  ASSERT(index < conn_pool_maps_.size());
+  return conn_pool_maps_[index]->getPool(key, factory);
+}
+
+template <typename KEY_TYPE, typename POOL_TYPE>
+size_t PriorityConnPoolMap<KEY_TYPE, POOL_TYPE>::size() const {
+  size_t size = 0;
+  for (const auto& pool_map : conn_pool_maps_) {
+    size += pool_map->size();
+  }
+  return size;
+}
+
+template <typename KEY_TYPE, typename POOL_TYPE>
+void PriorityConnPoolMap<KEY_TYPE, POOL_TYPE>::clear() {
+  for (auto& pool_map : conn_pool_maps_) {
+    pool_map->clear();
+  }
+}
+
+template <typename KEY_TYPE, typename POOL_TYPE>
+void PriorityConnPoolMap<KEY_TYPE, POOL_TYPE>::addDrainedCallback(const DrainedCb& cb) {
+  for (auto& pool_map : conn_pool_maps_) {
+    pool_map->addDrainedCallback(cb);
+  }
+}
+
+template <typename KEY_TYPE, typename POOL_TYPE>
+void PriorityConnPoolMap<KEY_TYPE, POOL_TYPE>::drainConnections() {
+  for (auto& pool_map : conn_pool_maps_) {
+    pool_map->drainConnections();
+  }
+}
+
+} // namespace Upstream
+} // namespace Envoy
diff --git a/source/common/upstream/resource_manager_impl.h b/source/common/upstream/resource_manager_impl.h
index 4ee575a639042..a887182c63983 100644
--- a/source/common/upstream/resource_manager_impl.h
+++ b/source/common/upstream/resource_manager_impl.h
@@ -27,7 +27,7 @@ class ResourceManagerImpl : public ResourceManager {
 public:
   ResourceManagerImpl(Runtime::Loader& runtime, const std::string& runtime_key,
                       uint64_t max_connections, uint64_t max_pending_requests,
-                      uint64_t max_requests, uint64_t max_retries,
+                      uint64_t max_requests, uint64_t max_retries, uint64_t max_connection_pools,
                       ClusterCircuitBreakersStats cb_stats)
       : connections_(max_connections, runtime, runtime_key + "max_connections", cb_stats.cx_open_,
                      cb_stats.remaining_cx_),
@@ -36,13 +36,16 @@ class ResourceManagerImpl : public ResourceManager {
         requests_(max_requests, runtime, runtime_key + "max_requests", cb_stats.rq_open_,
                   cb_stats.remaining_rq_),
         retries_(max_retries, runtime, runtime_key + "max_retries", cb_stats.rq_retry_open_,
-                 cb_stats.remaining_retries_) {}
+                 cb_stats.remaining_retries_),
+        connection_pools_(max_connection_pools, runtime, runtime_key + "max_connection_pools",
+                          cb_stats.cx_pool_open_, cb_stats.remaining_cx_pools_) {}
 
   // Upstream::ResourceManager
   Resource& connections() override { return connections_; }
   Resource& pendingRequests() override { return pending_requests_; }
   Resource& requests() override { return requests_; }
   Resource& retries() override { return retries_; }
+  Resource& connectionPools() override { return connection_pools_; }
 
 private:
   struct ResourceImpl : public Resource {
@@ -61,9 +64,10 @@ class ResourceManagerImpl : public ResourceManager {
       updateRemaining();
       open_gauge_.set(canCreate() ? 0 : 1);
     }
-    void dec() override {
-      ASSERT(current_ > 0);
-      current_--;
+    void dec() override { decBy(1); }
+    void decBy(uint64_t amount) override {
+      ASSERT(current_ >= amount);
+      current_ -= amount;
       updateRemaining();
       open_gauge_.set(canCreate() ? 0 : 1);
     }
@@ -105,6 +109,7 @@ class ResourceManagerImpl : public ResourceManager {
   ResourceImpl pending_requests_;
   ResourceImpl requests_;
   ResourceImpl retries_;
+  ResourceImpl connection_pools_;
 };
 
 typedef std::unique_ptr<ResourceManagerImpl> ResourceManagerImplPtr;
diff --git a/source/common/upstream/upstream_impl.cc b/source/common/upstream/upstream_impl.cc
index ef436d8a084c7..301fb23536213 100644
--- a/source/common/upstream/upstream_impl.cc
+++ b/source/common/upstream/upstream_impl.cc
@@ -2,6 +2,7 @@
 
 #include <chrono>
 #include <cstdint>
+#include <limits>
 #include <list>
 #include <memory>
 #include <string>
@@ -846,6 +847,7 @@ ClusterInfoImpl::ResourceManagers::load(const envoy::api::v2::Cluster& config,
   uint64_t max_pending_requests = 1024;
   uint64_t max_requests = 1024;
   uint64_t max_retries = 3;
+  uint64_t max_connection_pools = std::numeric_limits<uint64_t>::max();
 
   bool track_remaining = false;
 
@@ -877,9 +879,12 @@ ClusterInfoImpl::ResourceManagers::load(const envoy::api::v2::Cluster& config,
     max_requests = PROTOBUF_GET_WRAPPED_OR_DEFAULT(*it, max_requests, max_requests);
     max_retries = PROTOBUF_GET_WRAPPED_OR_DEFAULT(*it, max_retries, max_retries);
     track_remaining = it->track_remaining();
+    max_connection_pools =
+        PROTOBUF_GET_WRAPPED_OR_DEFAULT(*it, max_connection_pools, max_connection_pools);
   }
   return std::make_unique<ResourceManagerImpl>(
       runtime, runtime_prefix, max_connections, max_pending_requests, max_requests, max_retries,
+      max_connection_pools,
       ClusterInfoImpl::generateCircuitBreakersStats(stats_scope, priority_name, track_remaining));
 }
 
diff --git a/test/common/http/http1/conn_pool_test.cc b/test/common/http/http1/conn_pool_test.cc
index c66a2d9a41d55..27324c4a6dbf6 100644
--- a/test/common/http/http1/conn_pool_test.cc
+++ b/test/common/http/http1/conn_pool_test.cc
@@ -207,7 +207,7 @@ struct ActiveTestRequest {
  * Verify that connections are drained when requested.
  */
 TEST_F(Http1ConnPoolImplTest, DrainConnections) {
-  cluster_->resetResourceManager(2, 1024, 1024, 1);
+  cluster_->resetResourceManager(2, 1024, 1024, 1, 1);
   InSequence s;
 
   ActiveTestRequest r1(*this, 0, ActiveTestRequest::Type::CreateConnection);
@@ -292,7 +292,7 @@ TEST_F(Http1ConnPoolImplTest, MultipleRequestAndResponse) {
  * Test when we overflow max pending requests.
  */
 TEST_F(Http1ConnPoolImplTest, MaxPendingRequests) {
-  cluster_->resetResourceManager(1, 1, 1024, 1);
+  cluster_->resetResourceManager(1, 1, 1024, 1, 1);
 
   EXPECT_EQ(0U, cluster_->circuit_breakers_stats_.rq_pending_open_.value());
 
@@ -614,7 +614,7 @@ TEST_F(Http1ConnPoolImplTest, MaxRequestsPerConnection) {
 }
 
 TEST_F(Http1ConnPoolImplTest, ConcurrentConnections) {
-  cluster_->resetResourceManager(2, 1024, 1024, 1);
+  cluster_->resetResourceManager(2, 1024, 1024, 1, 1);
   InSequence s;
 
   ActiveTestRequest r1(*this, 0, ActiveTestRequest::Type::CreateConnection);
diff --git a/test/common/http/http2/conn_pool_test.cc b/test/common/http/http2/conn_pool_test.cc
index 554b873fdea8f..c7b5577154a36 100644
--- a/test/common/http/http2/conn_pool_test.cc
+++ b/test/common/http/http2/conn_pool_test.cc
@@ -609,7 +609,7 @@ TEST_F(Http2ConnPoolImplTest, ConnectTimeout) {
 }
 
 TEST_F(Http2ConnPoolImplTest, MaxGlobalRequests) {
-  cluster_->resetResourceManager(1024, 1024, 1, 1);
+  cluster_->resetResourceManager(1024, 1024, 1, 1, 1);
   InSequence s;
 
   expectClientCreate();
diff --git a/test/common/router/retry_state_impl_test.cc b/test/common/router/retry_state_impl_test.cc
index b126f70cca3f1..813daa4038b08 100644
--- a/test/common/router/retry_state_impl_test.cc
+++ b/test/common/router/retry_state_impl_test.cc
@@ -424,7 +424,7 @@ TEST_F(RouterRetryStateImplTest, RouteConfigNoHeaderConfig) {
 }
 
 TEST_F(RouterRetryStateImplTest, NoAvailableRetries) {
-  cluster_.resetResourceManager(0, 0, 0, 0);
+  cluster_.resetResourceManager(0, 0, 0, 0, 0);
 
   Http::TestHeaderMapImpl request_headers{{"x-envoy-retry-on", "connect-failure"}};
   setup(request_headers);
diff --git a/test/common/tcp/conn_pool_test.cc b/test/common/tcp/conn_pool_test.cc
index 034e4098dbaca..149354109c055 100644
--- a/test/common/tcp/conn_pool_test.cc
+++ b/test/common/tcp/conn_pool_test.cc
@@ -271,7 +271,7 @@ struct ActiveTestConn {
  * Verify that connections are drained when requested.
  */
 TEST_F(TcpConnPoolImplTest, DrainConnections) {
-  cluster_->resetResourceManager(3, 1024, 1024, 1);
+  cluster_->resetResourceManager(3, 1024, 1024, 1, 1);
   InSequence s;
 
   ActiveTestConn c1(*this, 0, ActiveTestConn::Type::CreateConnection);
@@ -484,7 +484,7 @@ TEST_F(TcpConnPoolImplTest, ConnectionStateLifecycle) {
  * Test when we overflow max pending requests.
  */
 TEST_F(TcpConnPoolImplTest, MaxPendingRequests) {
-  cluster_->resetResourceManager(1, 1, 1024, 1);
+  cluster_->resetResourceManager(1, 1, 1024, 1, 1);
 
   ConnPoolCallbacks callbacks;
   conn_pool_.expectConnCreate();
@@ -657,7 +657,7 @@ TEST_F(TcpConnPoolImplTest, DisconnectWhileBound) {
  * Test upstream disconnection of one request while another is pending.
  */
 TEST_F(TcpConnPoolImplTest, DisconnectWhilePending) {
-  cluster_->resetResourceManager(1, 1024, 1024, 1);
+  cluster_->resetResourceManager(1, 1024, 1024, 1, 1);
   InSequence s;
 
   // First request connected.
@@ -767,7 +767,7 @@ TEST_F(TcpConnPoolImplTest, MaxRequestsPerConnection) {
  * Test that multiple connections can be assigned at once.
  */
 TEST_F(TcpConnPoolImplTest, ConcurrentConnections) {
-  cluster_->resetResourceManager(2, 1024, 1024, 1);
+  cluster_->resetResourceManager(2, 1024, 1024, 1, 1);
   InSequence s;
 
   ActiveTestConn c1(*this, 0, ActiveTestConn::Type::CreateConnection);
@@ -803,7 +803,7 @@ TEST_F(TcpConnPoolImplTest, ConnectionStateWithConcurrentConnections) {
   auto* s2 = new TestConnectionState(2, [&]() -> void { state_destroyed |= 2; });
   auto* s3 = new TestConnectionState(2, [&]() -> void { state_destroyed |= 4; });
 
-  cluster_->resetResourceManager(2, 1024, 1024, 1);
+  cluster_->resetResourceManager(2, 1024, 1024, 1, 1);
   ActiveTestConn c1(*this, 0, ActiveTestConn::Type::CreateConnection);
   c1.callbacks_.conn_data_->setConnectionState(std::unique_ptr<TestConnectionState>(s1));
   ActiveTestConn c2(*this, 1, ActiveTestConn::Type::CreateConnection);
diff --git a/test/common/tcp_proxy/tcp_proxy_test.cc b/test/common/tcp_proxy/tcp_proxy_test.cc
index 27a056340fe15..ecf3fb691b6cf 100644
--- a/test/common/tcp_proxy/tcp_proxy_test.cc
+++ b/test/common/tcp_proxy/tcp_proxy_test.cc
@@ -799,7 +799,7 @@ TEST_F(TcpProxyTest, UpstreamConnectFailure) {
 TEST_F(TcpProxyTest, UpstreamConnectionLimit) {
   configure(accessLogConfig("%RESPONSE_FLAGS%"));
   factory_context_.cluster_manager_.thread_local_cluster_.cluster_.info_->resetResourceManager(
-      0, 0, 0, 0);
+      0, 0, 0, 0, 0);
 
   // setup sets up expectation for tcpConnForCluster but this test is expected to NOT call that
   filter_ = std::make_unique<Filter>(config_, factory_context_.cluster_manager_, timeSystem());
diff --git a/test/common/upstream/BUILD b/test/common/upstream/BUILD
index 2653380f0070c..b65aa9ef552ab 100644
--- a/test/common/upstream/BUILD
+++ b/test/common/upstream/BUILD
@@ -71,6 +71,7 @@ envoy_cc_test(
         "//test/mocks:common_lib",
         "//test/mocks/event:event_mocks",
         "//test/mocks/http:conn_pool_mocks",
+        "//test/mocks/upstream:host_mocks",
         "//test/test_common:utility_lib",
     ],
 )
@@ -267,6 +268,20 @@ envoy_cc_test(
     ],
 )
 
+envoy_cc_test(
+    name = "priority_conn_pool_map_impl_test",
+    srcs = ["priority_conn_pool_map_impl_test.cc"],
+    deps = [
+        "//include/envoy/http:conn_pool_interface",
+        "//source/common/upstream:priority_conn_pool_map_impl_lib",
+        "//test/mocks:common_lib",
+        "//test/mocks/event:event_mocks",
+        "//test/mocks/http:conn_pool_mocks",
+        "//test/mocks/upstream:host_mocks",
+        "//test/test_common:utility_lib",
+    ],
+)
+
 envoy_cc_test(
     name = "resource_manager_impl_test",
     srcs = ["resource_manager_impl_test.cc"],
diff --git a/test/common/upstream/cluster_manager_impl_test.cc b/test/common/upstream/cluster_manager_impl_test.cc
index 5731e0c403c68..a15b364c8b2d6 100644
--- a/test/common/upstream/cluster_manager_impl_test.cc
+++ b/test/common/upstream/cluster_manager_impl_test.cc
@@ -1688,7 +1688,6 @@ TEST_F(ClusterManagerImplTest, DynamicHostRemove) {
   // drain callbacks, etc.
   dns_timer_->callback_();
   dns_callback(TestUtility::makeDnsResponse({"127.0.0.2", "127.0.0.3"}));
-
   factory_.tls_.shutdownThread();
 }
 
@@ -1915,7 +1914,6 @@ TEST_F(ClusterManagerImplTest, DynamicHostRemoveWithTls) {
   // drain callbacks, etc.
   dns_timer_->callback_();
   dns_callback(TestUtility::makeDnsResponse({"127.0.0.2", "127.0.0.3"}));
-
   factory_.tls_.shutdownThread();
 }
 
diff --git a/test/common/upstream/conn_pool_map_impl_test.cc b/test/common/upstream/conn_pool_map_impl_test.cc
index 93f5a7cc6ed7e..b8183d2eae29b 100644
--- a/test/common/upstream/conn_pool_map_impl_test.cc
+++ b/test/common/upstream/conn_pool_map_impl_test.cc
@@ -8,11 +8,13 @@
 #include "test/mocks/common.h"
 #include "test/mocks/event/mocks.h"
 #include "test/mocks/http/conn_pool.h"
+#include "test/mocks/upstream/host.h"
 #include "test/test_common/utility.h"
 
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
 
+using testing::AtLeast;
 using testing::Invoke;
 using testing::InvokeArgument;
 using testing::NiceMock;
@@ -30,10 +32,17 @@ class ConnPoolMapImplTest : public testing::Test {
   using TestMap = ConnPoolMap<int, Http::ConnectionPool::Instance>;
   using TestMapPtr = std::unique_ptr<TestMap>;
 
-  TestMapPtr makeTestMap() { return std::make_unique<TestMap>(dispatcher_, absl::nullopt); }
+  TestMapPtr makeTestMap() {
+    return std::make_unique<TestMap>(dispatcher_, host_, ResourcePriority::Default);
+  }
 
   TestMapPtr makeTestMapWithLimit(uint64_t limit) {
-    return std::make_unique<TestMap>(dispatcher_, absl::make_optional(limit));
+    return makeTestMapWithLimitAtPriority(limit, ResourcePriority::Default);
+  }
+
+  TestMapPtr makeTestMapWithLimitAtPriority(uint64_t limit, ResourcePriority priority) {
+    host_->cluster_.resetResourceManager(1024, 1024, 1024, 1024, limit);
+    return std::make_unique<TestMap>(dispatcher_, host_, priority);
   }
 
   TestMap::PoolFactory getBasicFactory() {
@@ -73,6 +82,7 @@ class ConnPoolMapImplTest : public testing::Test {
 protected:
   NiceMock<Event::MockDispatcher> dispatcher_;
   std::vector<NiceMock<Http::ConnectionPool::MockInstance>*> mock_pools_;
+  std::shared_ptr<NiceMock<MockHost>> host_ = std::make_shared<NiceMock<MockHost>>();
 };
 
 TEST_F(ConnPoolMapImplTest, TestMapIsEmptyOnConstruction) {
@@ -295,6 +305,58 @@ TEST_F(ConnPoolMapImplTest, GetPoolFailStateIsCleared) {
   EXPECT_EQ(test_map->size(), 2);
 }
 
+TEST_F(ConnPoolMapImplTest, CircuitBreakerNotSetOnClear) {
+  TestMapPtr test_map = makeTestMapWithLimit(1);
+
+  test_map->getPool(1, getBasicFactory());
+  test_map->getPool(2, getBasicFactory());
+  test_map->getPool(3, getBasicFactory());
+
+  test_map->clear();
+
+  EXPECT_EQ(host_->cluster_.circuit_breakers_stats_.cx_pool_open_.value(), 0);
+}
+
+TEST_F(ConnPoolMapImplTest, CircuitBreakerSetAtLimit) {
+  TestMapPtr test_map = makeTestMapWithLimit(2);
+
+  test_map->getPool(1, getBasicFactory());
+  test_map->getPool(2, getBasicFactory());
+
+  EXPECT_EQ(host_->cluster_.circuit_breakers_stats_.cx_pool_open_.value(), 1);
+}
+
+TEST_F(ConnPoolMapImplTest, CircuitBreakerClearedOnDestroy) {
+  {
+    TestMapPtr test_map = makeTestMapWithLimit(2);
+
+    test_map->getPool(1, getBasicFactory());
+    test_map->getPool(2, getBasicFactory());
+  }
+
+  EXPECT_EQ(host_->cluster_.circuit_breakers_stats_.cx_pool_open_.value(), 0);
+}
+
+TEST_F(ConnPoolMapImplTest, CircuitBreakerUsesProvidedPriorityDefault) {
+  TestMapPtr test_map = makeTestMapWithLimitAtPriority(2, ResourcePriority::Default);
+
+  EXPECT_CALL(host_->cluster_, resourceManager(ResourcePriority::High)).Times(0);
+  EXPECT_CALL(host_->cluster_, resourceManager(ResourcePriority::Default)).Times(AtLeast(1));
+
+  test_map->getPool(1, getBasicFactory());
+  test_map->getPool(2, getBasicFactory());
+}
+
+TEST_F(ConnPoolMapImplTest, CircuitBreakerUsesProvidedPriorityHigh) {
+  TestMapPtr test_map = makeTestMapWithLimitAtPriority(2, ResourcePriority::High);
+
+  EXPECT_CALL(host_->cluster_, resourceManager(ResourcePriority::High)).Times(AtLeast(1));
+  EXPECT_CALL(host_->cluster_, resourceManager(ResourcePriority::Default)).Times(0);
+
+  test_map->getPool(1, getBasicFactory());
+  test_map->getPool(2, getBasicFactory());
+}
+
 // The following tests only die in debug builds, so don't run them if this isn't one.
 #if !defined(NDEBUG)
 class ConnPoolMapImplDeathTest : public ConnPoolMapImplTest {};
diff --git a/test/common/upstream/priority_conn_pool_map_impl_test.cc b/test/common/upstream/priority_conn_pool_map_impl_test.cc
new file mode 100644
index 0000000000000..fbe98e4d17e1c
--- /dev/null
+++ b/test/common/upstream/priority_conn_pool_map_impl_test.cc
@@ -0,0 +1,142 @@
+#include <memory>
+
+#include "envoy/http/conn_pool.h"
+
+#include "common/upstream/priority_conn_pool_map_impl.h"
+
+#include "test/mocks/common.h"
+#include "test/mocks/event/mocks.h"
+#include "test/mocks/http/conn_pool.h"
+#include "test/mocks/upstream/host.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+using testing::AtLeast;
+using testing::Return;
+using testing::SaveArg;
+
+namespace Envoy {
+namespace Upstream {
+namespace {
+
+class PriorityConnPoolMapImplTest : public testing::Test {
+public:
+  using TestMap = PriorityConnPoolMap<int, Http::ConnectionPool::Instance>;
+  using TestMapPtr = std::unique_ptr<TestMap>;
+
+  TestMapPtr makeTestMap() { return std::make_unique<TestMap>(dispatcher_, host_); }
+
+  TestMap::PoolFactory getBasicFactory() {
+    return [&]() {
+      auto pool = std::make_unique<NiceMock<Http::ConnectionPool::MockInstance>>();
+      ON_CALL(*pool, hasActiveConnections).WillByDefault(Return(false));
+      mock_pools_.push_back(pool.get());
+      return pool;
+    };
+  }
+
+protected:
+  NiceMock<Event::MockDispatcher> dispatcher_;
+  std::vector<NiceMock<Http::ConnectionPool::MockInstance>*> mock_pools_;
+  std::shared_ptr<NiceMock<MockHost>> host_ = std::make_shared<NiceMock<MockHost>>();
+};
+
+// Show that we return a non-null value, and that we invoke the default resource manager
+TEST_F(PriorityConnPoolMapImplTest, DefaultPriorityProxiedThrough) {
+  TestMapPtr test_map = makeTestMap();
+
+  EXPECT_CALL(host_->cluster_, resourceManager(ResourcePriority::High)).Times(0);
+  EXPECT_CALL(host_->cluster_, resourceManager(ResourcePriority::Default)).Times(AtLeast(1));
+
+  auto pool = test_map->getPool(ResourcePriority::Default, 0, getBasicFactory());
+  EXPECT_TRUE(pool.has_value());
+
+  // At this point, we may clean up/decrement by 0, etc, so allow any number.
+  EXPECT_CALL(host_->cluster_, resourceManager(ResourcePriority::High)).Times(AtLeast(1));
+}
+
+// Show that we return a non-null value, and that we invoke the high resource manager
+TEST_F(PriorityConnPoolMapImplTest, HighPriorityProxiedThrough) {
+  TestMapPtr test_map = makeTestMap();
+
+  EXPECT_CALL(host_->cluster_, resourceManager(ResourcePriority::Default)).Times(0);
+  EXPECT_CALL(host_->cluster_, resourceManager(ResourcePriority::High)).Times(AtLeast(1));
+
+  auto pool = test_map->getPool(ResourcePriority::High, 0, getBasicFactory());
+  EXPECT_TRUE(pool.has_value());
+
+  // At this point, we may clean up/decrement by 0, etc, so allow any number.
+  EXPECT_CALL(host_->cluster_, resourceManager(ResourcePriority::Default)).Times(AtLeast(1));
+}
+
+TEST_F(PriorityConnPoolMapImplTest, TestSizeForSinglePriority) {
+  TestMapPtr test_map = makeTestMap();
+
+  test_map->getPool(ResourcePriority::High, 0, getBasicFactory());
+  test_map->getPool(ResourcePriority::High, 1, getBasicFactory());
+
+  EXPECT_EQ(test_map->size(), 2);
+}
+
+TEST_F(PriorityConnPoolMapImplTest, TestSizeForMultiplePriorities) {
+  TestMapPtr test_map = makeTestMap();
+
+  test_map->getPool(ResourcePriority::High, 0, getBasicFactory());
+  test_map->getPool(ResourcePriority::High, 1, getBasicFactory());
+  test_map->getPool(ResourcePriority::Default, 0, getBasicFactory());
+  test_map->getPool(ResourcePriority::Default, 1, getBasicFactory());
+  test_map->getPool(ResourcePriority::Default, 2, getBasicFactory());
+
+  EXPECT_EQ(test_map->size(), 5);
+}
+
+TEST_F(PriorityConnPoolMapImplTest, TestClearEmptiesOut) {
+  TestMapPtr test_map = makeTestMap();
+
+  test_map->getPool(ResourcePriority::High, 0, getBasicFactory());
+  test_map->getPool(ResourcePriority::High, 1, getBasicFactory());
+  test_map->getPool(ResourcePriority::Default, 0, getBasicFactory());
+  test_map->getPool(ResourcePriority::Default, 1, getBasicFactory());
+  test_map->getPool(ResourcePriority::Default, 2, getBasicFactory());
+  test_map->clear();
+
+  EXPECT_EQ(test_map->size(), 0);
+}
+
+// Show that the drained callback is invoked once for the high priority pool, and once for
+// the default priority pool.
+TEST_F(PriorityConnPoolMapImplTest, TestAddDrainedCbProxiedThrough) {
+  TestMapPtr test_map = makeTestMap();
+
+  test_map->getPool(ResourcePriority::High, 0, getBasicFactory());
+  test_map->getPool(ResourcePriority::Default, 0, getBasicFactory());
+
+  Http::ConnectionPool::Instance::DrainedCb cbHigh;
+  EXPECT_CALL(*mock_pools_[0], addDrainedCallback(_)).WillOnce(SaveArg<0>(&cbHigh));
+  Http::ConnectionPool::Instance::DrainedCb cbDefault;
+  EXPECT_CALL(*mock_pools_[1], addDrainedCallback(_)).WillOnce(SaveArg<0>(&cbDefault));
+
+  ReadyWatcher watcher;
+  test_map->addDrainedCallback([&watcher] { watcher.ready(); });
+
+  EXPECT_CALL(watcher, ready()).Times(2);
+  cbHigh();
+  cbDefault();
+}
+
+TEST_F(PriorityConnPoolMapImplTest, TestDrainConnectionsProxiedThrough) {
+  TestMapPtr test_map = makeTestMap();
+
+  test_map->getPool(ResourcePriority::High, 0, getBasicFactory());
+  test_map->getPool(ResourcePriority::Default, 0, getBasicFactory());
+
+  EXPECT_CALL(*mock_pools_[0], drainConnections());
+  EXPECT_CALL(*mock_pools_[1], drainConnections());
+
+  test_map->drainConnections();
+}
+
+} // namespace
+} // namespace Upstream
+} // namespace Envoy
diff --git a/test/common/upstream/resource_manager_impl_test.cc b/test/common/upstream/resource_manager_impl_test.cc
index 0e9c6e1eb3bc2..118a453fb8519 100644
--- a/test/common/upstream/resource_manager_impl_test.cc
+++ b/test/common/upstream/resource_manager_impl_test.cc
@@ -25,7 +25,7 @@ TEST(ResourceManagerImplTest, RuntimeResourceManager) {
   ON_CALL(store, gauge(_)).WillByDefault(ReturnRef(gauge));
 
   ResourceManagerImpl resource_manager(
-      runtime, "circuit_breakers.runtime_resource_manager_test.default.", 0, 0, 0, 1,
+      runtime, "circuit_breakers.runtime_resource_manager_test.default.", 0, 0, 0, 1, 0,
       ClusterCircuitBreakersStats{
           ALL_CLUSTER_CIRCUIT_BREAKERS_STATS(POOL_GAUGE(store), POOL_GAUGE(store))});
 
@@ -58,6 +58,13 @@ TEST(ResourceManagerImplTest, RuntimeResourceManager) {
       .WillRepeatedly(Return(0U));
   EXPECT_EQ(0U, resource_manager.retries().max());
   EXPECT_FALSE(resource_manager.retries().canCreate());
+  EXPECT_CALL(
+      runtime.snapshot_,
+      getInteger("circuit_breakers.runtime_resource_manager_test.default.max_connection_pools", 0U))
+      .Times(2)
+      .WillRepeatedly(Return(5U));
+  EXPECT_EQ(5U, resource_manager.connectionPools().max());
+  EXPECT_TRUE(resource_manager.connectionPools().canCreate());
 }
 
 TEST(ResourceManagerImplTest, RemainingResourceGauges) {
@@ -67,7 +74,7 @@ TEST(ResourceManagerImplTest, RemainingResourceGauges) {
   auto stats = ClusterCircuitBreakersStats{
       ALL_CLUSTER_CIRCUIT_BREAKERS_STATS(POOL_GAUGE(store), POOL_GAUGE(store))};
   ResourceManagerImpl resource_manager(
-      runtime, "circuit_breakers.runtime_resource_manager_test.default.", 1, 2, 1, 0, stats);
+      runtime, "circuit_breakers.runtime_resource_manager_test.default.", 1, 2, 1, 0, 3, stats);
 
   // Test remaining_cx_ gauge
   EXPECT_EQ(1U, resource_manager.connections().max());
@@ -104,7 +111,14 @@ TEST(ResourceManagerImplTest, RemainingResourceGauges) {
   resource_manager.retries().inc();
   EXPECT_EQ(0U, stats.remaining_retries_.value());
   resource_manager.retries().dec();
-  EXPECT_EQ(0U, stats.remaining_retries_.value());
+
+  // Test remaining_cx_pools gauge.
+  EXPECT_EQ(3U, resource_manager.connectionPools().max());
+  EXPECT_EQ(3U, stats.remaining_cx_pools_.value());
+  resource_manager.connectionPools().inc();
+  EXPECT_EQ(2U, stats.remaining_cx_pools_.value());
+  resource_manager.connectionPools().dec();
+  EXPECT_EQ(3U, stats.remaining_cx_pools_.value());
 }
 } // namespace
 } // namespace Upstream
diff --git a/test/integration/integration_test.cc b/test/integration/integration_test.cc
index 0141026a9478c..1cb548c9b9ae0 100644
--- a/test/integration/integration_test.cc
+++ b/test/integration/integration_test.cc
@@ -749,6 +749,34 @@ TEST_P(IntegrationTest, TestDelayedConnectionTeardownTimeoutTrigger) {
             1);
 }
 
+// Test that if no connection pools are free, Envoy fails to establish an upstream connection.
+TEST_P(IntegrationTest, NoConnectionPoolsFree) {
+  config_helper_.addConfigModifier([](envoy::config::bootstrap::v2::Bootstrap& bootstrap) {
+    auto* static_resources = bootstrap.mutable_static_resources();
+    auto* cluster = static_resources->mutable_clusters(0);
+
+    // Somewhat contrived with 0, but this is the simplest way to test right now.
+    auto* circuit_breakers = cluster->mutable_circuit_breakers();
+    circuit_breakers->add_thresholds()->mutable_max_connection_pools()->set_value(0);
+  });
+
+  initialize();
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  // Request 1.
+  auto response = codec_client_->makeRequestWithBody(default_request_headers_, 1024);
+
+  // Validate none active.
+  test_server_->waitForGaugeEq("cluster.cluster_0.upstream_rq_active", 0);
+  test_server_->waitForGaugeEq("cluster.cluster_0.upstream_rq_pending_active", 0);
+
+  response->waitForEndStream();
+
+  EXPECT_STREQ("503", response->headers().Status()->value().c_str());
+  test_server_->waitForCounterGe("cluster.cluster_0.upstream_rq_503", 1);
+}
+
 INSTANTIATE_TEST_SUITE_P(IpVersions, UpstreamEndpointIntegrationTest,
                          testing::ValuesIn(TestEnvironment::getIpVersionsForTest()),
                          TestUtility::ipTestParamsToString);
diff --git a/test/mocks/upstream/cluster_info.cc b/test/mocks/upstream/cluster_info.cc
index a0ef49d23331b..5497a0cd2bbe2 100644
--- a/test/mocks/upstream/cluster_info.cc
+++ b/test/mocks/upstream/cluster_info.cc
@@ -1,5 +1,7 @@
 #include "test/mocks/upstream/cluster_info.h"
 
+#include <limits>
+
 #include "common/network/raw_buffer_socket.h"
 #include "common/upstream/upstream_impl.h"
 
@@ -35,6 +37,7 @@ MockClusterInfo::MockClusterInfo()
       circuit_breakers_stats_(
           ClusterInfoImpl::generateCircuitBreakersStats(stats_store_, "default", true)),
       resource_manager_(new Upstream::ResourceManagerImpl(runtime_, "fake_key", 1, 1024, 1024, 1,
+                                                          std::numeric_limits<uint64_t>::max(),
                                                           circuit_breakers_stats_)) {
   ON_CALL(*this, connectTimeout()).WillByDefault(Return(std::chrono::milliseconds(1)));
   ON_CALL(*this, idleTimeout()).WillByDefault(Return(absl::optional<std::chrono::milliseconds>()));
diff --git a/test/mocks/upstream/cluster_info.h b/test/mocks/upstream/cluster_info.h
index d52c58b00221a..dd498b8fae26c 100644
--- a/test/mocks/upstream/cluster_info.h
+++ b/test/mocks/upstream/cluster_info.h
@@ -46,9 +46,10 @@ class MockClusterInfo : public ClusterInfo {
   MockClusterInfo();
   ~MockClusterInfo();
 
-  void resetResourceManager(uint64_t cx, uint64_t rq_pending, uint64_t rq, uint64_t rq_retry) {
-    resource_manager_ = std::make_unique<ResourceManagerImpl>(runtime_, name_, cx, rq_pending, rq,
-                                                              rq_retry, circuit_breakers_stats_);
+  void resetResourceManager(uint64_t cx, uint64_t rq_pending, uint64_t rq, uint64_t rq_retry,
+                            uint64_t conn_pool) {
+    resource_manager_ = std::make_unique<ResourceManagerImpl>(
+        runtime_, name_, cx, rq_pending, rq, rq_retry, conn_pool, circuit_breakers_stats_);
   }
 
   // Upstream::ClusterInfo