diff --git a/api/envoy/api/v2/srds.proto b/api/envoy/api/v2/srds.proto
index 9038cb1e3257..617fdf9ac644 100644
--- a/api/envoy/api/v2/srds.proto
+++ b/api/envoy/api/v2/srds.proto
@@ -2,36 +2,27 @@ syntax = "proto3";
 
 package envoy.api.v2;
 
-option java_outer_classname = "SrdsProto";
-option java_package = "io.envoyproxy.envoy.api.v2";
-option java_multiple_files = true;
-option java_generic_services = true;
-
 import "envoy/api/v2/discovery.proto";
-
+import "gogoproto/gogo.proto";
 import "google/api/annotations.proto";
-
 import "validate/validate.proto";
-import "gogoproto/gogo.proto";
 
+option java_outer_classname = "SrdsProto";
+option java_package = "io.envoyproxy.envoy.api.v2";
+option java_multiple_files = true;
+option java_generic_services = true;
 option (gogoproto.equal_all) = true;
 
 // [#protodoc-title: HTTP scoped routing configuration]
 // * Routing :ref:`architecture overview <arch_overview_http_routing>`
 //
-// .. attention::
-//
-//   The Scoped RDS API is not yet fully implemented and *should not* be enabled in
-//   :ref:`envoy_api_msg_config.filter.network.http_connection_manager.v2.HttpConnectionManager`.
-//
-// TODO(AndresGuedez): Update :ref:`arch_overview_http_routing` with scoped routing overview and
-// configuration details.
-
 // The Scoped Routes Discovery Service (SRDS) API distributes
-// :ref:`ScopedRouteConfiguration<envoy_api_msg.ScopedRouteConfiguration>` resources. Each
-// ScopedRouteConfiguration resource represents a "routing scope" containing a mapping that allows
-// the HTTP connection manager to dynamically assign a routing table (specified via
-// a :ref:`RouteConfiguration<envoy_api_msg_RouteConfiguration>` message) to each HTTP request.
+// :ref:`ScopedRouteConfiguration<envoy_api_msg.ScopedRouteConfiguration>`
+// resources. Each ScopedRouteConfiguration resource represents a "routing
+// scope" containing a mapping that allows the HTTP connection manager to
+// dynamically assign a routing table (specified via a
+// :ref:`RouteConfiguration<envoy_api_msg_RouteConfiguration>` message) to each
+// HTTP request.
 // [#proto-status: experimental]
 service ScopedRoutesDiscoveryService {
   rpc StreamScopedRoutes(stream DiscoveryRequest) returns (stream DiscoveryResponse) {
@@ -52,9 +43,9 @@ service ScopedRoutesDiscoveryService {
 // :ref:`Key<envoy_api_msg_ScopedRouteConfiguration.Key>` to a
 // :ref:`envoy_api_msg_RouteConfiguration` (identified by its resource name).
 //
-// The HTTP connection manager builds up a table consisting of these Key to RouteConfiguration
-// mappings, and looks up the RouteConfiguration to use per request according to the algorithm
-// specified in the
+// The HTTP connection manager builds up a table consisting of these Key to
+// RouteConfiguration mappings, and looks up the RouteConfiguration to use per
+// request according to the algorithm specified in the
 // :ref:`scope_key_builder<envoy_api_field_config.filter.network.http_connection_manager.v2.ScopedRoutes.scope_key_builder>`
 // assigned to the HttpConnectionManager.
 //
@@ -104,8 +95,8 @@ service ScopedRoutesDiscoveryService {
 //     Host: foo.com
 //     X-Route-Selector: vip=172.10.10.20
 //
-// would result in the routing table defined by the `route-config1` RouteConfiguration being
-// assigned to the HTTP request/stream.
+// would result in the routing table defined by the `route-config1`
+// RouteConfiguration being assigned to the HTTP request/stream.
 //
 // [#comment:next free field: 4]
 // [#proto-status: experimental]
@@ -115,8 +106,9 @@ message ScopedRouteConfiguration {
 
   // Specifies a key which is matched against the output of the
   // :ref:`scope_key_builder<envoy_api_field_config.filter.network.http_connection_manager.v2.ScopedRoutes.scope_key_builder>`
-  // specified in the HttpConnectionManager. The matching is done per HTTP request and is dependent
-  // on the order of the fragments contained in the Key.
+  // specified in the HttpConnectionManager. The matching is done per HTTP
+  // request and is dependent on the order of the fragments contained in the
+  // Key.
   message Key {
     message Fragment {
       oneof type {
@@ -127,14 +119,15 @@ message ScopedRouteConfiguration {
       }
     }
 
-    // The ordered set of fragments to match against. The order must match the fragments in the
-    // corresponding
+    // The ordered set of fragments to match against. The order must match the
+    // fragments in the corresponding
     // :ref:`scope_key_builder<envoy_api_field_config.filter.network.http_connection_manager.v2.ScopedRoutes.scope_key_builder>`.
     repeated Fragment fragments = 1 [(validate.rules).repeated .min_items = 1];
   }
 
-  // The resource name to use for a :ref:`envoy_api_msg_DiscoveryRequest` to an RDS server to
-  // fetch the :ref:`envoy_api_msg_RouteConfiguration` associated with this scope.
+  // The resource name to use for a :ref:`envoy_api_msg_DiscoveryRequest` to an
+  // RDS server to fetch the :ref:`envoy_api_msg_RouteConfiguration` associated
+  // with this scope.
   string route_configuration_name = 2 [(validate.rules).string.min_bytes = 1];
 
   // The key to match against.
diff --git a/docs/root/intro/arch_overview/http/http_routing.rst b/docs/root/intro/arch_overview/http/http_routing.rst
index 6a191be26821..574efa611ecf 100644
--- a/docs/root/intro/arch_overview/http/http_routing.rst
+++ b/docs/root/intro/arch_overview/http/http_routing.rst
@@ -50,6 +50,35 @@ request. The router filter supports the following features:
 * :ref:`Hash policy <envoy_api_field_route.RouteAction.hash_policy>` based routing.
 * :ref:`Absolute urls <envoy_api_field_config.filter.network.http_connection_manager.v2.HttpConnectionManager.http_protocol_options>` are supported for non-tls forward proxies.
 
+.. _arch_overview_http_routing_route_scope:
+
+Route Scope
+--------------
+
+Scoped routing enables Envoy to put constraints on search space of domains and route rules.
+A :ref:`Route Scope<envoy_api_msg_ScopedRouteConfiguration>` associates a key with a :ref:`route table <arch_overview_http_routing_route_table>`.
+For each request, a scope key is computed dynamically by the HTTP connection manager to pick the :ref:`route table<envoy_api_msg_RouteConfiguration>`.
+
+The Scoped RDS (SRDS) API contains a set of :ref:`Scopes <envoy_api_msg_ScopedRouteConfiguration>` resources, each defining independent routing configuration,
+along with a :ref:`ScopeKeyBuilder <envoy_api_msg_config.filter.network.http_connection_manager.v2.ScopedRoutes.ScopeKeyBuilder>`
+defining the key construction algorithm used by Envoy to look up the scope corresponding to each request.
+
+For example, for the following scoped route configuration, Envoy will look into the "addr" header value, split the header value by ";" first, and use the first value for key 'x-foo-key' as the scope key.
+If the "addr" header value is "foo=1;x-foo-key=127.0.0.1;x-bar-key=1.1.1.1", then "127.0.0.1" will be computed as the scope key to look up for corresponding route configuration.
+
+.. code-block:: yaml
+
+  name: scope_by_addr
+  fragments:
+    - header_value_extractor:
+        name: Addr
+        element_separator: ;
+        element:
+          key: x-foo-key
+          separator: =
+
+.. _arch_overview_http_routing_route_table:
+
 Route table
 -----------
 
diff --git a/docs/root/intro/version_history.rst b/docs/root/intro/version_history.rst
index c4e6c2f2ac52..653bc4f65d6d 100644
--- a/docs/root/intro/version_history.rst
+++ b/docs/root/intro/version_history.rst
@@ -43,16 +43,17 @@ Version history
   <deprecated>` for more information.
 * rbac: added conditions to the policy, see :ref:`condition <envoy_api_field_config.rbac.v2.Policy.condition>`.
 * router: added :ref:`rq_retry_skipped_request_not_complete <config_http_filters_router_stats>` counter stat to router stats.
+* router: :ref:`Scoped routing <arch_overview_http_routing_route_scope>` is supported.
 * router check tool: add coverage reporting & enforcement.
 * router check tool: add comprehensive coverage reporting.
-* tracing: added support to the Zipkin reporter for sending list of spans as Zipkin JSON v2 and protobuf message over HTTP.
 * router check tool: add deprecated field check.
 * tls: added verification of IP address SAN fields in certificates against configured SANs in the
+* tracing: added support to the Zipkin reporter for sending list of spans as Zipkin JSON v2 and protobuf message over HTTP.
   certificate validation context.
 * tracing: added tags for gRPC response status and meesage.
+* upstream: added :ref:`an option <envoy_api_field_Cluster.CommonLbConfig.close_connections_on_host_set_change>` that allows draining HTTP, TCP connection pools on cluster membership change.
 * upstream: added network filter chains to upstream connections, see :ref:`filters<envoy_api_field_Cluster.filters>`.
 * upstream: use p2c to select hosts for least-requests load balancers if all host weights are the same, even in cases where weights are not equal to 1.
-* upstream: added :ref:`an option <envoy_api_field_Cluster.CommonLbConfig.close_connections_on_host_set_change>` that allows draining HTTP, TCP connection pools on cluster membership change.
 * zookeeper: parse responses and emit latency stats.
 
 1.11.1 (August 13, 2019)
diff --git a/include/envoy/stream_info/stream_info.h b/include/envoy/stream_info/stream_info.h
index 1558503ccfe5..442d8a1699a1 100644
--- a/include/envoy/stream_info/stream_info.h
+++ b/include/envoy/stream_info/stream_info.h
@@ -109,6 +109,8 @@ struct ResponseCodeDetailValues {
   // The request was rejected because it attempted an unsupported upgrade.
   const std::string UpgradeFailed = "upgrade_failed";
 
+  // The request was rejected by the HCM because there was no route configuration found.
+  const std::string RouteConfigurationNotFound = "route_configuration_not_found";
   // The request was rejected by the router filter because there was no route found.
   const std::string RouteNotFound = "route_not_found";
   // A direct response was generated by the router filter.
diff --git a/source/common/http/BUILD b/source/common/http/BUILD
index a41fdf2bb925..df01a16a57db 100644
--- a/source/common/http/BUILD
+++ b/source/common/http/BUILD
@@ -164,6 +164,7 @@ envoy_cc_library(
         "//include/envoy/router:rds_interface",
         "//include/envoy/router:scopes_interface",
         "//include/envoy/runtime:runtime_interface",
+        "//include/envoy/server:admin_interface",
         "//include/envoy/server:overload_manager_interface",
         "//include/envoy/ssl:connection_interface",
         "//include/envoy/stats:stats_interface",
diff --git a/source/common/http/conn_manager_impl.cc b/source/common/http/conn_manager_impl.cc
index c830114f5107..102373f9cadf 100644
--- a/source/common/http/conn_manager_impl.cc
+++ b/source/common/http/conn_manager_impl.cc
@@ -12,6 +12,7 @@
 #include "envoy/event/dispatcher.h"
 #include "envoy/network/drain_decision.h"
 #include "envoy/router/router.h"
+#include "envoy/server/admin.h"
 #include "envoy/ssl/connection.h"
 #include "envoy/stats/scope.h"
 #include "envoy/tracing/http_tracer.h"
@@ -431,12 +432,27 @@ void ConnectionManagerImpl::chargeTracingStats(const Tracing::Reason& tracing_re
 
 ConnectionManagerImpl::ActiveStream::ActiveStream(ConnectionManagerImpl& connection_manager)
     : connection_manager_(connection_manager),
-      snapped_route_config_(connection_manager.config_.routeConfigProvider()->config()),
       stream_id_(connection_manager.random_generator_.random()),
       request_response_timespan_(new Stats::Timespan(
           connection_manager_.stats_.named_.downstream_rq_time_, connection_manager_.timeSource())),
       stream_info_(connection_manager_.codec_->protocol(), connection_manager_.timeSource()),
       upstream_options_(std::make_shared<Network::Socket::Options>()) {
+  // For Server::Admin, no routeConfigProvider or SRDS route provider is used.
+  ASSERT(dynamic_cast<Server::Admin*>(&connection_manager_.config_) != nullptr ||
+             ((connection_manager.config_.routeConfigProvider() == nullptr &&
+               connection_manager.config_.scopedRouteConfigProvider() != nullptr) ||
+              (connection_manager.config_.routeConfigProvider() != nullptr &&
+               connection_manager.config_.scopedRouteConfigProvider() == nullptr)),
+         "Either routeConfigProvider or scopedRouteConfigProvider should be set in "
+         "ConnectionManagerImpl.");
+  if (connection_manager.config_.routeConfigProvider() != nullptr) {
+    snapped_route_config_ = connection_manager.config_.routeConfigProvider()->config();
+  } else if (connection_manager.config_.scopedRouteConfigProvider() != nullptr) {
+    snapped_scoped_routes_config_ =
+        connection_manager_.config_.scopedRouteConfigProvider()->config<Router::ScopedConfig>();
+    ASSERT(snapped_scoped_routes_config_ != nullptr,
+           "Scoped rds provider returns null for scoped routes config.");
+  }
   ScopeTrackerScopeState scope(this,
                                connection_manager_.read_callbacks_->connection().dispatcher());
 
@@ -613,6 +629,17 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
   ScopeTrackerScopeState scope(this,
                                connection_manager_.read_callbacks_->connection().dispatcher());
   request_headers_ = std::move(headers);
+  // For Admin thread, we don't use routeConfigProvider or SRDS route provider.
+  if (dynamic_cast<Server::Admin*>(&connection_manager_.config_) == nullptr &&
+      connection_manager_.config_.scopedRouteConfigProvider() != nullptr) {
+    ASSERT(snapped_route_config_ == nullptr,
+           "Route config already latched to the active stream when scoped RDS is enabled.");
+    // We need to snap snapped_route_config_ here as it's used in mutateRequestHeaders later.
+    if (!snapScopedRouteConfig()) {
+      return;
+    }
+  }
+
   if (Http::Headers::get().MethodValues.Head ==
       request_headers_->Method()->value().getStringView()) {
     is_head_request_ = true;
@@ -1220,10 +1247,36 @@ void ConnectionManagerImpl::startDrainSequence() {
   drain_timer_->enableTimer(config_.drainTimeout());
 }
 
+bool ConnectionManagerImpl::ActiveStream::snapScopedRouteConfig() {
+  ASSERT(request_headers_ != nullptr,
+         "Try to snap scoped route config when there is no request headers.");
+
+  snapped_route_config_ = snapped_scoped_routes_config_->getRouteConfig(*request_headers_);
+  // NOTE: if a RDS subscription hasn't got a RouteConfiguration back, a Router::NullConfigImpl is
+  // returned, in that case we let it pass.
+  if (snapped_route_config_ == nullptr) {
+    ENVOY_STREAM_LOG(trace, "can't find SRDS scope.", *this);
+    // Stop decoding now.
+    maybeEndDecode(true);
+    sendLocalReply(Grpc::Common::hasGrpcContentType(*request_headers_), Http::Code::NotFound,
+                   "route scope not found", nullptr, is_head_request_, absl::nullopt,
+                   StreamInfo::ResponseCodeDetails::get().RouteConfigurationNotFound);
+    return false;
+  }
+  return true;
+}
+
 void ConnectionManagerImpl::ActiveStream::refreshCachedRoute() {
   Router::RouteConstSharedPtr route;
   if (request_headers_ != nullptr) {
-    route = snapped_route_config_->route(*request_headers_, stream_id_);
+    if (dynamic_cast<Server::Admin*>(&connection_manager_.config_) == nullptr &&
+        connection_manager_.config_.scopedRouteConfigProvider() != nullptr) {
+      // NOTE: re-select scope as well in case the scope key header has been changed by a filter.
+      snapScopedRouteConfig();
+    }
+    if (snapped_route_config_ != nullptr) {
+      route = snapped_route_config_->route(*request_headers_, stream_id_);
+    }
   }
   stream_info_.route_entry_ = route ? route->routeEntry() : nullptr;
   cached_route_ = std::move(route);
diff --git a/source/common/http/conn_manager_impl.h b/source/common/http/conn_manager_impl.h
index 84dde922406e..661871ea991a 100644
--- a/source/common/http/conn_manager_impl.h
+++ b/source/common/http/conn_manager_impl.h
@@ -496,6 +496,11 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
 
     void traceRequest();
 
+    // Updates the snapped_route_config_ if scope found, or ends the stream by
+    // sending local reply.
+    // Returns true if scoped route config snapped, false otherwise.
+    bool snapScopedRouteConfig();
+
     void refreshCachedRoute();
 
     // Pass on watermark callbacks to watermark subscribers. This boils down to passing watermark
@@ -585,7 +590,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
 
     ConnectionManagerImpl& connection_manager_;
     Router::ConfigConstSharedPtr snapped_route_config_;
-    Router::ScopedConfigConstSharedPtr snapped_scoped_route_config_;
+    Router::ScopedConfigConstSharedPtr snapped_scoped_routes_config_;
     Tracing::SpanPtr active_span_;
     const uint64_t stream_id_;
     StreamEncoder* response_encoder_{};
diff --git a/source/common/router/scoped_rds.cc b/source/common/router/scoped_rds.cc
index 989e284668c4..9c710540930f 100644
--- a/source/common/router/scoped_rds.cc
+++ b/source/common/router/scoped_rds.cc
@@ -316,7 +316,7 @@ void ScopedRdsConfigSubscription::onConfigUpdate(
     *to_remove_repeated.Add() = scoped_route.first;
   }
   onConfigUpdate(to_add_repeated, to_remove_repeated, version_info);
-} // namespace Router
+}
 
 ScopedRdsConfigProvider::ScopedRdsConfigProvider(
     ScopedRdsConfigSubscriptionSharedPtr&& subscription)
diff --git a/test/common/grpc/grpc_client_integration.h b/test/common/grpc/grpc_client_integration.h
index ff6d4d3a7b53..bdfc0c6ae1ba 100644
--- a/test/common/grpc/grpc_client_integration.h
+++ b/test/common/grpc/grpc_client_integration.h
@@ -43,7 +43,6 @@ class GrpcClientIntegrationParamTest
     : public BaseGrpcClientIntegrationParamTest,
       public testing::TestWithParam<std::tuple<Network::Address::IpVersion, ClientType>> {
 public:
-  ~GrpcClientIntegrationParamTest() override = default;
   static std::string protocolTestParamsToString(
       const ::testing::TestParamInfo<std::tuple<Network::Address::IpVersion, ClientType>>& p) {
     return fmt::format("{}_{}",
@@ -54,10 +53,26 @@ class GrpcClientIntegrationParamTest
   ClientType clientType() const override { return std::get<1>(GetParam()); }
 };
 
+class DeltaSotwGrpcClientIntegrationParamTest
+    : public BaseGrpcClientIntegrationParamTest,
+      public testing::TestWithParam<std::tuple<Network::Address::IpVersion, ClientType, bool>> {
+public:
+  static std::string protocolTestParamsToString(
+      const ::testing::TestParamInfo<std::tuple<Network::Address::IpVersion, ClientType, bool>>&
+          p) {
+    return fmt::format("{}_{}",
+                       std::get<0>(p.param) == Network::Address::IpVersion::v4 ? "IPv4" : "IPv6",
+                       std::get<1>(p.param) == ClientType::GoogleGrpc ? "GoogleGrpc" : "EnvoyGrpc",
+                       std::get<2>(p.param) ? "Delta" : "StateOfTheWorld");
+  }
+  Network::Address::IpVersion ipVersion() const override { return std::get<0>(GetParam()); }
+  ClientType clientType() const override { return std::get<1>(GetParam()); }
+  bool isDelta() { return std::get<2>(GetParam()); }
+};
+
 class DeltaSotwIntegrationParamTest
     : public testing::TestWithParam<std::tuple<Network::Address::IpVersion, SotwOrDelta>> {
 public:
-  ~DeltaSotwIntegrationParamTest() override = default;
   static std::string protocolTestParamsToString(
       const ::testing::TestParamInfo<std::tuple<Network::Address::IpVersion, SotwOrDelta>>& p) {
     return fmt::format("{}_{}_{}",
@@ -84,10 +99,17 @@ class DeltaSotwIntegrationParamTest
 #define GRPC_CLIENT_INTEGRATION_PARAMS                                                             \
   testing::Combine(testing::ValuesIn(TestEnvironment::getIpVersionsForTest()),                     \
                    testing::Values(Grpc::ClientType::EnvoyGrpc, Grpc::ClientType::GoogleGrpc))
+#define DELTA_SOTW_GRPC_CLIENT_INTEGRATION_PARAMS                                                  \
+  testing::Combine(testing::ValuesIn(TestEnvironment::getIpVersionsForTest()),                     \
+                   testing::Values(Grpc::ClientType::EnvoyGrpc, Grpc::ClientType::GoogleGrpc),     \
+                   testing::Bool())
 #else
 #define GRPC_CLIENT_INTEGRATION_PARAMS                                                             \
   testing::Combine(testing::ValuesIn(TestEnvironment::getIpVersionsForTest()),                     \
                    testing::Values(Grpc::ClientType::EnvoyGrpc))
+#define DELTA_SOTW_GRPC_CLIENT_INTEGRATION_PARAMS                                                  \
+  testing::Combine(testing::ValuesIn(TestEnvironment::getIpVersionsForTest()),                     \
+                   testing::Values(Grpc::ClientType::EnvoyGrpc), testing::Bool())
 #endif // ENVOY_GOOGLE_GRPC
 
 #define DELTA_INTEGRATION_PARAMS                                                                   \
diff --git a/test/common/http/BUILD b/test/common/http/BUILD
index 7982b15c462f..1766f89aa0d7 100644
--- a/test/common/http/BUILD
+++ b/test/common/http/BUILD
@@ -129,17 +129,6 @@ envoy_cc_test_library(
     ],
 )
 
-envoy_cc_test_library(
-    name = "conn_manager_impl_common_lib",
-    hdrs = ["conn_manager_impl_common.h"],
-    deps = [
-        "//include/envoy/common:time_interface",
-        "//include/envoy/config:config_provider_interface",
-        "//include/envoy/router:rds_interface",
-        "//test/mocks/router:router_mocks",
-    ],
-)
-
 envoy_proto_library(
     name = "conn_manager_impl_fuzz_proto",
     srcs = ["conn_manager_impl_fuzz.proto"],
@@ -153,7 +142,6 @@ envoy_cc_fuzz_test(
     srcs = ["conn_manager_impl_fuzz_test.cc"],
     corpus = "conn_manager_impl_corpus",
     deps = [
-        ":conn_manager_impl_common_lib",
         ":conn_manager_impl_fuzz_proto_cc",
         "//source/common/common:empty_string",
         "//source/common/http:conn_manager_lib",
@@ -167,6 +155,7 @@ envoy_cc_fuzz_test(
         "//test/mocks/http:http_mocks",
         "//test/mocks/local_info:local_info_mocks",
         "//test/mocks/network:network_mocks",
+        "//test/mocks/router:router_mocks",
         "//test/mocks/runtime:runtime_mocks",
         "//test/mocks/ssl:ssl_mocks",
         "//test/mocks/tracing:tracing_mocks",
@@ -180,7 +169,6 @@ envoy_cc_test(
     name = "conn_manager_impl_test",
     srcs = ["conn_manager_impl_test.cc"],
     deps = [
-        ":conn_manager_impl_common_lib",
         "//include/envoy/access_log:access_log_interface",
         "//include/envoy/buffer:buffer_interface",
         "//include/envoy/event:dispatcher_interface",
@@ -207,6 +195,7 @@ envoy_cc_test(
         "//test/mocks/http:http_mocks",
         "//test/mocks/local_info:local_info_mocks",
         "//test/mocks/network:network_mocks",
+        "//test/mocks/router:router_mocks",
         "//test/mocks/runtime:runtime_mocks",
         "//test/mocks/server:server_mocks",
         "//test/mocks/ssl:ssl_mocks",
diff --git a/test/common/http/conn_manager_impl_fuzz_test.cc b/test/common/http/conn_manager_impl_fuzz_test.cc
index df9a0c2b4b35..60a097dec870 100644
--- a/test/common/http/conn_manager_impl_fuzz_test.cc
+++ b/test/common/http/conn_manager_impl_fuzz_test.cc
@@ -21,7 +21,6 @@
 #include "common/network/utility.h"
 #include "common/stats/symbol_table_creator.h"
 
-#include "test/common/http/conn_manager_impl_common.h"
 #include "test/common/http/conn_manager_impl_fuzz.pb.h"
 #include "test/fuzz/fuzz_runner.h"
 #include "test/fuzz/utility.h"
@@ -30,6 +29,7 @@
 #include "test/mocks/http/mocks.h"
 #include "test/mocks/local_info/mocks.h"
 #include "test/mocks/network/mocks.h"
+#include "test/mocks/router/mocks.h"
 #include "test/mocks/runtime/mocks.h"
 #include "test/mocks/ssl/mocks.h"
 #include "test/mocks/tracing/mocks.h"
@@ -47,13 +47,15 @@ namespace Http {
 class FuzzConfig : public ConnectionManagerConfig {
 public:
   FuzzConfig()
-      : route_config_provider_(time_system_), scoped_route_config_provider_(time_system_),
-        stats_{{ALL_HTTP_CONN_MAN_STATS(POOL_COUNTER(fake_stats_), POOL_GAUGE(fake_stats_),
+      : stats_{{ALL_HTTP_CONN_MAN_STATS(POOL_COUNTER(fake_stats_), POOL_GAUGE(fake_stats_),
                                         POOL_HISTOGRAM(fake_stats_))},
                "",
                fake_stats_},
         tracing_stats_{CONN_MAN_TRACING_STATS(POOL_COUNTER(fake_stats_))},
         listener_stats_{CONN_MAN_LISTENER_STATS(POOL_COUNTER(fake_stats_))} {
+    ON_CALL(route_config_provider_, lastUpdated()).WillByDefault(Return(time_system_.systemTime()));
+    ON_CALL(scoped_route_config_provider_, lastUpdated())
+        .WillByDefault(Return(time_system_.systemTime()));
     access_logs_.emplace_back(std::make_shared<NiceMock<AccessLog::MockInstance>>());
   }
 
@@ -86,9 +88,17 @@ class FuzzConfig : public ConnectionManagerConfig {
   std::chrono::milliseconds streamIdleTimeout() const override { return stream_idle_timeout_; }
   std::chrono::milliseconds requestTimeout() const override { return request_timeout_; }
   std::chrono::milliseconds delayedCloseTimeout() const override { return delayed_close_timeout_; }
-  Router::RouteConfigProvider* routeConfigProvider() override { return &route_config_provider_; }
+  Router::RouteConfigProvider* routeConfigProvider() override {
+    if (use_srds_) {
+      return nullptr;
+    }
+    return &route_config_provider_;
+  }
   Config::ConfigProvider* scopedRouteConfigProvider() override {
-    return &scoped_route_config_provider_;
+    if (use_srds_) {
+      return &scoped_route_config_provider_;
+    }
+    return nullptr;
   }
   const std::string& serverName() override { return server_name_; }
   HttpConnectionManagerProto::ServerHeaderTransformation serverHeaderTransformation() override {
@@ -124,8 +134,9 @@ class FuzzConfig : public ConnectionManagerConfig {
   NiceMock<MockFilterChainFactory> filter_factory_;
   Event::SimulatedTimeSystem time_system_;
   SlowDateProviderImpl date_provider_{time_system_};
-  ConnectionManagerImplHelper::RouteConfigProvider route_config_provider_;
-  ConnectionManagerImplHelper::ScopedRouteConfigProvider scoped_route_config_provider_;
+  bool use_srds_{};
+  Router::MockRouteConfigProvider route_config_provider_;
+  Router::MockScopedRouteConfigProvider scoped_route_config_provider_;
   std::string server_name_;
   HttpConnectionManagerProto::ServerHeaderTransformation server_transformation_{
       HttpConnectionManagerProto::OVERWRITE};
diff --git a/test/common/http/conn_manager_impl_test.cc b/test/common/http/conn_manager_impl_test.cc
index 496b2af6c7ea..ebf2c71d59ad 100644
--- a/test/common/http/conn_manager_impl_test.cc
+++ b/test/common/http/conn_manager_impl_test.cc
@@ -26,13 +26,13 @@
 
 #include "extensions/access_loggers/file/file_access_log_impl.h"
 
-#include "test/common/http/conn_manager_impl_common.h"
 #include "test/mocks/access_log/mocks.h"
 #include "test/mocks/buffer/mocks.h"
 #include "test/mocks/common.h"
 #include "test/mocks/http/mocks.h"
 #include "test/mocks/local_info/mocks.h"
 #include "test/mocks/network/mocks.h"
+#include "test/mocks/router/mocks.h"
 #include "test/mocks/runtime/mocks.h"
 #include "test/mocks/server/mocks.h"
 #include "test/mocks/ssl/mocks.h"
@@ -69,9 +69,7 @@ namespace Http {
 class HttpConnectionManagerImplTest : public testing::Test, public ConnectionManagerConfig {
 public:
   HttpConnectionManagerImplTest()
-      : route_config_provider_(test_time_.timeSystem()),
-        scoped_route_config_provider_(test_time_.timeSystem()),
-        http_context_(fake_stats_.symbolTable()), access_log_path_("dummy_path"),
+      : http_context_(fake_stats_.symbolTable()), access_log_path_("dummy_path"),
         access_logs_{
             AccessLog::InstanceSharedPtr{new Extensions::AccessLoggers::File::FileAccessLog(
                 access_log_path_, {}, AccessLog::AccessLogFormatUtils::defaultAccessLogFormatter(),
@@ -86,6 +84,10 @@ class HttpConnectionManagerImplTest : public testing::Test, public ConnectionMan
 
     http_context_.setTracer(tracer_);
 
+    ON_CALL(route_config_provider_, lastUpdated())
+        .WillByDefault(Return(test_time_.timeSystem().systemTime()));
+    ON_CALL(scoped_route_config_provider_, lastUpdated())
+        .WillByDefault(Return(test_time_.timeSystem().systemTime()));
     // response_encoder_ is not a NiceMock on purpose. This prevents complaining about this
     // method only.
     EXPECT_CALL(response_encoder_, getStream()).Times(AtLeast(0));
@@ -95,7 +97,8 @@ class HttpConnectionManagerImplTest : public testing::Test, public ConnectionMan
     filter_callbacks_.connection_.dispatcher_.clearDeferredDeleteList();
   }
 
-  void setup(bool ssl, const std::string& server_name, bool tracing = true) {
+  void setup(bool ssl, const std::string& server_name, bool tracing = true, bool use_srds = false) {
+    use_srds_ = use_srds;
     if (ssl) {
       ssl_connection_ = std::make_unique<Ssl::MockConnectionInfo>();
     }
@@ -271,9 +274,18 @@ class HttpConnectionManagerImplTest : public testing::Test, public ConnectionMan
   std::chrono::milliseconds streamIdleTimeout() const override { return stream_idle_timeout_; }
   std::chrono::milliseconds requestTimeout() const override { return request_timeout_; }
   std::chrono::milliseconds delayedCloseTimeout() const override { return delayed_close_timeout_; }
-  Router::RouteConfigProvider* routeConfigProvider() override { return &route_config_provider_; }
+  bool use_srds_{};
+  Router::RouteConfigProvider* routeConfigProvider() override {
+    if (use_srds_) {
+      return nullptr;
+    }
+    return &route_config_provider_;
+  }
   Config::ConfigProvider* scopedRouteConfigProvider() override {
-    return &scoped_route_config_provider_;
+    if (use_srds_) {
+      return &scoped_route_config_provider_;
+    }
+    return nullptr;
   }
   const std::string& serverName() override { return server_name_; }
   HttpConnectionManagerProto::ServerHeaderTransformation serverHeaderTransformation() override {
@@ -302,8 +314,9 @@ class HttpConnectionManagerImplTest : public testing::Test, public ConnectionMan
   bool shouldMergeSlashes() const override { return merge_slashes_; }
 
   DangerousDeprecatedTestTime test_time_;
-  ConnectionManagerImplHelper::RouteConfigProvider route_config_provider_;
-  ConnectionManagerImplHelper::ScopedRouteConfigProvider scoped_route_config_provider_;
+  NiceMock<Router::MockRouteConfigProvider> route_config_provider_;
+  std::shared_ptr<Router::MockConfig> route_config_{new NiceMock<Router::MockConfig>()};
+  NiceMock<Router::MockScopedRouteConfigProvider> scoped_route_config_provider_;
   NiceMock<Tracing::MockHttpTracer> tracer_;
   Stats::IsolatedStoreImpl fake_stats_;
   Http::ContextImpl http_context_;
@@ -1890,7 +1903,7 @@ TEST_F(HttpConnectionManagerImplTest, PerStreamIdleTimeoutAfterBidiData) {
 TEST_F(HttpConnectionManagerImplTest, RequestTimeoutDisabledByDefault) {
   setup(false, "");
 
-  EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance&) -> void {
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> void {
     EXPECT_CALL(filter_callbacks_.connection_.dispatcher_, createTimer_).Times(0);
     conn_manager_->newStream(response_encoder_);
   }));
@@ -1903,7 +1916,7 @@ TEST_F(HttpConnectionManagerImplTest, RequestTimeoutDisabledIfSetToZero) {
   request_timeout_ = std::chrono::milliseconds(0);
   setup(false, "");
 
-  EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance&) -> void {
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> void {
     EXPECT_CALL(filter_callbacks_.connection_.dispatcher_, createTimer_).Times(0);
     conn_manager_->newStream(response_encoder_);
   }));
@@ -1916,7 +1929,7 @@ TEST_F(HttpConnectionManagerImplTest, RequestTimeoutValidlyConfigured) {
   request_timeout_ = std::chrono::milliseconds(10);
   setup(false, "");
 
-  EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance&) -> void {
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> void {
     Event::MockTimer* request_timer = setUpTimer();
     EXPECT_CALL(*request_timer, enableTimer(request_timeout_, _));
 
@@ -1932,7 +1945,7 @@ TEST_F(HttpConnectionManagerImplTest, RequestTimeoutCallbackDisarmsAndReturns408
   setup(false, "");
 
   std::string response_body;
-  EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance&) -> void {
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> void {
     Event::MockTimer* request_timer = setUpTimer();
     EXPECT_CALL(*request_timer, enableTimer(request_timeout_, _)).Times(1);
     EXPECT_CALL(*request_timer, disableTimer()).Times(AtLeast(1));
@@ -1959,7 +1972,7 @@ TEST_F(HttpConnectionManagerImplTest, RequestTimeoutIsNotDisarmedOnIncompleteReq
   request_timeout_ = std::chrono::milliseconds(10);
   setup(false, "");
 
-  EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance&) -> void {
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> void {
     Event::MockTimer* request_timer = setUpTimer();
     EXPECT_CALL(*request_timer, enableTimer(request_timeout_, _)).Times(1);
     EXPECT_CALL(*request_timer, disableTimer()).Times(0);
@@ -1982,7 +1995,7 @@ TEST_F(HttpConnectionManagerImplTest, RequestTimeoutIsDisarmedOnCompleteRequestW
   request_timeout_ = std::chrono::milliseconds(10);
   setup(false, "");
 
-  EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance&) -> void {
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> void {
     Event::MockTimer* request_timer = setUpTimer();
     EXPECT_CALL(*request_timer, enableTimer(request_timeout_, _)).Times(1);
 
@@ -2058,7 +2071,7 @@ TEST_F(HttpConnectionManagerImplTest, RequestTimeoutIsDisarmedOnEncodeHeaders) {
       }));
   EXPECT_CALL(response_encoder_, encodeHeaders(_, _));
 
-  EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance&) -> void {
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> void {
     Event::MockTimer* request_timer = setUpTimer();
     EXPECT_CALL(*request_timer, enableTimer(request_timeout_, _)).Times(1);
 
@@ -2084,7 +2097,7 @@ TEST_F(HttpConnectionManagerImplTest, RequestTimeoutIsDisarmedOnConnectionTermin
   setup(false, "");
 
   Event::MockTimer* request_timer = setUpTimer();
-  EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance&) -> void {
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> void {
     StreamDecoder* decoder = &conn_manager_->newStream(response_encoder_);
     HeaderMapPtr headers{
         new TestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "GET"}}};
@@ -4274,7 +4287,7 @@ TEST_F(HttpConnectionManagerImplTest, OverlyLongHeadersRejected) {
 
   std::string response_code;
   std::string response_body;
-  EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance&) -> void {
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> void {
     StreamDecoder* decoder = &conn_manager_->newStream(response_encoder_);
     HeaderMapPtr headers{
         new TestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "GET"}}};
@@ -4299,7 +4312,7 @@ TEST_F(HttpConnectionManagerImplTest, OverlyLongHeadersAcceptedIfConfigured) {
   max_request_headers_kb_ = 62;
   setup(false, "");
 
-  EXPECT_CALL(*codec_, dispatch(_)).Times(1).WillOnce(Invoke([&](Buffer::Instance&) -> void {
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> void {
     StreamDecoder* decoder = &conn_manager_->newStream(response_encoder_);
     HeaderMapPtr headers{
         new TestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "GET"}}};
@@ -4493,5 +4506,236 @@ TEST_F(HttpConnectionManagerImplTest, TestSessionTrace) {
   }
 }
 
+// SRDS no scope found.
+TEST_F(HttpConnectionManagerImplTest, TestSRDSRouteNotFound) {
+  setup(false, "", true, true);
+
+  EXPECT_CALL(*static_cast<const Router::MockScopedConfig*>(
+                  scopedRouteConfigProvider()->config<Router::ScopedConfig>().get()),
+              getRouteConfig(_))
+      .WillOnce(Return(nullptr));
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance& data) -> void {
+    StreamDecoder* decoder = &conn_manager_->newStream(response_encoder_);
+    HeaderMapPtr headers{
+        new TestHeaderMapImpl{{":authority", "host"}, {":method", "GET"}, {":path", "/foo"}}};
+    decoder->decodeHeaders(std::move(headers), true);
+    data.drain(4);
+  }));
+
+  EXPECT_CALL(response_encoder_, encodeHeaders(_, false))
+      .WillOnce(Invoke([](const HeaderMap& headers, bool) -> void {
+        EXPECT_EQ("404", headers.Status()->value().getStringView());
+      }));
+
+  std::string response_body;
+  EXPECT_CALL(response_encoder_, encodeData(_, true)).WillOnce(AddBufferToString(&response_body));
+
+  Buffer::OwnedImpl fake_input("1234");
+  conn_manager_->onData(fake_input, false);
+  EXPECT_EQ(response_body, "route scope not found");
+}
+
+// SRDS updating scopes affects routing.
+TEST_F(HttpConnectionManagerImplTest, TestSRDSUpdate) {
+  setup(false, "", true, true);
+
+  EXPECT_CALL(*static_cast<const Router::MockScopedConfig*>(
+                  scopedRouteConfigProvider()->config<Router::ScopedConfig>().get()),
+              getRouteConfig(_))
+      .Times(3)
+      .WillOnce(Return(nullptr))
+      .WillOnce(Return(route_config_))
+      .WillOnce(Return(route_config_)); // refreshCachedRoute
+  EXPECT_CALL(*codec_, dispatch(_))
+      .Times(2) // Once for no scoped routes, once for scoped routing
+      .WillRepeatedly(Invoke([&](Buffer::Instance& data) -> void {
+        StreamDecoder* decoder = &conn_manager_->newStream(response_encoder_);
+        HeaderMapPtr headers{
+            new TestHeaderMapImpl{{":authority", "host"}, {":method", "GET"}, {":path", "/foo"}}};
+        decoder->decodeHeaders(std::move(headers), true);
+        data.drain(4);
+      }));
+  EXPECT_CALL(response_encoder_, encodeHeaders(_, false))
+      .WillOnce(Invoke([](const HeaderMap& headers, bool) -> void {
+        EXPECT_EQ("404", headers.Status()->value().getStringView());
+      }));
+
+  std::string response_body;
+  EXPECT_CALL(response_encoder_, encodeData(_, true)).WillOnce(AddBufferToString(&response_body));
+
+  Buffer::OwnedImpl fake_input("1234");
+  conn_manager_->onData(fake_input, false);
+  EXPECT_EQ(response_body, "route scope not found");
+
+  // Now route config provider returns something.
+  setupFilterChain(1, 0); // Recreate the chain for second stream.
+  const std::string fake_cluster1_name = "fake_cluster1";
+  std::shared_ptr<Router::MockRoute> route1 = std::make_shared<NiceMock<Router::MockRoute>>();
+  EXPECT_CALL(route1->route_entry_, clusterName()).WillRepeatedly(ReturnRef(fake_cluster1_name));
+  std::shared_ptr<Upstream::MockThreadLocalCluster> fake_cluster1 =
+      std::make_shared<NiceMock<Upstream::MockThreadLocalCluster>>();
+  EXPECT_CALL(cluster_manager_, get(_)).WillOnce(Return(fake_cluster1.get()));
+  EXPECT_CALL(*route_config_, route(_, _)).WillOnce(Return(route1));
+  EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, true))
+      .WillOnce(InvokeWithoutArgs([&]() -> FilterHeadersStatus {
+        EXPECT_EQ(route1, decoder_filters_[0]->callbacks_->route());
+        EXPECT_EQ(route1->routeEntry(), decoder_filters_[0]->callbacks_->streamInfo().routeEntry());
+        EXPECT_EQ(fake_cluster1->info(), decoder_filters_[0]->callbacks_->clusterInfo());
+        return FilterHeadersStatus::StopIteration;
+      }));
+  EXPECT_CALL(*decoder_filters_[0], decodeComplete());
+  Buffer::OwnedImpl fake_input2("1234");
+  conn_manager_->onData(fake_input2, false);
+}
+
+// SRDS Scope header update cause cross-scope reroute.
+TEST_F(HttpConnectionManagerImplTest, TestSRDSCrossScopeReroute) {
+  setup(false, "", true, true);
+
+  std::shared_ptr<Router::MockConfig> route_config1 =
+      std::make_shared<NiceMock<Router::MockConfig>>();
+  std::shared_ptr<Router::MockConfig> route_config2 =
+      std::make_shared<NiceMock<Router::MockConfig>>();
+  std::shared_ptr<Router::MockRoute> route1 = std::make_shared<NiceMock<Router::MockRoute>>();
+  std::shared_ptr<Router::MockRoute> route2 = std::make_shared<NiceMock<Router::MockRoute>>();
+  EXPECT_CALL(*route_config1, route(_, _)).WillRepeatedly(Return(route1));
+  EXPECT_CALL(*route_config2, route(_, _)).WillRepeatedly(Return(route2));
+  EXPECT_CALL(*static_cast<const Router::MockScopedConfig*>(
+                  scopedRouteConfigProvider()->config<Router::ScopedConfig>().get()),
+              getRouteConfig(_))
+      // 1. Snap scoped route config;
+      // 2. refreshCachedRoute (both in decodeHeaders(headers,end_stream);
+      // 3. then refreshCachedRoute triggered by decoder_filters_[1]->callbacks_->route().
+      .Times(3)
+      .WillRepeatedly(Invoke([&](const HeaderMap& headers) -> Router::ConfigConstSharedPtr {
+        auto& test_headers = static_cast<const TestHeaderMapImpl&>(headers);
+        if (test_headers.get_("scope_key") == "foo") {
+          return route_config1;
+        }
+        return route_config2;
+      }));
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance& data) -> void {
+    StreamDecoder* decoder = &conn_manager_->newStream(response_encoder_);
+    HeaderMapPtr headers{new TestHeaderMapImpl{
+        {":authority", "host"}, {":method", "GET"}, {"scope_key", "foo"}, {":path", "/foo"}}};
+    decoder->decodeHeaders(std::move(headers), false);
+    data.drain(4);
+  }));
+  setupFilterChain(2, 0);
+  EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, false))
+      .WillOnce(Invoke([&](Http::HeaderMap& headers, bool) -> FilterHeadersStatus {
+        EXPECT_EQ(route1, decoder_filters_[0]->callbacks_->route());
+        auto& test_headers = static_cast<TestHeaderMapImpl&>(headers);
+        // Clear cached route and change scope key to "bar".
+        decoder_filters_[0]->callbacks_->clearRouteCache();
+        test_headers.remove("scope_key");
+        test_headers.addCopy("scope_key", "bar");
+        return FilterHeadersStatus::Continue;
+      }));
+  EXPECT_CALL(*decoder_filters_[1], decodeHeaders(_, false))
+      .WillOnce(Invoke([&](Http::HeaderMap& headers, bool) -> FilterHeadersStatus {
+        auto& test_headers = static_cast<TestHeaderMapImpl&>(headers);
+        EXPECT_EQ(test_headers.get_("scope_key"), "bar");
+        // Route now switched to route2 as header "scope_key" has changed.
+        EXPECT_EQ(route2, decoder_filters_[1]->callbacks_->route());
+        EXPECT_EQ(route2->routeEntry(), decoder_filters_[1]->callbacks_->streamInfo().routeEntry());
+        return FilterHeadersStatus::StopIteration;
+      }));
+
+  Buffer::OwnedImpl fake_input("1234");
+  conn_manager_->onData(fake_input, false);
+}
+
+// SRDS scoped RouteConfiguration found and route found.
+TEST_F(HttpConnectionManagerImplTest, TestSRDSRouteFound) {
+  setup(false, "", true, true);
+  setupFilterChain(1, 0);
+
+  const std::string fake_cluster1_name = "fake_cluster1";
+  std::shared_ptr<Router::MockRoute> route1 = std::make_shared<NiceMock<Router::MockRoute>>();
+  EXPECT_CALL(route1->route_entry_, clusterName()).WillRepeatedly(ReturnRef(fake_cluster1_name));
+  std::shared_ptr<Upstream::MockThreadLocalCluster> fake_cluster1 =
+      std::make_shared<NiceMock<Upstream::MockThreadLocalCluster>>();
+  EXPECT_CALL(cluster_manager_, get(_)).WillOnce(Return(fake_cluster1.get()));
+  EXPECT_CALL(*scopedRouteConfigProvider()->config<Router::MockScopedConfig>(), getRouteConfig(_))
+      // 1. decodeHeaders() snaping route config.
+      // 2. refreshCachedRoute() later in the same decodeHeaders().
+      .Times(2);
+  EXPECT_CALL(
+      *static_cast<const Router::MockConfig*>(
+          scopedRouteConfigProvider()->config<Router::MockScopedConfig>()->route_config_.get()),
+      route(_, _))
+      .WillOnce(Return(route1));
+  StreamDecoder* decoder = nullptr;
+  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance& data) -> void {
+    decoder = &conn_manager_->newStream(response_encoder_);
+    HeaderMapPtr headers{
+        new TestHeaderMapImpl{{":authority", "host"}, {":method", "GET"}, {":path", "/foo"}}};
+    decoder->decodeHeaders(std::move(headers), true);
+    data.drain(4);
+  }));
+  EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, true))
+      .WillOnce(InvokeWithoutArgs([&]() -> FilterHeadersStatus {
+        EXPECT_EQ(route1, decoder_filters_[0]->callbacks_->route());
+        EXPECT_EQ(route1->routeEntry(), decoder_filters_[0]->callbacks_->streamInfo().routeEntry());
+        EXPECT_EQ(fake_cluster1->info(), decoder_filters_[0]->callbacks_->clusterInfo());
+        return FilterHeadersStatus::StopIteration;
+      }));
+  EXPECT_CALL(*decoder_filters_[0], decodeComplete());
+
+  Buffer::OwnedImpl fake_input("1234");
+  conn_manager_->onData(fake_input, false);
+}
+
+class HttpConnectionManagerImplDeathTest : public HttpConnectionManagerImplTest {
+public:
+  Router::RouteConfigProvider* routeConfigProvider() override {
+    return route_config_provider2_.get();
+  }
+  Config::ConfigProvider* scopedRouteConfigProvider() override {
+    return scoped_route_config_provider2_.get();
+  }
+
+  std::shared_ptr<Router::MockRouteConfigProvider> route_config_provider2_;
+  std::shared_ptr<Router::MockScopedRouteConfigProvider> scoped_route_config_provider2_;
+};
+
+// HCM config can only have either RouteConfigProvider or ScopedRoutesConfigProvider.
+TEST_F(HttpConnectionManagerImplDeathTest, InvalidConnectionManagerConfig) {
+  setup(false, "");
+
+  Buffer::OwnedImpl fake_input("1234");
+  EXPECT_CALL(*codec_, dispatch(_)).WillRepeatedly(Invoke([&](Buffer::Instance&) -> void {
+    conn_manager_->newStream(response_encoder_);
+  }));
+  // Either RDS or SRDS should be set.
+  EXPECT_DEBUG_DEATH(conn_manager_->onData(fake_input, false),
+                     "Either routeConfigProvider or scopedRouteConfigProvider should be set in "
+                     "ConnectionManagerImpl.");
+
+  route_config_provider2_ = std::make_shared<NiceMock<Router::MockRouteConfigProvider>>();
+
+  // Only route config provider valid.
+  EXPECT_NO_THROW(conn_manager_->onData(fake_input, false));
+
+  scoped_route_config_provider2_ =
+      std::make_shared<NiceMock<Router::MockScopedRouteConfigProvider>>();
+  // Can't have RDS and SRDS provider in the same time.
+  EXPECT_DEBUG_DEATH(conn_manager_->onData(fake_input, false),
+                     "Either routeConfigProvider or scopedRouteConfigProvider should be set in "
+                     "ConnectionManagerImpl.");
+
+  route_config_provider2_.reset();
+  // Only scoped route config provider valid.
+  EXPECT_NO_THROW(conn_manager_->onData(fake_input, false));
+
+#if !defined(NDEBUG)
+  EXPECT_CALL(*scoped_route_config_provider2_, getConfig()).WillRepeatedly(Return(nullptr));
+  // ASSERT failure when SRDS provider returns a nullptr.
+  EXPECT_DEBUG_DEATH(conn_manager_->onData(fake_input, false),
+                     "Scoped rds provider returns null for scoped routes config.");
+#endif // !defined(NDEBUG)
+}
+
 } // namespace Http
 } // namespace Envoy
diff --git a/test/common/router/scoped_rds_test.cc b/test/common/router/scoped_rds_test.cc
index 00936537bc7f..8ee6cfe8aecf 100644
--- a/test/common/router/scoped_rds_test.cc
+++ b/test/common/router/scoped_rds_test.cc
@@ -247,7 +247,7 @@ route_configuration_name: foo_routes
 }
 
 // Tests that multiple uniquely named non-conflict resources are allowed in config updates.
-TEST_F(ScopedRdsTest, MultipleResourcesStow) {
+TEST_F(ScopedRdsTest, MultipleResourcesSotw) {
   setup();
 
   const std::string config_yaml = R"EOF(
diff --git a/test/integration/scoped_rds_integration_test.cc b/test/integration/scoped_rds_integration_test.cc
index e14e87d08fa7..1cefe11ac397 100644
--- a/test/integration/scoped_rds_integration_test.cc
+++ b/test/integration/scoped_rds_integration_test.cc
@@ -4,6 +4,7 @@
 
 #include "test/common/grpc/grpc_client_integration.h"
 #include "test/integration/http_integration.h"
+#include "test/test_common/printers.h"
 
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
@@ -12,12 +13,12 @@ namespace Envoy {
 namespace {
 
 class ScopedRdsIntegrationTest : public HttpIntegrationTest,
-                                 public Grpc::GrpcClientIntegrationParamTest {
+                                 public Grpc::DeltaSotwGrpcClientIntegrationParamTest {
 protected:
   struct FakeUpstreamInfo {
     FakeHttpConnectionPtr connection_;
     FakeUpstream* upstream_{};
-    FakeStreamPtr stream_;
+    absl::flat_hash_map<std::string, FakeStreamPtr> stream_by_resource_name_;
   };
 
   ScopedRdsIntegrationTest()
@@ -29,7 +30,15 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest,
   }
 
   void initialize() override {
+    // Setup two upstream hosts, one for each cluster.
+    setUpstreamCount(2);
+
     config_helper_.addConfigModifier([](envoy::config::bootstrap::v2::Bootstrap& bootstrap) {
+      // Add the static cluster to serve SRDS.
+      auto* cluster_1 = bootstrap.mutable_static_resources()->add_clusters();
+      cluster_1->MergeFrom(bootstrap.static_resources().clusters()[0]);
+      cluster_1->set_name("cluster_1");
+
       // Add the static cluster to serve SRDS.
       auto* scoped_rds_cluster = bootstrap.mutable_static_resources()->add_clusters();
       scoped_rds_cluster->MergeFrom(bootstrap.static_resources().clusters()[0]);
@@ -50,15 +59,16 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest,
 fragments:
   - header_value_extractor:
       name: Addr
+      element_separator: ;
       element:
         key: x-foo-key
-        separator: ;
+        separator: =
 )EOF";
           envoy::config::filter::network::http_connection_manager::v2::ScopedRoutes::ScopeKeyBuilder
               scope_key_builder;
           TestUtility::loadFromYaml(scope_key_builder_config_yaml, scope_key_builder);
           auto* scoped_routes = http_connection_manager.mutable_scoped_routes();
-          scoped_routes->set_name("foo-scoped-routes");
+          scoped_routes->set_name(srds_config_name_);
           *scoped_routes->mutable_scope_key_builder() = scope_key_builder;
 
           envoy::api::v2::core::ApiConfigSource* rds_api_config_source =
@@ -72,7 +82,11 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest,
               scoped_routes->mutable_scoped_rds()
                   ->mutable_scoped_rds_config_source()
                   ->mutable_api_config_source();
-          srds_api_config_source->set_api_type(envoy::api::v2::core::ApiConfigSource::GRPC);
+          if (isDelta()) {
+            srds_api_config_source->set_api_type(envoy::api::v2::core::ApiConfigSource::DELTA_GRPC);
+          } else {
+            srds_api_config_source->set_api_type(envoy::api::v2::core::ApiConfigSource::GRPC);
+          }
           grpc_service = srds_api_config_source->add_grpc_services();
           setGrpcService(*grpc_service, "srds_cluster", getScopedRdsFakeUpstream().localAddress());
         });
@@ -80,6 +94,41 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest,
     HttpIntegrationTest::initialize();
   }
 
+  // TODO(stevenzzzz): move these utility methods to base classes to share with other tests.
+  // Helper that verifies if given headers are in the response header map.
+  void verifyResponse(IntegrationStreamDecoderPtr response, const std::string& response_code,
+                      const Http::TestHeaderMapImpl& expected_headers,
+                      const std::string& expected_body) {
+    EXPECT_TRUE(response->complete());
+    EXPECT_EQ(response_code, response->headers().Status()->value().getStringView());
+    expected_headers.iterate(
+        [](const Http::HeaderEntry& header, void* context) -> Http::HeaderMap::Iterate {
+          auto response_headers = static_cast<Http::HeaderMap*>(context);
+          const Http::HeaderEntry* entry = response_headers->get(
+              Http::LowerCaseString{std::string(header.key().getStringView())});
+          EXPECT_NE(entry, nullptr);
+          EXPECT_EQ(header.value().getStringView(), entry->value().getStringView());
+          return Http::HeaderMap::Iterate::Continue;
+        },
+        const_cast<void*>(static_cast<const void*>(&response->headers())));
+    EXPECT_EQ(response->body(), expected_body);
+  }
+
+  // Helper that sends a request to Envoy, and verifies if Envoy response headers and body size is
+  // the same as the expected headers map.
+  void sendRequestAndVerifyResponse(const Http::TestHeaderMapImpl& request_headers,
+                                    const int request_size,
+                                    const Http::TestHeaderMapImpl& response_headers,
+                                    const int response_size, const int backend_idx) {
+    codec_client_ = makeHttpConnection(lookupPort("http"));
+    auto response = sendRequestAndWaitForResponse(request_headers, request_size, response_headers,
+                                                  response_size, backend_idx);
+    verifyResponse(std::move(response), "200", response_headers, std::string(response_size, 'a'));
+    EXPECT_TRUE(upstream_request_->complete());
+    EXPECT_EQ(request_size, upstream_request_->bodyLength());
+    cleanupUpstreamAndDownstream();
+  }
+
   void createUpstreams() override {
     HttpIntegrationTest::createUpstreams();
     // Create the SRDS upstream.
@@ -108,38 +157,87 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest,
     resetFakeUpstreamInfo(&scoped_rds_upstream_info_);
   }
 
-  FakeUpstream& getRdsFakeUpstream() const { return *fake_upstreams_[2]; }
+  FakeUpstream& getRdsFakeUpstream() const { return *fake_upstreams_[3]; }
 
-  FakeUpstream& getScopedRdsFakeUpstream() const { return *fake_upstreams_[1]; }
+  FakeUpstream& getScopedRdsFakeUpstream() const { return *fake_upstreams_[2]; }
 
-  void createStream(FakeUpstreamInfo* upstream_info, FakeUpstream& upstream) {
-    upstream_info->upstream_ = &upstream;
-    AssertionResult result =
-        upstream_info->upstream_->waitForHttpConnection(*dispatcher_, upstream_info->connection_);
-    RELEASE_ASSERT(result, result.message());
-    result = upstream_info->connection_->waitForNewStream(*dispatcher_, upstream_info->stream_);
+  void createStream(FakeUpstreamInfo* upstream_info, FakeUpstream& upstream,
+                    const std::string& resource_name) {
+    if (upstream_info->upstream_ == nullptr) {
+      // bind upstream if not yet.
+      upstream_info->upstream_ = &upstream;
+      AssertionResult result =
+          upstream_info->upstream_->waitForHttpConnection(*dispatcher_, upstream_info->connection_);
+      RELEASE_ASSERT(result, result.message());
+    }
+    if (!upstream_info->stream_by_resource_name_.try_emplace(resource_name, nullptr).second) {
+      RELEASE_ASSERT(false,
+                     fmt::format("stream with resource name '{}' already exists!", resource_name));
+    }
+    auto result = upstream_info->connection_->waitForNewStream(
+        *dispatcher_, upstream_info->stream_by_resource_name_[resource_name]);
     RELEASE_ASSERT(result, result.message());
-    upstream_info->stream_->startGrpcStream();
+    upstream_info->stream_by_resource_name_[resource_name]->startGrpcStream();
   }
 
-  void createRdsStream() { createStream(&rds_upstream_info_, getRdsFakeUpstream()); }
+  void createRdsStream(const std::string& resource_name) {
+    createStream(&rds_upstream_info_, getRdsFakeUpstream(), resource_name);
+  }
 
   void createScopedRdsStream() {
-    createStream(&scoped_rds_upstream_info_, getScopedRdsFakeUpstream());
+    createStream(&scoped_rds_upstream_info_, getScopedRdsFakeUpstream(), srds_config_name_);
   }
 
   void sendRdsResponse(const std::string& route_config, const std::string& version) {
     envoy::api::v2::DiscoveryResponse response;
     response.set_version_info(version);
     response.set_type_url(Config::TypeUrl::get().RouteConfiguration);
-    response.add_resources()->PackFrom(
-        TestUtility::parseYaml<envoy::api::v2::RouteConfiguration>(route_config));
-    rds_upstream_info_.stream_->sendGrpcMessage(response);
+    auto route_configuration =
+        TestUtility::parseYaml<envoy::api::v2::RouteConfiguration>(route_config);
+    response.add_resources()->PackFrom(route_configuration);
+    ASSERT(rds_upstream_info_.stream_by_resource_name_[route_configuration.name()] != nullptr);
+    rds_upstream_info_.stream_by_resource_name_[route_configuration.name()]->sendGrpcMessage(
+        response);
+  }
+
+  void sendSrdsResponse(const std::vector<std::string>& sotw_list,
+                        const std::vector<std::string>& to_add_list,
+                        const std::vector<std::string>& to_delete_list,
+                        const std::string& version) {
+    if (isDelta()) {
+      sendDeltaScopedRdsResponse(to_add_list, to_delete_list, version);
+    } else {
+      sendSotwScopedRdsResponse(sotw_list, version);
+    }
+  }
+
+  void sendDeltaScopedRdsResponse(const std::vector<std::string>& to_add_list,
+                                  const std::vector<std::string>& to_delete_list,
+                                  const std::string& version) {
+    ASSERT(scoped_rds_upstream_info_.stream_by_resource_name_[srds_config_name_] != nullptr);
+
+    envoy::api::v2::DeltaDiscoveryResponse response;
+    response.set_system_version_info(version);
+    response.set_type_url(Config::TypeUrl::get().ScopedRouteConfiguration);
+
+    for (const auto& scope_name : to_delete_list) {
+      *response.add_removed_resources() = scope_name;
+    }
+    for (const auto& resource_proto : to_add_list) {
+      envoy::api::v2::ScopedRouteConfiguration scoped_route_proto;
+      TestUtility::loadFromYaml(resource_proto, scoped_route_proto);
+      auto resource = response.add_resources();
+      resource->set_name(scoped_route_proto.name());
+      resource->set_version(version);
+      resource->mutable_resource()->PackFrom(scoped_route_proto);
+    }
+    scoped_rds_upstream_info_.stream_by_resource_name_[srds_config_name_]->sendGrpcMessage(
+        response);
   }
 
-  void sendScopedRdsResponse(const std::vector<std::string>& resource_protos,
-                             const std::string& version) {
-    ASSERT(scoped_rds_upstream_info_.stream_ != nullptr);
+  void sendSotwScopedRdsResponse(const std::vector<std::string>& resource_protos,
+                                 const std::string& version) {
+    ASSERT(scoped_rds_upstream_info_.stream_by_resource_name_[srds_config_name_] != nullptr);
 
     envoy::api::v2::DiscoveryResponse response;
     response.set_version_info(version);
@@ -150,33 +248,29 @@ class ScopedRdsIntegrationTest : public HttpIntegrationTest,
       TestUtility::loadFromYaml(resource_proto, scoped_route_proto);
       response.add_resources()->PackFrom(scoped_route_proto);
     }
-
-    scoped_rds_upstream_info_.stream_->sendGrpcMessage(response);
+    scoped_rds_upstream_info_.stream_by_resource_name_[srds_config_name_]->sendGrpcMessage(
+        response);
   }
 
+  const std::string srds_config_name_{"foo-scoped-routes"};
   FakeUpstreamInfo scoped_rds_upstream_info_;
   FakeUpstreamInfo rds_upstream_info_;
 };
 
 INSTANTIATE_TEST_SUITE_P(IpVersionsAndGrpcTypes, ScopedRdsIntegrationTest,
-                         GRPC_CLIENT_INTEGRATION_PARAMS);
+                         DELTA_SOTW_GRPC_CLIENT_INTEGRATION_PARAMS);
 
 // Test that a SRDS DiscoveryResponse is successfully processed.
 TEST_P(ScopedRdsIntegrationTest, BasicSuccess) {
-  const std::string scope_route1 = R"EOF(
-name: foo_scope1
-route_configuration_name: foo_route1
+  const std::string scope_tmpl = R"EOF(
+name: {}
+route_configuration_name: {}
 key:
   fragments:
-    - string_key: x-foo-key
-)EOF";
-  const std::string scope_route2 = R"EOF(
-name: foo_scope2
-route_configuration_name: foo_route1
-key:
-  fragments:
-    - string_key: x-bar-key
+    - string_key: {}
 )EOF";
+  const std::string scope_route1 = fmt::format(scope_tmpl, "foo_scope1", "foo_route1", "foo-route");
+  const std::string scope_route2 = fmt::format(scope_tmpl, "foo_scope2", "foo_route1", "bar-route");
 
   const std::string route_config_tmpl = R"EOF(
       name: {}
@@ -190,35 +284,127 @@ route_configuration_name: foo_route1
 
   on_server_init_function_ = [&]() {
     createScopedRdsStream();
-    sendScopedRdsResponse({scope_route1, scope_route2}, "1");
-    createRdsStream();
-    sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_foo_1"), "1");
-    sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_foo_2"), "2");
+    sendSrdsResponse({scope_route1, scope_route2}, {scope_route1, scope_route2}, {}, "1");
+    createRdsStream("foo_route1");
+    // CreateRdsStream waits for connection which is fired by RDS subscription.
+    sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_0"), "1");
   };
   initialize();
-  test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_attempt", 2);
+  registerTestServerPorts({"http"});
+
+  // No scope key matches "xyz-route".
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  auto response = codec_client_->makeHeaderOnlyRequest(
+      Http::TestHeaderMapImpl{{":method", "GET"},
+                              {":path", "/meh"},
+                              {":authority", "host"},
+                              {":scheme", "http"},
+                              {"Addr", "x-foo-key=xyz-route"}});
+  response->waitForEndStream();
+  verifyResponse(std::move(response), "404", Http::TestHeaderMapImpl{}, "route scope not found");
+  cleanupUpstreamAndDownstream();
+
+  // Test "foo-route" and 'bar-route' both gets routed to cluster_0.
+  test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_success", 1);
+  for (const std::string& scope_key : std::vector<std::string>{"foo-route", "bar-route"}) {
+    sendRequestAndVerifyResponse(
+        Http::TestHeaderMapImpl{{":method", "GET"},
+                                {":path", "/meh"},
+                                {":authority", "host"},
+                                {":scheme", "http"},
+                                {"Addr", fmt::format("x-foo-key={}", scope_key)}},
+        456, Http::TestHeaderMapImpl{{":status", "200"}, {"service", scope_key}}, 123,
+        /*cluster_0*/ 0);
+  }
+  test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_attempt",
+                                 // update_attempt only increase after a response
+                                 isDelta() ? 1 : 2);
   test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_success", 1);
   // The version gauge should be set to xxHash64("1").
   test_server_->waitForGaugeEq("http.config_test.scoped_rds.foo-scoped-routes.version",
                                13237225503670494420UL);
 
-  const std::string scope_route3 = R"EOF(
-name: foo_scope3
-route_configuration_name: foo_route1
-key:
-  fragments:
-    - string_key: x-baz-key
-)EOF";
-  sendScopedRdsResponse({scope_route3}, "2");
+  // Add a new scope scope_route3 with a brand new RouteConfiguration foo_route2.
+  const std::string scope_route3 = fmt::format(scope_tmpl, "foo_scope3", "foo_route2", "baz-route");
+
+  sendSrdsResponse({scope_route1, scope_route2, scope_route3}, /*added*/ {scope_route3}, {}, "2");
+  test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_attempt", 2);
+  sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_1"), "3");
+  test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_success", 2);
+  createRdsStream("foo_route2");
+  test_server_->waitForCounterGe("http.config_test.rds.foo_route2.update_attempt", 1);
+  sendRdsResponse(fmt::format(route_config_tmpl, "foo_route2", "cluster_0"), "1");
+  test_server_->waitForCounterGe("http.config_test.rds.foo_route2.update_success", 1);
   test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_success", 2);
+  // The version gauge should be set to xxHash64("2").
   test_server_->waitForGaugeEq("http.config_test.scoped_rds.foo-scoped-routes.version",
                                6927017134761466251UL);
-  test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_attempt", 3);
-  sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_foo_3"), "3");
-  test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_success", 3);
-  // RDS updates won't affect SRDS.
-  test_server_->waitForGaugeEq("http.config_test.scoped_rds.foo-scoped-routes.version",
-                               6927017134761466251UL);
+  // After RDS update, requests within scope 'foo_scope1' or 'foo_scope2' get routed to
+  // 'cluster_1'.
+  for (const std::string& scope_key : std::vector<std::string>{"foo-route", "bar-route"}) {
+    sendRequestAndVerifyResponse(
+        Http::TestHeaderMapImpl{{":method", "GET"},
+                                {":path", "/meh"},
+                                {":authority", "host"},
+                                {":scheme", "http"},
+                                {"Addr", fmt::format("x-foo-key={}", scope_key)}},
+        456, Http::TestHeaderMapImpl{{":status", "200"}, {"service", scope_key}}, 123,
+        /*cluster_1*/ 1);
+  }
+  // Now requests within scope 'foo_scope3' get routed to 'cluster_0'.
+  sendRequestAndVerifyResponse(
+      Http::TestHeaderMapImpl{{":method", "GET"},
+                              {":path", "/meh"},
+                              {":authority", "host"},
+                              {":scheme", "http"},
+                              {"Addr", fmt::format("x-foo-key={}", "baz-route")}},
+      456, Http::TestHeaderMapImpl{{":status", "200"}, {"service", "bluh"}}, 123,
+      /*cluster_0*/ 0);
+
+  // Delete foo_scope1 and requests within the scope gets 400s.
+  sendSrdsResponse({scope_route2, scope_route3}, {}, {"foo_scope1"}, "3");
+  test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_success", 3);
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  response = codec_client_->makeHeaderOnlyRequest(
+      Http::TestHeaderMapImpl{{":method", "GET"},
+                              {":path", "/meh"},
+                              {":authority", "host"},
+                              {":scheme", "http"},
+                              {"Addr", "x-foo-key=foo-route"}});
+  response->waitForEndStream();
+  verifyResponse(std::move(response), "404", Http::TestHeaderMapImpl{}, "route scope not found");
+  cleanupUpstreamAndDownstream();
+  // Add a new scope foo_scope4.
+  const std::string& scope_route4 =
+      fmt::format(scope_tmpl, "foo_scope4", "foo_route4", "xyz-route");
+  sendSrdsResponse({scope_route3, scope_route2, scope_route4}, {scope_route4}, {}, "4");
+  test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_success", 4);
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  response = codec_client_->makeHeaderOnlyRequest(
+      Http::TestHeaderMapImpl{{":method", "GET"},
+                              {":path", "/meh"},
+                              {":authority", "host"},
+                              {":scheme", "http"},
+                              {"Addr", "x-foo-key=xyz-route"}});
+  response->waitForEndStream();
+  // Get 404 because RDS hasn't pushed route configuration "foo_route4" yet.
+  // But scope is found and the Router::NullConfigImpl is returned.
+  verifyResponse(std::move(response), "404", Http::TestHeaderMapImpl{}, "");
+  cleanupUpstreamAndDownstream();
+
+  // RDS updated foo_route4, requests with scope key "xyz-route" now hit cluster_1.
+  test_server_->waitForCounterGe("http.config_test.rds.foo_route4.update_attempt", 1);
+  createRdsStream("foo_route4");
+  sendRdsResponse(fmt::format(route_config_tmpl, "foo_route4", "cluster_1"), "3");
+  test_server_->waitForCounterGe("http.config_test.rds.foo_route4.update_success", 1);
+  sendRequestAndVerifyResponse(
+      Http::TestHeaderMapImpl{{":method", "GET"},
+                              {":path", "/meh"},
+                              {":authority", "host"},
+                              {":scheme", "http"},
+                              {"Addr", "x-foo-key=xyz-route"}},
+      456, Http::TestHeaderMapImpl{{":status", "200"}, {"service", "xyz-route"}}, 123,
+      /*cluster_1 */ 1);
 }
 
 // Test that a bad config update updates the corresponding stats.
@@ -229,16 +415,56 @@ TEST_P(ScopedRdsIntegrationTest, ConfigUpdateFailure) {
 route_configuration_name: foo_route1
 key:
   fragments:
-    - string_key: x-foo-key
+    - string_key: foo
 )EOF";
   on_server_init_function_ = [this, &scope_route1]() {
     createScopedRdsStream();
-    sendScopedRdsResponse({scope_route1}, "1");
+    sendSrdsResponse({scope_route1}, {scope_route1}, {}, "1");
   };
   initialize();
 
   test_server_->waitForCounterGe("http.config_test.scoped_rds.foo-scoped-routes.update_rejected",
                                  1);
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+  auto response =
+      codec_client_->makeHeaderOnlyRequest(Http::TestHeaderMapImpl{{":method", "GET"},
+                                                                   {":path", "/meh"},
+                                                                   {":authority", "host"},
+                                                                   {":scheme", "http"},
+                                                                   {"Addr", "x-foo-key=foo"}});
+  response->waitForEndStream();
+  verifyResponse(std::move(response), "404", Http::TestHeaderMapImpl{}, "route scope not found");
+  cleanupUpstreamAndDownstream();
+
+  // SRDS update fixed the problem.
+  const std::string scope_route2 = R"EOF(
+name: foo_scope1
+route_configuration_name: foo_route1
+key:
+  fragments:
+    - string_key: foo
+)EOF";
+  sendSrdsResponse({scope_route2}, {scope_route2}, {}, "1");
+  test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_attempt", 1);
+  createRdsStream("foo_route1");
+  const std::string route_config_tmpl = R"EOF(
+      name: {}
+      virtual_hosts:
+      - name: integration
+        domains: ["*"]
+        routes:
+        - match: {{ prefix: "/" }}
+          route: {{ cluster: {} }}
+)EOF";
+  sendRdsResponse(fmt::format(route_config_tmpl, "foo_route1", "cluster_0"), "1");
+  test_server_->waitForCounterGe("http.config_test.rds.foo_route1.update_success", 1);
+  sendRequestAndVerifyResponse(
+      Http::TestHeaderMapImpl{{":method", "GET"},
+                              {":path", "/meh"},
+                              {":authority", "host"},
+                              {":scheme", "http"},
+                              {"Addr", "x-foo-key=foo"}},
+      456, Http::TestHeaderMapImpl{{":status", "200"}, {"service", "bluh"}}, 123, /*cluster_0*/ 0);
 }
 
 } // namespace
diff --git a/test/mocks/router/mocks.h b/test/mocks/router/mocks.h
index 06e5875c032a..b4e7c4ff7700 100644
--- a/test/mocks/router/mocks.h
+++ b/test/mocks/router/mocks.h
@@ -391,6 +391,7 @@ class MockRouteConfigProvider : public RouteConfigProvider {
   MOCK_CONST_METHOD0(configInfo, absl::optional<ConfigInfo>());
   MOCK_CONST_METHOD0(lastUpdated, SystemTime());
   MOCK_METHOD0(onConfigUpdate, void());
+  MOCK_CONST_METHOD1(validateConfig, void(const envoy::api::v2::RouteConfiguration&));
 
   std::shared_ptr<NiceMock<MockConfig>> route_config_{new NiceMock<MockConfig>()};
 };
diff --git a/test/test_common/utility.cc b/test/test_common/utility.cc
index 8cf850b1a300..c236ac4ce7d4 100644
--- a/test/test_common/utility.cc
+++ b/test/test_common/utility.cc
@@ -423,9 +423,11 @@ void TestHeaderMapImpl::addCopy(const std::string& key, const std::string& value
 
 void TestHeaderMapImpl::remove(const std::string& key) { remove(LowerCaseString(key)); }
 
-std::string TestHeaderMapImpl::get_(const std::string& key) { return get_(LowerCaseString(key)); }
+std::string TestHeaderMapImpl::get_(const std::string& key) const {
+  return get_(LowerCaseString(key));
+}
 
-std::string TestHeaderMapImpl::get_(const LowerCaseString& key) {
+std::string TestHeaderMapImpl::get_(const LowerCaseString& key) const {
   const HeaderEntry* header = get(key);
   if (!header) {
     return EMPTY_STRING;
diff --git a/test/test_common/utility.h b/test/test_common/utility.h
index 4ff0ac413097..b4f048396ac6 100644
--- a/test/test_common/utility.h
+++ b/test/test_common/utility.h
@@ -637,8 +637,8 @@ class TestHeaderMapImpl : public HeaderMapImpl {
   using HeaderMapImpl::remove;
   void addCopy(const std::string& key, const std::string& value);
   void remove(const std::string& key);
-  std::string get_(const std::string& key);
-  std::string get_(const LowerCaseString& key);
+  std::string get_(const std::string& key) const;
+  std::string get_(const LowerCaseString& key) const;
   bool has(const std::string& key);
   bool has(const LowerCaseString& key);
 };