diff --git a/include/envoy/config/config_provider_manager.h b/include/envoy/config/config_provider_manager.h
index 9c9dce88d1071..eaf6d4d0ec114 100644
--- a/include/envoy/config/config_provider_manager.h
+++ b/include/envoy/config/config_provider_manager.h
@@ -44,6 +44,7 @@ class ConfigProviderManager {
    * and Config are shared amongst all providers relying on the same config source.
    * @param config_source_proto supplies the proto containing the xDS API configuration.
    * @param factory_context is the context to use for the provider.
+   * @param init_manager is the Init::Manager to use for the provider.
    * @param stat_prefix supplies the prefix to use for statistics.
    * @param optarg supplies an optional argument with data specific to the concrete class.
    * @return ConfigProviderPtr a newly allocated dynamic config provider which shares underlying
@@ -52,8 +53,9 @@ class ConfigProviderManager {
    */
   virtual ConfigProviderPtr
   createXdsConfigProvider(const Protobuf::Message& config_source_proto,
-                          Server::Configuration::FactoryContext& factory_context,
-                          const std::string& stat_prefix, const OptionalArg& optarg) PURE;
+                          Server::Configuration::ServerFactoryContext& factory_context,
+                          Init::Manager& init_manager, const std::string& stat_prefix,
+                          const OptionalArg& optarg) PURE;
 
   /**
    * Returns a ConfigProvider associated with a statically specified configuration.
@@ -64,7 +66,7 @@ class ConfigProviderManager {
    */
   virtual ConfigProviderPtr
   createStaticConfigProvider(const Protobuf::Message& config_proto,
-                             Server::Configuration::FactoryContext& factory_context,
+                             Server::Configuration::ServerFactoryContext& factory_context,
                              const OptionalArg& optarg) {
     UNREFERENCED_PARAMETER(config_proto);
     UNREFERENCED_PARAMETER(factory_context);
@@ -82,7 +84,7 @@ class ConfigProviderManager {
    */
   virtual ConfigProviderPtr
   createStaticConfigProvider(ProtobufTypes::ConstMessagePtrVector&& config_protos,
-                             Server::Configuration::FactoryContext& factory_context,
+                             Server::Configuration::ServerFactoryContext& factory_context,
                              const OptionalArg& optarg) {
     UNREFERENCED_PARAMETER(config_protos);
     UNREFERENCED_PARAMETER(factory_context);
diff --git a/include/envoy/router/route_config_provider_manager.h b/include/envoy/router/route_config_provider_manager.h
index 96e34322a6edf..f266407a36515 100644
--- a/include/envoy/router/route_config_provider_manager.h
+++ b/include/envoy/router/route_config_provider_manager.h
@@ -39,19 +39,20 @@ class RouteConfigProviderManager {
    */
   virtual RouteConfigProviderSharedPtr createRdsRouteConfigProvider(
       const envoy::extensions::filters::network::http_connection_manager::v3::Rds& rds,
-      Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
+      Server::Configuration::ServerFactoryContext& factory_context, const std::string& stat_prefix,
       Init::Manager& init_manager) PURE;
 
   /**
    * Get a RouteConfigSharedPtr for a statically defined route. Ownership is as described for
    * getRdsRouteConfigProvider above. This method always create a new RouteConfigProvider.
    * @param route_config supplies the RouteConfiguration for this route
-   * @param runtime supplies the runtime loader.
-   * @param cm supplies the ClusterManager.
+   * @param factory_context is the context to use for the route config provider.
+   * @param validator is the message validator for route config.
    */
   virtual RouteConfigProviderPtr
   createStaticRouteConfigProvider(const envoy::config::route::v3::RouteConfiguration& route_config,
-                                  Server::Configuration::FactoryContext& factory_context) PURE;
+                                  Server::Configuration::ServerFactoryContext& factory_context,
+                                  ProtobufMessage::ValidationVisitor& validator) PURE;
 };
 
 } // namespace Router
diff --git a/include/envoy/server/filter_config.h b/include/envoy/server/filter_config.h
index 8cc47f2d2a2c9..c3afd9cb00cc5 100644
--- a/include/envoy/server/filter_config.h
+++ b/include/envoy/server/filter_config.h
@@ -55,6 +55,12 @@ class CommonFactoryContext {
    */
   virtual const LocalInfo::LocalInfo& localInfo() const PURE;
 
+  /**
+   * @return ProtobufMessage::ValidationContext& validation visitor for xDS and static configuration
+   *         messages.
+   */
+  virtual ProtobufMessage::ValidationContext& messageValidationContext() PURE;
+
   /**
    * @return RandomGenerator& the random generator for the server.
    */
diff --git a/source/common/config/BUILD b/source/common/config/BUILD
index f3faacc16391d..f5528c140a07a 100644
--- a/source/common/config/BUILD
+++ b/source/common/config/BUILD
@@ -36,7 +36,9 @@ envoy_cc_library(
         "//include/envoy/server:config_tracker_interface",
         "//include/envoy/singleton:instance_interface",
         "//include/envoy/thread_local:thread_local_interface",
+        "//source/common/init:manager_lib",
         "//source/common/init:target_lib",
+        "//source/common/init:watcher_lib",
         "//source/common/protobuf",
     ],
 )
diff --git a/source/common/config/config_provider_impl.cc b/source/common/config/config_provider_impl.cc
index 5745647e2dbf5..3be865d2642d2 100644
--- a/source/common/config/config_provider_impl.cc
+++ b/source/common/config/config_provider_impl.cc
@@ -4,7 +4,7 @@ namespace Envoy {
 namespace Config {
 
 ImmutableConfigProviderBase::ImmutableConfigProviderBase(
-    Server::Configuration::FactoryContext& factory_context,
+    Server::Configuration::ServerFactoryContext& factory_context,
     ConfigProviderManagerImplBase& config_provider_manager,
     ConfigProviderInstanceType instance_type, ApiType api_type)
     : last_updated_(factory_context.timeSource().systemTime()),
@@ -20,7 +20,7 @@ ImmutableConfigProviderBase::~ImmutableConfigProviderBase() {
 }
 
 ConfigSubscriptionCommonBase::~ConfigSubscriptionCommonBase() {
-  init_target_.ready();
+  local_init_target_.ready();
   config_provider_manager_.unbindSubscription(manager_identifier_);
 }
 
diff --git a/source/common/config/config_provider_impl.h b/source/common/config/config_provider_impl.h
index 76a06aceb4d07..157941124d29f 100644
--- a/source/common/config/config_provider_impl.h
+++ b/source/common/config/config_provider_impl.h
@@ -11,7 +11,9 @@
 #include "common/common/thread.h"
 #include "common/common/utility.h"
 #include "common/config/utility.h"
+#include "common/init/manager_impl.h"
 #include "common/init/target_impl.h"
+#include "common/init/watcher_impl.h"
 #include "common/protobuf/protobuf.h"
 
 namespace Envoy {
@@ -115,7 +117,7 @@ class ImmutableConfigProviderBase : public ConfigProvider {
   ConfigProviderInstanceType instanceType() const { return instance_type_; }
 
 protected:
-  ImmutableConfigProviderBase(Server::Configuration::FactoryContext& factory_context,
+  ImmutableConfigProviderBase(Server::Configuration::ServerFactoryContext& factory_context,
                               ConfigProviderManagerImplBase& config_provider_manager,
                               ConfigProviderInstanceType instance_type, ApiType api_type);
 
@@ -137,12 +139,6 @@ class MutableConfigProviderCommonBase;
  * A subscription is intended to be co-owned by config providers with the same config source, it's
  * designed to be created/destructed on admin thread only.
  *
- * xDS config providers and subscriptions are split to avoid lifetime issues with arguments
- * required by the config providers. An example is the Server::Configuration::FactoryContext, which
- * is owned by listeners and therefore may be destroyed while an associated config provider is still
- * in use (see #3960). This split enables single ownership of the config providers, while enabling
- * shared ownership of the underlying subscription.
- *
  */
 class ConfigSubscriptionCommonBase : protected Logger::Loggable<Logger::Id::config> {
 public:
@@ -180,7 +176,7 @@ class ConfigSubscriptionCommonBase : protected Logger::Loggable<Logger::Id::conf
    */
   void onConfigUpdate() {
     setLastUpdated();
-    init_target_.ready();
+    local_init_target_.ready();
   }
 
   /**
@@ -189,7 +185,7 @@ class ConfigSubscriptionCommonBase : protected Logger::Loggable<Logger::Id::conf
    */
   void onConfigUpdateFailed() {
     setLastUpdated();
-    init_target_.ready();
+    local_init_target_.ready();
   }
 
 protected:
@@ -202,13 +198,22 @@ class ConfigSubscriptionCommonBase : protected Logger::Loggable<Logger::Id::conf
 
   ConfigSubscriptionCommonBase(const std::string& name, const uint64_t manager_identifier,
                                ConfigProviderManagerImplBase& config_provider_manager,
-                               Server::Configuration::FactoryContext& factory_context)
+                               Server::Configuration::ServerFactoryContext& factory_context)
       : name_(name), tls_(factory_context.threadLocal().allocateSlot()),
-        init_target_(absl::StrCat("ConfigSubscriptionCommonBase ", name_), [this]() { start(); }),
+        local_init_target_(
+            fmt::format("ConfigSubscriptionCommonBase local init target '{}'", name_),
+            [this]() { start(); }),
+        parent_init_target_(fmt::format("ConfigSubscriptionCommonBase init target '{}'", name_),
+                            [this]() { local_init_manager_.initialize(local_init_watcher_); }),
+        local_init_watcher_(fmt::format("ConfigSubscriptionCommonBase local watcher '{}'", name_),
+                            [this]() { parent_init_target_.ready(); }),
+        local_init_manager_(
+            fmt::format("ConfigSubscriptionCommonBase local init manager '{}'", name_)),
         manager_identifier_(manager_identifier), config_provider_manager_(config_provider_manager),
         time_source_(factory_context.timeSource()),
         last_updated_(factory_context.timeSource().systemTime()) {
     Envoy::Config::Utility::checkLocalInfo(name, factory_context.localInfo());
+    local_init_manager_.add(local_init_target_);
   }
 
   /**
@@ -220,7 +225,7 @@ class ConfigSubscriptionCommonBase : protected Logger::Loggable<Logger::Id::conf
   void applyConfigUpdate(const ConfigUpdateCb& update_fn);
 
   void setLastUpdated() { last_updated_ = time_source_.systemTime(); }
-
+  Init::Manager& localInitManager() { return local_init_manager_; }
   void setLastConfigInfo(absl::optional<LastConfigInfo>&& config_info) {
     config_info_ = std::move(config_info);
   }
@@ -232,7 +237,16 @@ class ConfigSubscriptionCommonBase : protected Logger::Loggable<Logger::Id::conf
   ThreadLocal::SlotPtr tls_;
 
 private:
-  Init::TargetImpl init_target_;
+  // Local init target which signals first RPC interaction with management server.
+  Init::TargetImpl local_init_target_;
+  // Target added to factory context's initManager.
+  Init::TargetImpl parent_init_target_;
+  // Watcher that marks parent_init_target_ ready when the local init manager is ready.
+  Init::WatcherImpl local_init_watcher_;
+  // Local manager that tracks the subscription initialization, it is also used for sub-resource
+  // initialization if the sub-resource is not initialized.
+  Init::ManagerImpl local_init_manager_;
+
   const uint64_t manager_identifier_;
   ConfigProviderManagerImplBase& config_provider_manager_;
   TimeSource& time_source_;
@@ -259,7 +273,7 @@ class ConfigSubscriptionInstance : public ConfigSubscriptionCommonBase {
 public:
   ConfigSubscriptionInstance(const std::string& name, const uint64_t manager_identifier,
                              ConfigProviderManagerImplBase& config_provider_manager,
-                             Server::Configuration::FactoryContext& factory_context)
+                             Server::Configuration::ServerFactoryContext& factory_context)
       : ConfigSubscriptionCommonBase(name, manager_identifier, config_provider_manager,
                                      factory_context) {}
 
@@ -419,7 +433,7 @@ class ConfigProviderManagerImplBase : public ConfigProviderManager, public Singl
       // around it. However, since this is not a performance critical path we err on the side
       // of simplicity.
       subscription = subscription_factory_fn(manager_identifier, *this);
-      init_manager.add(subscription->init_target_);
+      init_manager.add(subscription->parent_init_target_);
 
       bindSubscription(manager_identifier, subscription);
     } else {
diff --git a/source/common/router/rds_impl.cc b/source/common/router/rds_impl.cc
index 5fb324ed42ba0..11fdc8c4cbe57 100644
--- a/source/common/router/rds_impl.cc
+++ b/source/common/router/rds_impl.cc
@@ -28,17 +28,20 @@ namespace Router {
 RouteConfigProviderSharedPtr RouteConfigProviderUtil::create(
     const envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
         config,
-    Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
-    RouteConfigProviderManager& route_config_provider_manager) {
+    Server::Configuration::ServerFactoryContext& factory_context,
+    ProtobufMessage::ValidationVisitor& validator, Init::Manager& init_manager,
+    const std::string& stat_prefix, RouteConfigProviderManager& route_config_provider_manager) {
   switch (config.route_specifier_case()) {
   case envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager::
       RouteSpecifierCase::kRouteConfig:
-    return route_config_provider_manager.createStaticRouteConfigProvider(config.route_config(),
-                                                                         factory_context);
+    return route_config_provider_manager.createStaticRouteConfigProvider(
+        config.route_config(), factory_context, validator);
   case envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager::
       RouteSpecifierCase::kRds:
     return route_config_provider_manager.createRdsRouteConfigProvider(
-        config.rds(), factory_context, stat_prefix, factory_context.initManager());
+        // At the creation of a RDS route config provider, the factory_context's initManager is
+        // always valid, though the init manager may go away later when the listener goes away.
+        config.rds(), factory_context, stat_prefix, init_manager);
   default:
     NOT_REACHED_GCOVR_EXCL_LINE;
   }
@@ -46,11 +49,10 @@ RouteConfigProviderSharedPtr RouteConfigProviderUtil::create(
 
 StaticRouteConfigProviderImpl::StaticRouteConfigProviderImpl(
     const envoy::config::route::v3::RouteConfiguration& config,
-    Server::Configuration::FactoryContext& factory_context,
+    Server::Configuration::ServerFactoryContext& factory_context,
+    ProtobufMessage::ValidationVisitor& validator,
     RouteConfigProviderManagerImpl& route_config_provider_manager)
-    : config_(new ConfigImpl(config, factory_context.getServerFactoryContext(),
-                             factory_context.messageValidationVisitor(), true)),
-
+    : config_(new ConfigImpl(config, factory_context, validator, true)),
       route_config_proto_{config}, last_updated_(factory_context.timeSource().systemTime()),
       route_config_provider_manager_(route_config_provider_manager) {
   route_config_provider_manager_.static_route_config_providers_.insert(this);
@@ -64,13 +66,18 @@ StaticRouteConfigProviderImpl::~StaticRouteConfigProviderImpl() {
 RdsRouteConfigSubscription::RdsRouteConfigSubscription(
     const envoy::extensions::filters::network::http_connection_manager::v3::Rds& rds,
     const uint64_t manager_identifier, Server::Configuration::ServerFactoryContext& factory_context,
-    ProtobufMessage::ValidationVisitor& validator, Init::Manager& init_manager,
     const std::string& stat_prefix,
     Envoy::Router::RouteConfigProviderManagerImpl& route_config_provider_manager)
     : route_config_name_(rds.route_config_name()), factory_context_(factory_context),
-      validator_(validator), init_manager_(init_manager),
-      init_target_(fmt::format("RdsRouteConfigSubscription {}", route_config_name_),
-                   [this]() { subscription_->start({route_config_name_}); }),
+      validator_(factory_context.messageValidationContext().dynamicValidationVisitor()),
+      parent_init_target_(fmt::format("RdsRouteConfigSubscription init {}", route_config_name_),
+                          [this]() { local_init_manager_.initialize(local_init_watcher_); }),
+      local_init_watcher_(fmt::format("RDS local-init-watcher {}", rds.route_config_name()),
+                          [this]() { parent_init_target_.ready(); }),
+      local_init_target_(
+          fmt::format("RdsRouteConfigSubscription local-init-target {}", route_config_name_),
+          [this]() { subscription_->start({route_config_name_}); }),
+      local_init_manager_(fmt::format("RDS local-init-manager {}", route_config_name_)),
       scope_(factory_context.scope().createScope(stat_prefix + "rds." + route_config_name_ + ".")),
       stat_prefix_(stat_prefix), stats_({ALL_RDS_STATS(POOL_COUNTER(*scope_))}),
       route_config_provider_manager_(route_config_provider_manager),
@@ -80,13 +87,14 @@ RdsRouteConfigSubscription::RdsRouteConfigSubscription(
       factory_context.clusterManager().subscriptionFactory().subscriptionFromConfigSource(
           rds.config_source(), loadTypeUrl(rds.config_source().resource_api_version()), *scope_,
           *this);
+  local_init_manager_.add(local_init_target_);
   config_update_info_ =
-      std::make_unique<RouteConfigUpdateReceiverImpl>(factory_context.timeSource(), validator);
+      std::make_unique<RouteConfigUpdateReceiverImpl>(factory_context.timeSource(), validator_);
 }
 
 RdsRouteConfigSubscription::~RdsRouteConfigSubscription() {
   // If we get destroyed during initialization, make sure we signal that we "initialized".
-  init_target_.ready();
+  local_init_target_.ready();
 
   // The ownership of RdsRouteConfigProviderImpl is shared among all HttpConnectionManagers that
   // hold a shared_ptr to it. The RouteConfigProviderManager holds weak_ptrs to the
@@ -113,12 +121,10 @@ void RdsRouteConfigSubscription::onConfigUpdate(
     // especially when it comes with per_filter_config,
     provider->validateConfig(route_config);
   }
-
   std::unique_ptr<Init::ManagerImpl> noop_init_manager;
   std::unique_ptr<Cleanup> resume_rds;
   if (config_update_info_->onRdsUpdate(route_config, version_info)) {
     stats_.config_reload_.inc();
-
     if (config_update_info_->routeConfiguration().has_vhds() &&
         config_update_info_->vhdsConfigurationChanged()) {
       ENVOY_LOG(
@@ -130,7 +136,7 @@ void RdsRouteConfigSubscription::onConfigUpdate(
           config_update_info_, factory_context_, stat_prefix_, route_config_providers_,
           config_update_info_->routeConfiguration().vhds().config_source().resource_api_version());
       vhds_subscription_->registerInitTargetWithInitManager(
-          noop_init_manager == nullptr ? getRdsConfigInitManager() : *noop_init_manager);
+          noop_init_manager == nullptr ? local_init_manager_ : *noop_init_manager);
     } else {
       ENVOY_LOG(debug, "rds: loading new configuration: config_name={} hash={}", route_config_name_,
                 config_update_info_->configHash());
@@ -146,7 +152,7 @@ void RdsRouteConfigSubscription::onConfigUpdate(
     update_callback_manager_.runCallbacks();
   }
 
-  init_target_.ready();
+  local_init_target_.ready();
 }
 
 // Initialize a no-op InitManager in case the one in the factory_context has completed
@@ -155,7 +161,7 @@ void RdsRouteConfigSubscription::onConfigUpdate(
 void RdsRouteConfigSubscription::maybeCreateInitManager(
     const std::string& version_info, std::unique_ptr<Init::ManagerImpl>& init_manager,
     std::unique_ptr<Cleanup>& init_vhds) {
-  if (getRdsConfigInitManager().state() == Init::Manager::State::Initialized) {
+  if (local_init_manager_.state() == Init::Manager::State::Initialized) {
     init_manager = std::make_unique<Init::ManagerImpl>(
         fmt::format("VHDS {}:{}", route_config_name_, version_info));
     init_vhds = std::make_unique<Cleanup>([this, &init_manager, version_info] {
@@ -174,8 +180,8 @@ void RdsRouteConfigSubscription::onConfigUpdate(
     const Protobuf::RepeatedPtrField<envoy::service::discovery::v3::Resource>& added_resources,
     const Protobuf::RepeatedPtrField<std::string>& removed_resources, const std::string&) {
   if (!removed_resources.empty()) {
-    // TODO(#2500) when on-demand resource loading is supported, an RDS removal may make sense (see
-    // discussion in #6879), and so we should do something other than ignoring here.
+    // TODO(#2500) when on-demand resource loading is supported, an RDS removal may make sense
+    // (see discussion in #6879), and so we should do something other than ignoring here.
     ENVOY_LOG(
         error,
         "Server sent a delta RDS update attempting to remove a resource (name: {}). Ignoring.",
@@ -193,7 +199,7 @@ void RdsRouteConfigSubscription::onConfigUpdateFailed(
   ASSERT(Envoy::Config::ConfigUpdateFailureReason::ConnectionFailure != reason);
   // We need to allow server startup to continue, even if we have a bad
   // config.
-  init_target_.ready();
+  local_init_target_.ready();
 }
 
 void RdsRouteConfigSubscription::updateOnDemand(const std::string& aliases) {
@@ -207,7 +213,7 @@ bool RdsRouteConfigSubscription::validateUpdateSize(int num_resources) {
   if (num_resources == 0) {
     ENVOY_LOG(debug, "Missing RouteConfiguration for {} in onConfigUpdate()", route_config_name_);
     stats_.update_empty_.inc();
-    init_target_.ready();
+    local_init_target_.ready();
     return false;
   }
   if (num_resources != 1) {
@@ -235,11 +241,10 @@ RdsRouteConfigSubscription::loadTypeUrl(envoy::config::core::v3::ApiVersion reso
 
 RdsRouteConfigProviderImpl::RdsRouteConfigProviderImpl(
     RdsRouteConfigSubscriptionSharedPtr&& subscription,
-    Server::Configuration::FactoryContext& factory_context)
+    Server::Configuration::ServerFactoryContext& factory_context)
     : subscription_(std::move(subscription)),
-      config_update_info_(subscription_->routeConfigUpdate()),
-      factory_context_(factory_context.getServerFactoryContext()),
-      validator_(factory_context.messageValidationVisitor()),
+      config_update_info_(subscription_->routeConfigUpdate()), factory_context_(factory_context),
+      validator_(factory_context.messageValidationContext().dynamicValidationVisitor()),
       tls_(factory_context.threadLocal().allocateSlot()) {
   ConfigConstSharedPtr initial_config;
   if (config_update_info_->configInfo().has_value()) {
@@ -336,7 +341,7 @@ RouteConfigProviderManagerImpl::RouteConfigProviderManagerImpl(Server::Admin& ad
 
 Router::RouteConfigProviderSharedPtr RouteConfigProviderManagerImpl::createRdsRouteConfigProvider(
     const envoy::extensions::filters::network::http_connection_manager::v3::Rds& rds,
-    Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
+    Server::Configuration::ServerFactoryContext& factory_context, const std::string& stat_prefix,
     Init::Manager& init_manager) {
   // RdsRouteConfigSubscriptions are unique based on their serialized RDS config.
   const uint64_t manager_identifier = MessageUtil::hash(rds);
@@ -347,10 +352,8 @@ Router::RouteConfigProviderSharedPtr RouteConfigProviderManagerImpl::createRdsRo
     // around it. However, since this is not a performance critical path we err on the side
     // of simplicity.
     RdsRouteConfigSubscriptionSharedPtr subscription(new RdsRouteConfigSubscription(
-        rds, manager_identifier, factory_context.getServerFactoryContext(),
-        factory_context.messageValidationVisitor(), factory_context.initManager(), stat_prefix,
-        *this));
-    init_manager.add(subscription->init_target_);
+        rds, manager_identifier, factory_context, stat_prefix, *this));
+    init_manager.add(subscription->parent_init_target_);
     std::shared_ptr<RdsRouteConfigProviderImpl> new_provider{
         new RdsRouteConfigProviderImpl(std::move(subscription), factory_context)};
     dynamic_route_config_providers_.insert(
@@ -363,16 +366,17 @@ Router::RouteConfigProviderSharedPtr RouteConfigProviderManagerImpl::createRdsRo
     auto existing_provider = it->second.lock();
     RELEASE_ASSERT(existing_provider != nullptr,
                    absl::StrCat("cannot find subscribed rds resource ", rds.route_config_name()));
-    init_manager.add(existing_provider->subscription_->init_target_);
+    init_manager.add(existing_provider->subscription_->parent_init_target_);
     return existing_provider;
   }
 }
 
 RouteConfigProviderPtr RouteConfigProviderManagerImpl::createStaticRouteConfigProvider(
     const envoy::config::route::v3::RouteConfiguration& route_config,
-    Server::Configuration::FactoryContext& factory_context) {
-  auto provider =
-      std::make_unique<StaticRouteConfigProviderImpl>(route_config, factory_context, *this);
+    Server::Configuration::ServerFactoryContext& factory_context,
+    ProtobufMessage::ValidationVisitor& validator) {
+  auto provider = std::make_unique<StaticRouteConfigProviderImpl>(route_config, factory_context,
+                                                                  validator, *this);
   static_route_config_providers_.insert(provider.get());
   return provider;
 }
diff --git a/source/common/router/rds_impl.h b/source/common/router/rds_impl.h
index 1f244722cf2ab..f5121a4a8c419 100644
--- a/source/common/router/rds_impl.h
+++ b/source/common/router/rds_impl.h
@@ -53,8 +53,9 @@ class RouteConfigProviderUtil {
   static RouteConfigProviderSharedPtr create(
       const envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
           config,
-      Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
-      RouteConfigProviderManager& route_config_provider_manager);
+      Server::Configuration::ServerFactoryContext& factory_context,
+      ProtobufMessage::ValidationVisitor& validator, Init::Manager& init_manager,
+      const std::string& stat_prefix, RouteConfigProviderManager& route_config_provider_manager);
 };
 
 class RouteConfigProviderManagerImpl;
@@ -65,7 +66,8 @@ class RouteConfigProviderManagerImpl;
 class StaticRouteConfigProviderImpl : public RouteConfigProvider {
 public:
   StaticRouteConfigProviderImpl(const envoy::config::route::v3::RouteConfiguration& config,
-                                Server::Configuration::FactoryContext& factory_context,
+                                Server::Configuration::ServerFactoryContext& factory_context,
+                                ProtobufMessage::ValidationVisitor& validator,
                                 RouteConfigProviderManagerImpl& route_config_provider_manager);
   ~StaticRouteConfigProviderImpl() override;
 
@@ -145,22 +147,25 @@ class RdsRouteConfigSubscription : Envoy::Config::SubscriptionCallbacks,
   RdsRouteConfigSubscription(
       const envoy::extensions::filters::network::http_connection_manager::v3::Rds& rds,
       const uint64_t manager_identifier,
-      Server::Configuration::ServerFactoryContext& factory_context,
-      ProtobufMessage::ValidationVisitor& validator, Init::Manager& init_manager,
-      const std::string& stat_prefix,
+      Server::Configuration::ServerFactoryContext& factory_context, const std::string& stat_prefix,
       RouteConfigProviderManagerImpl& route_config_provider_manager);
 
   bool validateUpdateSize(int num_resources);
   static std::string loadTypeUrl(envoy::config::core::v3::ApiVersion resource_api_version);
 
-  Init::Manager& getRdsConfigInitManager() { return init_manager_; }
-
   std::unique_ptr<Envoy::Config::Subscription> subscription_;
   const std::string route_config_name_;
   Server::Configuration::ServerFactoryContext& factory_context_;
   ProtobufMessage::ValidationVisitor& validator_;
-  Init::Manager& init_manager_;
-  Init::SharedTargetImpl init_target_;
+
+  // Init target used to notify the parent init manager that the subscription [and its sub resource]
+  // is ready.
+  Init::SharedTargetImpl parent_init_target_;
+  // Init watcher on RDS and VHDS ready event. This watcher marks parent_init_target_ ready.
+  Init::WatcherImpl local_init_watcher_;
+  // Target which starts the RDS subscription.
+  Init::TargetImpl local_init_target_;
+  Init::ManagerImpl local_init_manager_;
   Stats::ScopePtr scope_;
   std::string stat_prefix_;
   RdsStats stats_;
@@ -215,7 +220,7 @@ class RdsRouteConfigProviderImpl : public RouteConfigProvider,
   };
 
   RdsRouteConfigProviderImpl(RdsRouteConfigSubscriptionSharedPtr&& subscription,
-                             Server::Configuration::FactoryContext& factory_context);
+                             Server::Configuration::ServerFactoryContext& factory_context);
 
   RdsRouteConfigSubscriptionSharedPtr subscription_;
   RouteConfigUpdatePtr& config_update_info_;
@@ -237,12 +242,13 @@ class RouteConfigProviderManagerImpl : public RouteConfigProviderManager,
   // RouteConfigProviderManager
   RouteConfigProviderSharedPtr createRdsRouteConfigProvider(
       const envoy::extensions::filters::network::http_connection_manager::v3::Rds& rds,
-      Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
+      Server::Configuration::ServerFactoryContext& factory_context, const std::string& stat_prefix,
       Init::Manager& init_manager) override;
 
   RouteConfigProviderPtr
   createStaticRouteConfigProvider(const envoy::config::route::v3::RouteConfiguration& route_config,
-                                  Server::Configuration::FactoryContext& factory_context) override;
+                                  Server::Configuration::ServerFactoryContext& factory_context,
+                                  ProtobufMessage::ValidationVisitor& validator) override;
 
 private:
   // TODO(jsedgwick) These two members are prime candidates for the owned-entry list/map
diff --git a/source/common/router/scoped_rds.cc b/source/common/router/scoped_rds.cc
index c6ddf526349df..08df35675ca27 100644
--- a/source/common/router/scoped_rds.cc
+++ b/source/common/router/scoped_rds.cc
@@ -35,8 +35,8 @@ namespace ScopedRoutesConfigProviderUtil {
 ConfigProviderPtr create(
     const envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
         config,
-    Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
-    ConfigProviderManager& scoped_routes_config_provider_manager) {
+    Server::Configuration::ServerFactoryContext& factory_context, Init::Manager& init_manager,
+    const std::string& stat_prefix, ConfigProviderManager& scoped_routes_config_provider_manager) {
   ASSERT(config.route_specifier_case() ==
          envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager::
              RouteSpecifierCase::kScopedRoutes);
@@ -59,7 +59,7 @@ ConfigProviderPtr create(
   case envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes::
       ConfigSpecifierCase::kScopedRds:
     return scoped_routes_config_provider_manager.createXdsConfigProvider(
-        config.scoped_routes().scoped_rds(), factory_context, stat_prefix,
+        config.scoped_routes().scoped_rds(), factory_context, init_manager, stat_prefix,
         ScopedRoutesConfigProviderManagerOptArg(config.scoped_routes().name(),
                                                 config.scoped_routes().rds_config_source(),
                                                 config.scoped_routes().scope_key_builder()));
@@ -73,7 +73,7 @@ ConfigProviderPtr create(
 
 InlineScopedRoutesConfigProvider::InlineScopedRoutesConfigProvider(
     ProtobufTypes::ConstMessagePtrVector&& config_protos, std::string name,
-    Server::Configuration::FactoryContext& factory_context,
+    Server::Configuration::ServerFactoryContext& factory_context,
     ScopedRoutesConfigProviderManager& config_provider_manager,
     envoy::config::core::v3::ConfigSource rds_config_source,
     envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes::ScopeKeyBuilder
@@ -92,7 +92,7 @@ ScopedRdsConfigSubscription::ScopedRdsConfigSubscription(
     const uint64_t manager_identifier, const std::string& name,
     const envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes::
         ScopeKeyBuilder& scope_key_builder,
-    Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
+    Server::Configuration::ServerFactoryContext& factory_context, const std::string& stat_prefix,
     envoy::config::core::v3::ConfigSource rds_config_source,
     RouteConfigProviderManager& route_config_provider_manager,
     ScopedRoutesConfigProviderManager& config_provider_manager)
@@ -102,8 +102,8 @@ ScopedRdsConfigSubscription::ScopedRdsConfigSubscription(
       scope_(factory_context.scope().createScope(stat_prefix + "scoped_rds." + name + ".")),
       stats_({ALL_SCOPED_RDS_STATS(POOL_COUNTER(*scope_))}),
       rds_config_source_(std::move(rds_config_source)),
-      validation_visitor_(factory_context.messageValidationVisitor()), stat_prefix_(stat_prefix),
-      route_config_provider_manager_(route_config_provider_manager) {
+      validation_visitor_(factory_context.messageValidationContext().dynamicValidationVisitor()),
+      stat_prefix_(stat_prefix), route_config_provider_manager_(route_config_provider_manager) {
   subscription_ =
       factory_context.clusterManager().subscriptionFactory().subscriptionFromConfigSource(
           scoped_rds.scoped_rds_config_source(),
@@ -223,28 +223,27 @@ void ScopedRdsConfigSubscription::onConfigUpdate(
   // NOTE: deletes are done before adds/updates.
 
   absl::flat_hash_map<std::string, ScopedRouteInfoConstSharedPtr> to_be_removed_scopes;
-  // If new route config sources come after the factory_context_.initManager()'s initialize() been
-  // called, that initManager can't accept new targets. Instead we use a local override which will
+  // If new route config sources come after the local init manager's initialize() been
+  // called, the init manager can't accept new targets. Instead we use a local override which will
   // start new subscriptions but not wait on them to be ready.
-  // NOTE: For now we use a local init-manager, in the future when Envoy supports on-demand xDS, we
-  // will probably make this init-manager as a member of the subscription.
   std::unique_ptr<Init::ManagerImpl> noop_init_manager;
   // NOTE: This should be defined after noop_init_manager as it depends on the
   // noop_init_manager.
   std::unique_ptr<Cleanup> resume_rds;
-  if (factory_context_.initManager().state() == Init::Manager::State::Initialized) {
+  // if local init manager is initialized, the parent init manager may have gone away.
+  if (localInitManager().state() == Init::Manager::State::Initialized) {
     noop_init_manager =
         std::make_unique<Init::ManagerImpl>(fmt::format("SRDS {}:{}", name_, version_info));
     // Pause RDS to not send a burst of RDS requests until we start all the new subscriptions.
-    // In the case if factory_context_.initManager() is uninitialized, RDS is already paused either
-    // by Server init or LDS init.
+    // In the case if factory_context_.init_manager() is uninitialized, RDS is already paused
+    // either by Server init or LDS init.
     if (factory_context_.clusterManager().adsMux()) {
       factory_context_.clusterManager().adsMux()->pause(
           Envoy::Config::TypeUrl::get().RouteConfiguration);
     }
     resume_rds = std::make_unique<Cleanup>([this, &noop_init_manager, version_info] {
-      // For new RDS subscriptions created after listener warming up, we don't wait for them to warm
-      // up.
+      // For new RDS subscriptions created after listener warming up, we don't wait for them to
+      // warm up.
       Init::WatcherImpl noop_watcher(
           // Note: we just throw it away.
           fmt::format("SRDS ConfigUpdate watcher {}:{}", name_, version_info),
@@ -266,10 +265,9 @@ void ScopedRdsConfigSubscription::onConfigUpdate(
   std::list<std::unique_ptr<ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper>>
       to_be_removed_rds_providers = removeScopes(removed_resources, version_info);
   bool any_applied =
-      addOrUpdateScopes(
-          added_resources,
-          (noop_init_manager == nullptr ? factory_context_.initManager() : *noop_init_manager),
-          version_info, exception_msgs) ||
+      addOrUpdateScopes(added_resources,
+                        (noop_init_manager == nullptr ? localInitManager() : *noop_init_manager),
+                        version_info, exception_msgs) ||
       !to_be_removed_rds_providers.empty();
   ConfigSubscriptionCommonBase::onConfigUpdate();
   if (any_applied) {
@@ -289,9 +287,9 @@ void ScopedRdsConfigSubscription::onRdsConfigUpdate(const std::string& scope_nam
          fmt::format("trying to update route config for non-existing scope {}", scope_name));
   auto new_scoped_route_info = std::make_shared<ScopedRouteInfo>(
       envoy::config::route::v3::ScopedRouteConfiguration(iter->second->configProto()),
-      std::make_shared<ConfigImpl>(rds_subscription.routeConfigUpdate()->routeConfiguration(),
-                                   factory_context_.getServerFactoryContext(),
-                                   factory_context_.messageValidationVisitor(), false));
+      std::make_shared<ConfigImpl>(
+          rds_subscription.routeConfigUpdate()->routeConfiguration(), factory_context_,
+          factory_context_.messageValidationContext().dynamicValidationVisitor(), false));
   applyConfigUpdate([new_scoped_route_info](ConfigProvider::ConfigConstSharedPtr config)
                         -> ConfigProvider::ConfigConstSharedPtr {
     auto* thread_local_scoped_config =
@@ -408,12 +406,12 @@ ProtobufTypes::MessagePtr ScopedRoutesConfigProviderManager::dumpConfigs() const
 
 ConfigProviderPtr ScopedRoutesConfigProviderManager::createXdsConfigProvider(
     const Protobuf::Message& config_source_proto,
-    Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
-    const ConfigProviderManager::OptionalArg& optarg) {
+    Server::Configuration::ServerFactoryContext& factory_context, Init::Manager& init_manager,
+    const std::string& stat_prefix, const ConfigProviderManager::OptionalArg& optarg) {
   const auto& typed_optarg = static_cast<const ScopedRoutesConfigProviderManagerOptArg&>(optarg);
   ScopedRdsConfigSubscriptionSharedPtr subscription =
       ConfigProviderManagerImplBase::getSubscription<ScopedRdsConfigSubscription>(
-          config_source_proto, factory_context.initManager(),
+          config_source_proto, init_manager,
           [&config_source_proto, &factory_context, &stat_prefix,
            &typed_optarg](const uint64_t manager_identifier,
                           ConfigProviderManagerImplBase& config_provider_manager)
@@ -435,7 +433,7 @@ ConfigProviderPtr ScopedRoutesConfigProviderManager::createXdsConfigProvider(
 
 ConfigProviderPtr ScopedRoutesConfigProviderManager::createStaticConfigProvider(
     ProtobufTypes::ConstMessagePtrVector&& config_protos,
-    Server::Configuration::FactoryContext& factory_context,
+    Server::Configuration::ServerFactoryContext& factory_context,
     const ConfigProviderManager::OptionalArg& optarg) {
   const auto& typed_optarg = static_cast<const ScopedRoutesConfigProviderManagerOptArg&>(optarg);
   return std::make_unique<InlineScopedRoutesConfigProvider>(
diff --git a/source/common/router/scoped_rds.h b/source/common/router/scoped_rds.h
index 891d0d3559cdf..372caef576cfe 100644
--- a/source/common/router/scoped_rds.h
+++ b/source/common/router/scoped_rds.h
@@ -27,7 +27,8 @@ namespace ScopedRoutesConfigProviderUtil {
 Envoy::Config::ConfigProviderPtr create(
     const envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
         config,
-    Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
+    Server::Configuration::ServerFactoryContext& factory_context, Init::Manager& init_manager,
+    const std::string& stat_prefix,
     Envoy::Config::ConfigProviderManager& scoped_routes_config_provider_manager);
 
 } // namespace ScopedRoutesConfigProviderUtil
@@ -39,7 +40,7 @@ class InlineScopedRoutesConfigProvider : public Envoy::Config::ImmutableConfigPr
 public:
   InlineScopedRoutesConfigProvider(ProtobufTypes::ConstMessagePtrVector&& config_protos,
                                    std::string name,
-                                   Server::Configuration::FactoryContext& factory_context,
+                                   Server::Configuration::ServerFactoryContext& factory_context,
                                    ScopedRoutesConfigProviderManager& config_provider_manager,
                                    envoy::config::core::v3::ConfigSource rds_config_source,
                                    envoy::extensions::filters::network::http_connection_manager::
@@ -95,7 +96,7 @@ class ScopedRdsConfigSubscription : public Envoy::Config::DeltaConfigSubscriptio
       const uint64_t manager_identifier, const std::string& name,
       const envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes::
           ScopeKeyBuilder& scope_key_builder,
-      Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
+      Server::Configuration::ServerFactoryContext& factory_context, const std::string& stat_prefix,
       envoy::config::core::v3::ConfigSource rds_config_source,
       RouteConfigProviderManager& route_config_provider_manager,
       ScopedRoutesConfigProviderManager& config_provider_manager);
@@ -170,13 +171,14 @@ class ScopedRdsConfigSubscription : public Envoy::Config::DeltaConfigSubscriptio
 
   // ScopedRouteInfo by scope name.
   ScopedRouteMap scoped_route_map_;
+
   // RdsRouteConfigProvider by scope name.
   absl::flat_hash_map<std::string, std::unique_ptr<RdsRouteConfigProviderHelper>>
       route_provider_by_scope_;
   // A map of (hash, scope-name), used to detect the key conflict between scopes.
   absl::flat_hash_map<uint64_t, std::string> scope_name_by_hash_;
   // For creating RDS subscriptions.
-  Server::Configuration::FactoryContext& factory_context_;
+  Server::Configuration::ServerFactoryContext& factory_context_;
   const std::string name_;
   std::unique_ptr<Envoy::Config::Subscription> subscription_;
   const envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes::
@@ -219,11 +221,11 @@ class ScopedRoutesConfigProviderManager : public Envoy::Config::ConfigProviderMa
   // Envoy::Config::ConfigProviderManager
   Envoy::Config::ConfigProviderPtr
   createXdsConfigProvider(const Protobuf::Message& config_source_proto,
-                          Server::Configuration::FactoryContext& factory_context,
-                          const std::string& stat_prefix,
+                          Server::Configuration::ServerFactoryContext& factory_context,
+                          Init::Manager& init_manager, const std::string& stat_prefix,
                           const Envoy::Config::ConfigProviderManager::OptionalArg& optarg) override;
   Envoy::Config::ConfigProviderPtr
-  createStaticConfigProvider(const Protobuf::Message&, Server::Configuration::FactoryContext&,
+  createStaticConfigProvider(const Protobuf::Message&, Server::Configuration::ServerFactoryContext&,
                              const Envoy::Config::ConfigProviderManager::OptionalArg&) override {
     ASSERT(false,
            "SRDS supports delta updates and requires the use of the createStaticConfigProvider() "
@@ -232,7 +234,7 @@ class ScopedRoutesConfigProviderManager : public Envoy::Config::ConfigProviderMa
   }
   Envoy::Config::ConfigProviderPtr createStaticConfigProvider(
       std::vector<std::unique_ptr<const Protobuf::Message>>&& config_protos,
-      Server::Configuration::FactoryContext& factory_context,
+      Server::Configuration::ServerFactoryContext& factory_context,
       const Envoy::Config::ConfigProviderManager::OptionalArg& optarg) override;
 
   RouteConfigProviderManager& route_config_provider_manager() {
diff --git a/source/common/upstream/upstream_impl.cc b/source/common/upstream/upstream_impl.cc
index f5e87d199e7be..37e12f1f2c1dd 100644
--- a/source/common/upstream/upstream_impl.cc
+++ b/source/common/upstream/upstream_impl.cc
@@ -272,6 +272,7 @@ Network::TransportSocketFactory& HostDescriptionImpl::resolveTransportSocketFact
   match.stats_.total_match_count_.inc();
   ENVOY_LOG(debug, "transport socket match, socket {} selected for host with address {}",
             match.name_, dest_address ? dest_address->asString() : "empty");
+
   return match.factory_;
 }
 
@@ -631,6 +632,10 @@ class FactoryContextImpl : public Server::Configuration::CommonFactoryContext {
   ThreadLocal::SlotAllocator& threadLocal() override { return tls_; }
   Server::Admin& admin() override { return admin_; }
   TimeSource& timeSource() override { return api().timeSource(); }
+  ProtobufMessage::ValidationContext& messageValidationContext() override {
+    // Not used.
+    NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
+  }
   Api::Api& api() override { return api_; }
 
 private:
diff --git a/source/extensions/filters/network/http_connection_manager/config.cc b/source/extensions/filters/network/http_connection_manager/config.cc
index b2ee828d1f7cd..fdda63afc7e56 100644
--- a/source/extensions/filters/network/http_connection_manager/config.cc
+++ b/source/extensions/filters/network/http_connection_manager/config.cc
@@ -218,12 +218,14 @@ HttpConnectionManagerConfig::HttpConnectionManagerConfig(
   case envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager::
       RouteSpecifierCase::kRouteConfig:
     route_config_provider_ = Router::RouteConfigProviderUtil::create(
-        config, context_, stats_prefix_, route_config_provider_manager_);
+        config, context_.getServerFactoryContext(), context_.messageValidationVisitor(),
+        context_.initManager(), stats_prefix_, route_config_provider_manager_);
     break;
   case envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager::
       RouteSpecifierCase::kScopedRoutes:
     scoped_routes_config_provider_ = Router::ScopedRoutesConfigProviderUtil::create(
-        config, context_, stats_prefix_, scoped_routes_config_provider_manager_);
+        config, context_.getServerFactoryContext(), context_.initManager(), stats_prefix_,
+        scoped_routes_config_provider_manager_);
     break;
   default:
     NOT_REACHED_GCOVR_EXCL_LINE;
diff --git a/source/server/BUILD b/source/server/BUILD
index 39d899f4856e5..c347334fa0ca0 100644
--- a/source/server/BUILD
+++ b/source/server/BUILD
@@ -322,6 +322,7 @@ envoy_cc_library(
         "//include/envoy/server:transport_socket_config_interface",
         "//source/common/config:utility_lib",
         "//source/common/init:manager_lib",
+        "//source/common/init:target_lib",
         "//source/common/network:connection_balancer_lib",
         "//source/common/network:listen_socket_lib",
         "//source/common/network:socket_option_factory_lib",
diff --git a/source/server/filter_chain_manager_impl.cc b/source/server/filter_chain_manager_impl.cc
index af11f35c486fd..c5a1a32dae5d2 100644
--- a/source/server/filter_chain_manager_impl.cc
+++ b/source/server/filter_chain_manager_impl.cc
@@ -53,6 +53,9 @@ envoy::config::core::v3::TrafficDirection FilterChainFactoryContextImpl::directi
   return parent_context_.direction();
 }
 
+ProtobufMessage::ValidationContext& FilterChainFactoryContextImpl::messageValidationContext() {
+  return parent_context_.messageValidationContext();
+}
 ProtobufMessage::ValidationVisitor& FilterChainFactoryContextImpl::messageValidationVisitor() {
   return parent_context_.messageValidationVisitor();
 }
@@ -607,6 +610,9 @@ OverloadManager& FactoryContextImpl::overloadManager() { return server_.overload
 ThreadLocal::SlotAllocator& FactoryContextImpl::threadLocal() { return server_.threadLocal(); }
 Admin& FactoryContextImpl::admin() { return server_.admin(); }
 TimeSource& FactoryContextImpl::timeSource() { return server_.timeSource(); }
+ProtobufMessage::ValidationContext& FactoryContextImpl::messageValidationContext() {
+  return server_.messageValidationContext();
+}
 ProtobufMessage::ValidationVisitor& FactoryContextImpl::messageValidationVisitor() {
   return server_.messageValidationContext().staticValidationVisitor();
 }
diff --git a/source/server/filter_chain_manager_impl.h b/source/server/filter_chain_manager_impl.h
index af09a01d0a19d..61870753a3a69 100644
--- a/source/server/filter_chain_manager_impl.h
+++ b/source/server/filter_chain_manager_impl.h
@@ -64,6 +64,7 @@ class FilterChainFactoryContextImpl : public Configuration::FilterChainFactoryCo
   envoy::config::core::v3::TrafficDirection direction() const override;
   TimeSource& timeSource() override;
   ProtobufMessage::ValidationVisitor& messageValidationVisitor() override;
+  ProtobufMessage::ValidationContext& messageValidationContext() override;
   Api::Api& api() override;
   ServerLifecycleNotifier& lifecycleNotifier() override;
   ProcessContextOptRef processContext() override;
@@ -102,6 +103,7 @@ class FactoryContextImpl : public Configuration::FactoryContext {
   ThreadLocal::SlotAllocator& threadLocal() override;
   Admin& admin() override;
   TimeSource& timeSource() override;
+  ProtobufMessage::ValidationContext& messageValidationContext() override;
   ProtobufMessage::ValidationVisitor& messageValidationVisitor() override;
   Api::Api& api() override;
   ServerLifecycleNotifier& lifecycleNotifier() override;
diff --git a/source/server/listener_impl.cc b/source/server/listener_impl.cc
index 781fe8e40c971..239cfa972bcca 100644
--- a/source/server/listener_impl.cc
+++ b/source/server/listener_impl.cc
@@ -65,7 +65,6 @@ ListenSocketFactoryImpl::ListenSocketFactoryImpl(ListenerComponentFactory& facto
   if (socket_ && local_address_->ip() && local_address_->ip()->port() == 0) {
     local_address_ = socket_->localAddress();
   }
-
   ENVOY_LOG(debug, "Set listener {} socket factory local address to {}", listener_name_,
             local_address_->asString());
 }
@@ -126,8 +125,7 @@ Network::SocketSharedPtr ListenSocketFactoryImpl::getListenSocket() {
 ListenerImpl::ListenerImpl(const envoy::config::listener::v3::Listener& config,
                            const std::string& version_info, ListenerManagerImpl& parent,
                            const std::string& name, bool added_via_api, bool workers_started,
-                           uint64_t hash, ProtobufMessage::ValidationVisitor& validation_visitor,
-                           uint32_t concurrency)
+                           uint64_t hash, uint32_t concurrency)
     : parent_(parent), address_(Network::Address::resolveProtoAddress(config.address())),
       filter_chain_manager_(address_, *this),
       global_scope_(parent_.server_.stats().createScope("")),
@@ -139,10 +137,23 @@ ListenerImpl::ListenerImpl(const envoy::config::listener::v3::Listener& config,
       per_connection_buffer_limit_bytes_(
           PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, per_connection_buffer_limit_bytes, 1024 * 1024)),
       listener_tag_(parent_.factory_.nextListenerTag()), name_(name), added_via_api_(added_via_api),
-      workers_started_(workers_started), hash_(hash), validation_visitor_(validation_visitor),
-      dynamic_init_manager_(fmt::format("Listener {}", name)),
-      init_watcher_(std::make_unique<Init::WatcherImpl>(
-          "ListenerImpl", [this] { parent_.onListenerWarmed(*this); })),
+      workers_started_(workers_started), hash_(hash),
+      validation_visitor_(
+          added_via_api_ ? parent_.server_.messageValidationContext().dynamicValidationVisitor()
+                         : parent_.server_.messageValidationContext().staticValidationVisitor()),
+      local_init_watcher_(fmt::format("Listener-local-init-watcher {}", name),
+                          [this] {
+                            if (workers_started_) {
+                              parent_.onListenerWarmed(*this);
+                            } else {
+                              // Notify Server that this listener is
+                              // ready.
+                              listener_init_target_.ready();
+                            }
+                          }),
+      listener_init_target_(fmt::format("Listener-init-target {}", name),
+                            [this]() { dynamic_init_manager_.initialize(local_init_watcher_); }),
+      dynamic_init_manager_(fmt::format("Listener-local-init-manager {}", name)),
       local_drain_manager_(parent.factory_.createDrainManager(config.drain_type())),
       config_(config), version_info_(version_info),
       listener_filters_timeout_(
@@ -226,8 +237,8 @@ ListenerImpl::ListenerImpl(const envoy::config::listener::v3::Listener& config,
       parent_.server_.admin(), parent_.server_.sslContextManager(), *listener_scope_,
       parent_.server_.clusterManager(), parent_.server_.localInfo(), parent_.server_.dispatcher(),
       parent_.server_.random(), parent_.server_.stats(), parent_.server_.singletonManager(),
-      parent_.server_.threadLocal(), validation_visitor, parent_.server_.api());
-  transport_factory_context.setInitManager(initManager());
+      parent_.server_.threadLocal(), validation_visitor_, parent_.server_.api());
+  transport_factory_context.setInitManager(dynamic_init_manager_);
   // The init manager is a little messy. Will refactor when filter chain manager could accept
   // network filter chain update.
   // TODO(lambdai): create builder from filter_chain_manager to obtain the init manager
@@ -304,15 +315,14 @@ ListenerImpl::ListenerImpl(const envoy::config::listener::v3::Listener& config,
     listener_filter_factories_.push_back(
         factory.createFilterFactoryFromProto(Envoy::ProtobufWkt::Empty(), *this));
   }
-}
 
-ListenerImpl::~ListenerImpl() {
-  // The filter factories may have pending initialize actions (like in the case of RDS). Those
-  // actions will fire in the destructor to avoid blocking initial server startup. If we are using
-  // a local init manager we should block the notification from trying to move us from warming to
-  // active. This is done here explicitly by resetting the watcher and then clearing the factory
-  // vector for clarity.
-  init_watcher_.reset();
+  if (!workers_started_) {
+    // Initialize dynamic_init_manager_ from Server's init manager if it's not initialized.
+    // NOTE: listener_init_target_ should be added to parent's initManager at the end of the
+    // listener constructor so that this listener's children entities could register their targets
+    // with their parent's initManager.
+    parent_.server_.initManager().add(listener_init_target_);
+  }
 }
 
 AccessLog::AccessLogManager& ListenerImpl::accessLogManager() {
@@ -345,6 +355,9 @@ envoy::config::core::v3::TrafficDirection ListenerImpl::direction() const {
 TimeSource& ListenerImpl::timeSource() { return api().timeSource(); }
 
 const Network::ListenerConfig& ListenerImpl::listenerConfig() const { return *this; }
+ProtobufMessage::ValidationContext& ListenerImpl::messageValidationContext() {
+  return getServerFactoryContext().messageValidationContext();
+}
 ProtobufMessage::ValidationVisitor& ListenerImpl::messageValidationVisitor() {
   return validation_visitor_;
 }
@@ -395,19 +408,23 @@ void ListenerImpl::initialize() {
   // per listener init manager. See ~ListenerImpl() for why we gate the onListenerWarmed() call
   // by resetting the watcher.
   if (workers_started_) {
-    dynamic_init_manager_.initialize(*init_watcher_);
+    ENVOY_LOG_MISC(debug, "Initialize listener {} local-init-manager.", name_);
+    // If workers_started_ is true, dynamic_init_manager_ should be initialized by listener manager
+    // directly.
+    dynamic_init_manager_.initialize(local_init_watcher_);
   }
 }
 
-Init::Manager& ListenerImpl::initManager() {
-  // See initialize() for why we choose different init managers to return.
-  if (workers_started_) {
-    return dynamic_init_manager_;
-  } else {
-    return parent_.server_.initManager();
+ListenerImpl::~ListenerImpl() {
+  if (!workers_started_) {
+    // We need to remove the listener_init_target_ handle from parent's initManager(), to unblock
+    // parent's initManager to get ready().
+    listener_init_target_.ready();
   }
 }
 
+Init::Manager& ListenerImpl::initManager() { return dynamic_init_manager_; }
+
 void ListenerImpl::setSocketFactory(const Network::ListenSocketFactorySharedPtr& socket_factory) {
   ASSERT(!socket_factory_);
   socket_factory_ = socket_factory;
diff --git a/source/server/listener_impl.h b/source/server/listener_impl.h
index a45bf23a74cf5..548b74794b7f0 100644
--- a/source/server/listener_impl.h
+++ b/source/server/listener_impl.h
@@ -12,6 +12,7 @@
 
 #include "common/common/logger.h"
 #include "common/init/manager_impl.h"
+#include "common/init/target_impl.h"
 
 #include "server/filter_chain_manager_impl.h"
 
@@ -95,8 +96,7 @@ class ListenerImpl : public Network::ListenerConfig,
    */
   ListenerImpl(const envoy::config::listener::v3::Listener& config, const std::string& version_info,
                ListenerManagerImpl& parent, const std::string& name, bool added_via_api,
-               bool workers_started, uint64_t hash,
-               ProtobufMessage::ValidationVisitor& validation_visitor, uint32_t concurrency);
+               bool workers_started, uint64_t hash, uint32_t concurrency);
   ~ListenerImpl() override;
 
   /**
@@ -173,6 +173,7 @@ class ListenerImpl : public Network::ListenerConfig,
   envoy::config::core::v3::TrafficDirection direction() const override;
   TimeSource& timeSource() override;
   const Network::ListenerConfig& listenerConfig() const override;
+  ProtobufMessage::ValidationContext& messageValidationContext() override;
   ProtobufMessage::ValidationVisitor& messageValidationVisitor() override;
   Api::Api& api() override;
   ServerLifecycleNotifier& lifecycleNotifier() override;
@@ -225,13 +226,15 @@ class ListenerImpl : public Network::ListenerConfig,
   const uint64_t hash_;
   ProtobufMessage::ValidationVisitor& validation_visitor_;
 
+  // This init watcher, if workers_started_ is false, notifies the "parent" listener manager when
+  // listener initialization is complete.
+  Init::WatcherImpl local_init_watcher_;
+  // A target is added to Server's InitManager if workers_started_ is false.
+  Init::TargetImpl listener_init_target_;
   // This init manager is populated with targets from the filter chain factories, namely
   // RdsRouteConfigSubscription::init_target_, so the listener can wait for route configs.
   Init::ManagerImpl dynamic_init_manager_;
 
-  // This init watcher, if available, notifies the "parent" listener manager when listener
-  // initialization is complete. It may be reset to cancel interest.
-  std::unique_ptr<Init::WatcherImpl> init_watcher_;
   std::vector<Network::ListenerFilterFactoryCb> listener_filter_factories_;
   std::vector<Network::UdpListenerFilterFactoryCb> udp_listener_filter_factories_;
   DrainManagerPtr local_drain_manager_;
diff --git a/source/server/listener_manager_impl.cc b/source/server/listener_manager_impl.cc
index c9e289291f1a9..5099bc1d7c803 100644
--- a/source/server/listener_manager_impl.cc
+++ b/source/server/listener_manager_impl.cc
@@ -389,11 +389,9 @@ bool ListenerManagerImpl::addOrUpdateListenerInternal(
     return false;
   }
 
-  ListenerImplPtr new_listener(
-      new ListenerImpl(config, version_info, *this, name, added_via_api, workers_started_, hash,
-                       added_via_api ? server_.messageValidationContext().dynamicValidationVisitor()
-                                     : server_.messageValidationContext().staticValidationVisitor(),
-                       server_.options().concurrency()));
+  ListenerImplPtr new_listener(new ListenerImpl(config, version_info, *this, name, added_via_api,
+                                                workers_started_, hash,
+                                                server_.options().concurrency()));
   ListenerImpl& new_listener_ref = *new_listener;
 
   // We mandate that a listener with the same name must have the same configured address. This
diff --git a/source/server/server.h b/source/server/server.h
index ee8155c454e84..e05f69525d988 100644
--- a/source/server/server.h
+++ b/source/server/server.h
@@ -160,6 +160,9 @@ class ServerFactoryContextImpl : public Configuration::ServerFactoryContext,
   Upstream::ClusterManager& clusterManager() override { return server_.clusterManager(); }
   Event::Dispatcher& dispatcher() override { return server_.dispatcher(); }
   const LocalInfo::LocalInfo& localInfo() const override { return server_.localInfo(); }
+  ProtobufMessage::ValidationContext& messageValidationContext() override {
+    return server_.messageValidationContext();
+  }
   Envoy::Runtime::RandomGenerator& random() override { return server_.random(); }
   Envoy::Runtime::Loader& runtime() override { return server_.runtime(); }
   Stats::Scope& scope() override { return *server_scope_; }
diff --git a/test/common/config/config_provider_impl_test.cc b/test/common/config/config_provider_impl_test.cc
index ecdd17d126ac5..63cdc00669d3b 100644
--- a/test/common/config/config_provider_impl_test.cc
+++ b/test/common/config/config_provider_impl_test.cc
@@ -41,7 +41,7 @@ class DummyConfig : public Envoy::Config::ConfigProvider::Config {
 class StaticDummyConfigProvider : public ImmutableConfigProviderBase {
 public:
   StaticDummyConfigProvider(const test::common::config::DummyConfig& config_proto,
-                            Server::Configuration::FactoryContext& factory_context,
+                            Server::Configuration::ServerFactoryContext& factory_context,
                             DummyConfigProviderManager& config_provider_manager);
 
   ~StaticDummyConfigProvider() override = default;
@@ -64,7 +64,7 @@ class DummyConfigSubscription : public ConfigSubscriptionInstance,
                                 Envoy::Config::SubscriptionCallbacks {
 public:
   DummyConfigSubscription(const uint64_t manager_identifier,
-                          Server::Configuration::FactoryContext& factory_context,
+                          Server::Configuration::ServerFactoryContext& factory_context,
                           DummyConfigProviderManager& config_provider_manager);
   ~DummyConfigSubscription() override = default;
 
@@ -173,11 +173,11 @@ class DummyConfigProviderManager : public ConfigProviderManagerImplBase {
   // Envoy::Config::ConfigProviderManager
   ConfigProviderPtr
   createXdsConfigProvider(const Protobuf::Message& config_source_proto,
-                          Server::Configuration::FactoryContext& factory_context,
-                          const std::string&,
+                          Server::Configuration::ServerFactoryContext& factory_context,
+                          Init::Manager& init_manager, const std::string&,
                           const Envoy::Config::ConfigProviderManager::OptionalArg&) override {
     DummyConfigSubscriptionSharedPtr subscription = getSubscription<DummyConfigSubscription>(
-        config_source_proto, factory_context.initManager(),
+        config_source_proto, init_manager,
         [&factory_context](const uint64_t manager_identifier,
                            ConfigProviderManagerImplBase& config_provider_manager)
             -> ConfigSubscriptionCommonBaseSharedPtr {
@@ -192,7 +192,7 @@ class DummyConfigProviderManager : public ConfigProviderManagerImplBase {
   // Envoy::Config::ConfigProviderManager
   ConfigProviderPtr
   createStaticConfigProvider(const Protobuf::Message& config_proto,
-                             Server::Configuration::FactoryContext& factory_context,
+                             Server::Configuration::ServerFactoryContext& factory_context,
                              const Envoy::Config::ConfigProviderManager::OptionalArg&) override {
     return std::make_unique<StaticDummyConfigProvider>(
         dynamic_cast<const test::common::config::DummyConfig&>(config_proto), factory_context,
@@ -200,14 +200,15 @@ class DummyConfigProviderManager : public ConfigProviderManagerImplBase {
   }
   ConfigProviderPtr
   createStaticConfigProvider(std::vector<std::unique_ptr<const Protobuf::Message>>&&,
-                             Server::Configuration::FactoryContext&, const OptionalArg&) override {
+                             Server::Configuration::ServerFactoryContext&,
+                             const OptionalArg&) override {
     ASSERT(false, "this provider does not expect multiple config protos");
     return nullptr;
   }
 };
 
 DummyConfigSubscription::DummyConfigSubscription(
-    const uint64_t manager_identifier, Server::Configuration::FactoryContext& factory_context,
+    const uint64_t manager_identifier, Server::Configuration::ServerFactoryContext& factory_context,
     DummyConfigProviderManager& config_provider_manager)
     : ConfigSubscriptionInstance("DummyDS", manager_identifier, config_provider_manager,
                                  factory_context) {
@@ -217,7 +218,7 @@ DummyConfigSubscription::DummyConfigSubscription(
 
 StaticDummyConfigProvider::StaticDummyConfigProvider(
     const test::common::config::DummyConfig& config_proto,
-    Server::Configuration::FactoryContext& factory_context,
+    Server::Configuration::ServerFactoryContext& factory_context,
     DummyConfigProviderManager& config_provider_manager)
     : ImmutableConfigProviderBase(factory_context, config_provider_manager,
                                   ConfigProviderInstanceType::Static, ApiType::Full),
@@ -226,15 +227,17 @@ StaticDummyConfigProvider::StaticDummyConfigProvider(
 class ConfigProviderImplTest : public testing::Test {
 public:
   void initialize() {
-    EXPECT_CALL(factory_context_.admin_.config_tracker_, add_("dummy", _));
-    provider_manager_ = std::make_unique<DummyConfigProviderManager>(factory_context_.admin_);
+    EXPECT_CALL(server_factory_context_.admin_.config_tracker_, add_("dummy", _));
+    provider_manager_ =
+        std::make_unique<DummyConfigProviderManager>(server_factory_context_.admin_);
   }
 
   Event::SimulatedTimeSystem& timeSystem() { return time_system_; }
 
 protected:
   Event::SimulatedTimeSystem time_system_;
-  NiceMock<Server::Configuration::MockFactoryContext> factory_context_;
+  NiceMock<Server::Configuration::MockServerFactoryContext> server_factory_context_;
+  NiceMock<Init::MockManager> init_manager_;
   std::unique_ptr<DummyConfigProviderManager> provider_manager_;
 };
 
@@ -250,12 +253,12 @@ test::common::config::DummyConfig parseDummyConfigFromYaml(const std::string& ya
 TEST_F(ConfigProviderImplTest, SharedOwnership) {
   initialize();
   Init::ExpectableWatcherImpl watcher;
-  factory_context_.init_manager_.initialize(watcher);
+  init_manager_.initialize(watcher);
 
   envoy::config::core::v3::ApiConfigSource config_source_proto;
   config_source_proto.set_api_type(envoy::config::core::v3::ApiConfigSource::GRPC);
   ConfigProviderPtr provider1 = provider_manager_->createXdsConfigProvider(
-      config_source_proto, factory_context_, "dummy_prefix",
+      config_source_proto, server_factory_context_, init_manager_, "dummy_prefix",
       ConfigProviderManager::NullOptionalArg());
 
   // No config protos have been received via the subscription yet.
@@ -271,7 +274,7 @@ TEST_F(ConfigProviderImplTest, SharedOwnership) {
   // Check that a newly created provider with the same config source will share
   // the subscription, config proto and resulting ConfigProvider::Config.
   ConfigProviderPtr provider2 = provider_manager_->createXdsConfigProvider(
-      config_source_proto, factory_context_, "dummy_prefix",
+      config_source_proto, server_factory_context_, init_manager_, "dummy_prefix",
       ConfigProviderManager::NullOptionalArg());
 
   EXPECT_TRUE(provider2->configProtoInfo<test::common::config::DummyConfig>().has_value());
@@ -285,7 +288,7 @@ TEST_F(ConfigProviderImplTest, SharedOwnership) {
   // Change the config source and verify that a new subscription is used.
   config_source_proto.set_api_type(envoy::config::core::v3::ApiConfigSource::REST);
   ConfigProviderPtr provider3 = provider_manager_->createXdsConfigProvider(
-      config_source_proto, factory_context_, "dummy_prefix",
+      config_source_proto, server_factory_context_, init_manager_, "dummy_prefix",
       ConfigProviderManager::NullOptionalArg());
 
   EXPECT_NE(&dynamic_cast<DummyDynamicConfigProvider&>(*provider1).subscription(),
@@ -330,11 +333,11 @@ class DummyConfigProviderManagerMockConfigProvider : public DummyConfigProviderM
 
   ConfigProviderPtr
   createXdsConfigProvider(const Protobuf::Message& config_source_proto,
-                          Server::Configuration::FactoryContext& factory_context,
-                          const std::string&,
+                          Server::Configuration::ServerFactoryContext& factory_context,
+                          Init::Manager& init_manager, const std::string&,
                           const Envoy::Config::ConfigProviderManager::OptionalArg&) override {
     DummyConfigSubscriptionSharedPtr subscription = getSubscription<DummyConfigSubscription>(
-        config_source_proto, factory_context.initManager(),
+        config_source_proto, init_manager,
         [&factory_context](const uint64_t manager_identifier,
                            ConfigProviderManagerImplBase& config_provider_manager)
             -> ConfigSubscriptionCommonBaseSharedPtr {
@@ -350,12 +353,12 @@ class DummyConfigProviderManagerMockConfigProvider : public DummyConfigProviderM
 TEST_F(ConfigProviderImplTest, DuplicateConfigProto) {
   InSequence sequence;
   // This provider manager returns a DummyDynamicConfigProvider.
-  auto provider_manager =
-      std::make_unique<DummyConfigProviderManagerMockConfigProvider>(factory_context_.admin_);
+  auto provider_manager = std::make_unique<DummyConfigProviderManagerMockConfigProvider>(
+      server_factory_context_.admin_);
   envoy::config::core::v3::ApiConfigSource config_source_proto;
   config_source_proto.set_api_type(envoy::config::core::v3::ApiConfigSource::GRPC);
   ConfigProviderPtr provider = provider_manager->createXdsConfigProvider(
-      config_source_proto, factory_context_, "dummy_prefix",
+      config_source_proto, server_factory_context_, init_manager_, "dummy_prefix",
       ConfigProviderManager::NullOptionalArg());
   auto* typed_provider = static_cast<DummyDynamicConfigProvider*>(provider.get());
   auto& subscription = static_cast<DummyConfigSubscription&>(typed_provider->subscription());
@@ -377,7 +380,7 @@ TEST_F(ConfigProviderImplTest, DuplicateConfigProto) {
 // An empty config provider tests on base class' constructor.
 class InlineDummyConfigProvider : public ImmutableConfigProviderBase {
 public:
-  InlineDummyConfigProvider(Server::Configuration::FactoryContext& factory_context,
+  InlineDummyConfigProvider(Server::Configuration::ServerFactoryContext& factory_context,
                             DummyConfigProviderManager& config_provider_manager,
                             ConfigProviderInstanceType instance_type)
       : ImmutableConfigProviderBase(factory_context, config_provider_manager, instance_type,
@@ -392,11 +395,11 @@ class ConfigProviderImplDeathTest : public ConfigProviderImplTest {};
 TEST_F(ConfigProviderImplDeathTest, AssertionFailureOnIncorrectInstanceType) {
   initialize();
 
-  InlineDummyConfigProvider foo(factory_context_, *provider_manager_,
+  InlineDummyConfigProvider foo(server_factory_context_, *provider_manager_,
                                 ConfigProviderInstanceType::Inline);
-  InlineDummyConfigProvider bar(factory_context_, *provider_manager_,
+  InlineDummyConfigProvider bar(server_factory_context_, *provider_manager_,
                                 ConfigProviderInstanceType::Static);
-  EXPECT_DEBUG_DEATH(InlineDummyConfigProvider(factory_context_, *provider_manager_,
+  EXPECT_DEBUG_DEATH(InlineDummyConfigProvider(server_factory_context_, *provider_manager_,
                                                ConfigProviderInstanceType::Xds),
                      "");
 }
@@ -407,7 +410,8 @@ TEST_F(ConfigProviderImplDeathTest, AssertionFailureOnIncorrectInstanceType) {
 TEST_F(ConfigProviderImplTest, ConfigDump) {
   initialize();
   // Empty dump first.
-  auto message_ptr = factory_context_.admin_.config_tracker_.config_tracker_callbacks_["dummy"]();
+  auto message_ptr =
+      server_factory_context_.admin_.config_tracker_.config_tracker_callbacks_["dummy"]();
   const auto& dummy_config_dump =
       static_cast<const test::common::config::DummyConfigsDump&>(*message_ptr);
 
@@ -424,9 +428,9 @@ TEST_F(ConfigProviderImplTest, ConfigDump) {
   timeSystem().setSystemTime(std::chrono::milliseconds(1234567891234));
 
   ConfigProviderPtr static_config = provider_manager_->createStaticConfigProvider(
-      parseDummyConfigFromYaml(config_yaml), factory_context_,
+      parseDummyConfigFromYaml(config_yaml), server_factory_context_,
       ConfigProviderManager::NullOptionalArg());
-  message_ptr = factory_context_.admin_.config_tracker_.config_tracker_callbacks_["dummy"]();
+  message_ptr = server_factory_context_.admin_.config_tracker_.config_tracker_callbacks_["dummy"]();
   const auto& dummy_config_dump2 =
       static_cast<const test::common::config::DummyConfigsDump&>(*message_ptr);
   TestUtility::loadFromYaml(R"EOF(
@@ -441,7 +445,7 @@ TEST_F(ConfigProviderImplTest, ConfigDump) {
   envoy::config::core::v3::ApiConfigSource config_source_proto;
   config_source_proto.set_api_type(envoy::config::core::v3::ApiConfigSource::GRPC);
   ConfigProviderPtr dynamic_provider = provider_manager_->createXdsConfigProvider(
-      config_source_proto, factory_context_, "dummy_prefix",
+      config_source_proto, server_factory_context_, init_manager_, "dummy_prefix",
       ConfigProviderManager::NullOptionalArg());
 
   // Static + dynamic config dump.
@@ -453,7 +457,7 @@ TEST_F(ConfigProviderImplTest, ConfigDump) {
       dynamic_cast<DummyDynamicConfigProvider&>(*dynamic_provider).subscription();
   subscription.onConfigUpdate(untyped_dummy_configs, "v1");
 
-  message_ptr = factory_context_.admin_.config_tracker_.config_tracker_callbacks_["dummy"]();
+  message_ptr = server_factory_context_.admin_.config_tracker_.config_tracker_callbacks_["dummy"]();
   const auto& dummy_config_dump3 =
       static_cast<const test::common::config::DummyConfigsDump&>(*message_ptr);
   TestUtility::loadFromYaml(R"EOF(
@@ -469,9 +473,9 @@ TEST_F(ConfigProviderImplTest, ConfigDump) {
   EXPECT_EQ(expected_config_dump.DebugString(), dummy_config_dump3.DebugString());
 
   ConfigProviderPtr static_config2 = provider_manager_->createStaticConfigProvider(
-      parseDummyConfigFromYaml("a: another static dummy config"), factory_context_,
+      parseDummyConfigFromYaml("a: another static dummy config"), server_factory_context_,
       ConfigProviderManager::NullOptionalArg());
-  message_ptr = factory_context_.admin_.config_tracker_.config_tracker_callbacks_["dummy"]();
+  message_ptr = server_factory_context_.admin_.config_tracker_.config_tracker_callbacks_["dummy"]();
   const auto& dummy_config_dump4 =
       static_cast<const test::common::config::DummyConfigsDump&>(*message_ptr);
   TestUtility::loadFromYaml(R"EOF(
@@ -494,14 +498,14 @@ TEST_F(ConfigProviderImplTest, ConfigDump) {
 // subscriptions.
 TEST_F(ConfigProviderImplTest, LocalInfoNotDefined) {
   initialize();
-  factory_context_.local_info_.node_.set_cluster("");
-  factory_context_.local_info_.node_.set_id("");
+  server_factory_context_.local_info_.node_.set_cluster("");
+  server_factory_context_.local_info_.node_.set_id("");
 
   envoy::config::core::v3::ApiConfigSource config_source_proto;
   config_source_proto.set_api_type(envoy::config::core::v3::ApiConfigSource::GRPC);
   EXPECT_THROW_WITH_MESSAGE(
-      provider_manager_->createXdsConfigProvider(config_source_proto, factory_context_,
-                                                 "dummy_prefix",
+      provider_manager_->createXdsConfigProvider(config_source_proto, server_factory_context_,
+                                                 init_manager_, "dummy_prefix",
                                                  ConfigProviderManager::NullOptionalArg()),
       EnvoyException,
       "DummyDS: node 'id' and 'cluster' are required. Set it either in 'node' config or "
@@ -516,7 +520,7 @@ class DeltaDummyConfigSubscription : public DeltaConfigSubscriptionInstance,
   using ProtoMap = std::map<std::string, test::common::config::DummyConfig>;
 
   DeltaDummyConfigSubscription(const uint64_t manager_identifier,
-                               Server::Configuration::FactoryContext& factory_context,
+                               Server::Configuration::ServerFactoryContext& factory_context,
                                DeltaDummyConfigProviderManager& config_provider_manager);
 
   // Envoy::Config::ConfigSubscriptionCommonBase
@@ -628,13 +632,13 @@ class DeltaDummyConfigProviderManager : public ConfigProviderManagerImplBase {
   // Envoy::Config::ConfigProviderManager
   ConfigProviderPtr
   createXdsConfigProvider(const Protobuf::Message& config_source_proto,
-                          Server::Configuration::FactoryContext& factory_context,
-                          const std::string&,
+                          Server::Configuration::ServerFactoryContext& factory_context,
+                          Init::Manager& init_manager, const std::string&,
                           const Envoy::Config::ConfigProviderManager::OptionalArg&) override {
     DeltaDummyConfigSubscriptionSharedPtr subscription =
 
         getSubscription<DeltaDummyConfigSubscription>(
-            config_source_proto, factory_context.initManager(),
+            config_source_proto, init_manager,
             [&factory_context](const uint64_t manager_identifier,
                                ConfigProviderManagerImplBase& config_provider_manager)
                 -> ConfigSubscriptionCommonBaseSharedPtr {
@@ -648,7 +652,7 @@ class DeltaDummyConfigProviderManager : public ConfigProviderManagerImplBase {
 };
 
 DeltaDummyConfigSubscription::DeltaDummyConfigSubscription(
-    const uint64_t manager_identifier, Server::Configuration::FactoryContext& factory_context,
+    const uint64_t manager_identifier, Server::Configuration::ServerFactoryContext& factory_context,
     DeltaDummyConfigProviderManager& config_provider_manager)
     : DeltaConfigSubscriptionInstance("Dummy", manager_identifier, config_provider_manager,
                                       factory_context) {
@@ -659,15 +663,17 @@ DeltaDummyConfigSubscription::DeltaDummyConfigSubscription(
 class DeltaConfigProviderImplTest : public testing::Test {
 public:
   DeltaConfigProviderImplTest() {
-    EXPECT_CALL(factory_context_.admin_.config_tracker_, add_("dummy", _));
-    provider_manager_ = std::make_unique<DeltaDummyConfigProviderManager>(factory_context_.admin_);
+    EXPECT_CALL(server_factory_context_.admin_.config_tracker_, add_("dummy", _));
+    provider_manager_ =
+        std::make_unique<DeltaDummyConfigProviderManager>(server_factory_context_.admin_);
   }
 
   Event::SimulatedTimeSystem& timeSystem() { return time_system_; }
 
 protected:
   Event::SimulatedTimeSystem time_system_;
-  NiceMock<Server::Configuration::MockFactoryContext> factory_context_;
+  NiceMock<Server::Configuration::MockServerFactoryContext> server_factory_context_;
+  NiceMock<Init::MockManager> init_manager_;
   std::unique_ptr<DeltaDummyConfigProviderManager> provider_manager_;
 };
 
@@ -677,7 +683,7 @@ TEST_F(DeltaConfigProviderImplTest, MultipleDeltaSubscriptions) {
   envoy::config::core::v3::ApiConfigSource config_source_proto;
   config_source_proto.set_api_type(envoy::config::core::v3::ApiConfigSource::GRPC);
   ConfigProviderPtr provider1 = provider_manager_->createXdsConfigProvider(
-      config_source_proto, factory_context_, "dummy_prefix",
+      config_source_proto, server_factory_context_, init_manager_, "dummy_prefix",
       ConfigProviderManager::NullOptionalArg());
 
   // No config protos have been received via the subscription yet.
@@ -692,7 +698,7 @@ TEST_F(DeltaConfigProviderImplTest, MultipleDeltaSubscriptions) {
   subscription.onConfigUpdate(untyped_dummy_configs, "1");
 
   ConfigProviderPtr provider2 = provider_manager_->createXdsConfigProvider(
-      config_source_proto, factory_context_, "dummy_prefix",
+      config_source_proto, server_factory_context_, init_manager_, "dummy_prefix",
       ConfigProviderManager::NullOptionalArg());
 
   // Providers, config implementations (i.e., the DummyConfig) and config protos are
@@ -725,7 +731,7 @@ TEST_F(DeltaConfigProviderImplTest, DeltaSubscriptionFailure) {
   envoy::config::core::v3::ApiConfigSource config_source_proto;
   config_source_proto.set_api_type(envoy::config::core::v3::ApiConfigSource::GRPC);
   ConfigProviderPtr provider = provider_manager_->createXdsConfigProvider(
-      config_source_proto, factory_context_, "dummy_prefix",
+      config_source_proto, server_factory_context_, init_manager_, "dummy_prefix",
       ConfigProviderManager::NullOptionalArg());
   DeltaDummyConfigSubscription& subscription =
       dynamic_cast<DeltaDummyDynamicConfigProvider&>(*provider).subscription();
diff --git a/test/common/router/BUILD b/test/common/router/BUILD
index f8d7b78f024ec..f09535bb5189f 100644
--- a/test/common/router/BUILD
+++ b/test/common/router/BUILD
@@ -69,6 +69,7 @@ envoy_cc_test(
         "//source/common/router:rds_lib",
         "//source/server/http:admin_lib",
         "//test/mocks/local_info:local_info_mocks",
+        "//test/mocks/protobuf:protobuf_mocks",
         "//test/mocks/server:server_mocks",
         "//test/mocks/thread_local:thread_local_mocks",
         "//test/mocks/upstream:upstream_mocks",
@@ -113,6 +114,7 @@ envoy_cc_test(
         "//source/server/http:admin_lib",
         "//test/mocks/config:config_mocks",
         "//test/mocks/init:init_mocks",
+        "//test/mocks/protobuf:protobuf_mocks",
         "//test/mocks/router:router_mocks",
         "//test/mocks/server:server_mocks",
         "//test/test_common:simulated_time_system_lib",
diff --git a/test/common/router/rds_impl_test.cc b/test/common/router/rds_impl_test.cc
index d8bbeecb79f37..13f35284ea391 100644
--- a/test/common/router/rds_impl_test.cc
+++ b/test/common/router/rds_impl_test.cc
@@ -17,6 +17,7 @@
 
 #include "test/mocks/init/mocks.h"
 #include "test/mocks/local_info/mocks.h"
+#include "test/mocks/protobuf/mocks.h"
 #include "test/mocks/server/mocks.h"
 #include "test/mocks/thread_local/mocks.h"
 #include "test/mocks/upstream/mocks.h"
@@ -50,12 +51,12 @@ class RdsTestBase : public testing::Test {
 public:
   RdsTestBase() {
     // For server_factory_context
-    ON_CALL(mock_factory_context_, getServerFactoryContext())
-        .WillByDefault(ReturnRef(server_factory_context_));
     ON_CALL(server_factory_context_, scope()).WillByDefault(ReturnRef(scope_));
-    ON_CALL(mock_factory_context_, scope()).WillByDefault(ReturnRef(scope_));
+    ON_CALL(server_factory_context_, messageValidationContext())
+        .WillByDefault(ReturnRef(validation_context_));
+    EXPECT_CALL(validation_context_, dynamicValidationVisitor())
+        .WillRepeatedly(ReturnRef(validation_visitor_));
 
-    ON_CALL(mock_factory_context_, initManager()).WillByDefault(ReturnRef(outer_init_manager_));
     ON_CALL(outer_init_manager_, add(_)).WillByDefault(Invoke([this](const Init::Target& target) {
       init_target_handle_ = target.createHandle("test");
     }));
@@ -67,7 +68,8 @@ class RdsTestBase : public testing::Test {
   Event::SimulatedTimeSystem& timeSystem() { return time_system_; }
 
   Event::SimulatedTimeSystem time_system_;
-  NiceMock<Server::Configuration::MockFactoryContext> mock_factory_context_;
+  NiceMock<ProtobufMessage::MockValidationContext> validation_context_;
+  NiceMock<ProtobufMessage::MockValidationVisitor> validation_visitor_;
   NiceMock<Init::MockManager> outer_init_manager_;
   NiceMock<Server::Configuration::MockServerFactoryContext> server_factory_context_;
   Init::ExpectableWatcherImpl init_watcher_;
@@ -103,9 +105,9 @@ stat_prefix: foo
     )EOF";
 
     EXPECT_CALL(outer_init_manager_, add(_));
-    rds_ = RouteConfigProviderUtil::create(parseHttpConnectionManagerFromYaml(config_yaml),
-                                           mock_factory_context_, "foo.",
-                                           *route_config_provider_manager_);
+    rds_ = RouteConfigProviderUtil::create(
+        parseHttpConnectionManagerFromYaml(config_yaml), server_factory_context_,
+        validation_visitor_, outer_init_manager_, "foo.", *route_config_provider_manager_);
     rds_callbacks_ = server_factory_context_.cluster_manager_.subscription_factory_.callbacks_;
     EXPECT_CALL(*server_factory_context_.cluster_manager_.subscription_factory_.subscription_,
                 start(_));
@@ -135,7 +137,8 @@ stat_prefix: foo
     )EOF";
 
   EXPECT_THROW(RouteConfigProviderUtil::create(parseHttpConnectionManagerFromYaml(config_yaml),
-                                               mock_factory_context_, "foo.",
+                                               server_factory_context_, validation_visitor_,
+                                               outer_init_manager_, "foo.",
                                                *route_config_provider_manager_),
                EnvoyException);
 }
@@ -220,7 +223,7 @@ TEST_F(RdsImplTest, Basic) {
       TestUtility::parseYaml<envoy::service::discovery::v3::DiscoveryResponse>(response2_json);
 
   // Make sure we don't lookup/verify clusters.
-  EXPECT_CALL(mock_factory_context_.cluster_manager_, get(Eq("bar"))).Times(0);
+  EXPECT_CALL(server_factory_context_.cluster_manager_, get(Eq("bar"))).Times(0);
   rds_callbacks_->onConfigUpdate(response2.resources(), response2.version_info());
   EXPECT_EQ("foo", route(Http::TestHeaderMapImpl{{":authority", "foo"}, {":path", "/foo"}})
                        ->routeEntry()
@@ -297,21 +300,30 @@ TEST_F(RdsRouteConfigSubscriptionTest, CreatesNoopInitManager) {
         envoy_grpc:
           cluster_name: xds_cluster
 )EOF";
-  EXPECT_CALL(outer_init_manager_, state()).WillOnce(Return(Init::Manager::State::Initialized));
   const auto rds =
       TestUtility::parseYaml<envoy::extensions::filters::network::http_connection_manager::v3::Rds>(
           rds_config);
   const auto route_config_provider = route_config_provider_manager_->createRdsRouteConfigProvider(
-      rds, mock_factory_context_, "stat_prefix", outer_init_manager_);
+      rds, server_factory_context_, "stat_prefix", outer_init_manager_);
   RdsRouteConfigSubscription& subscription =
       (dynamic_cast<RdsRouteConfigProviderImpl*>(route_config_provider.get()))->subscription();
-
+  init_watcher_.expectReady().Times(1); // The parent_init_target_ will call once.
+  outer_init_manager_.initialize(init_watcher_);
   std::unique_ptr<Init::ManagerImpl> noop_init_manager;
   std::unique_ptr<Cleanup> init_vhds;
   subscription.maybeCreateInitManager("version_info", noop_init_manager, init_vhds);
-
-  EXPECT_TRUE(init_vhds);
-  EXPECT_TRUE(noop_init_manager);
+  // local_init_manager_ is not ready yet as the local_init_target_ is not ready.
+  EXPECT_EQ(init_vhds, nullptr);
+  EXPECT_EQ(noop_init_manager, nullptr);
+  // Now mark local_init_target_ ready by forcing an update failure.
+  auto* rds_callbacks_ = server_factory_context_.cluster_manager_.subscription_factory_.callbacks_;
+  EnvoyException e("test");
+  rds_callbacks_->onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::UpdateRejected,
+                                       &e);
+  // Now noop init manager will be created as local_init_manager_ is initialized.
+  subscription.maybeCreateInitManager("version_info", noop_init_manager, init_vhds);
+  EXPECT_NE(init_vhds, nullptr);
+  EXPECT_NE(noop_init_manager, nullptr);
 }
 
 class RouteConfigProviderManagerImplTest : public RdsTestBase {
@@ -321,7 +333,7 @@ class RouteConfigProviderManagerImplTest : public RdsTestBase {
     rds_.set_route_config_name("foo_route_config");
     rds_.mutable_config_source()->set_path("foo_path");
     provider_ = route_config_provider_manager_->createRdsRouteConfigProvider(
-        rds_, mock_factory_context_, "foo_prefix.", outer_init_manager_);
+        rds_, server_factory_context_, "foo_prefix.", outer_init_manager_);
     rds_callbacks_ = server_factory_context_.cluster_manager_.subscription_factory_.callbacks_;
   }
 
@@ -377,7 +389,8 @@ name: foo
   // Only static route.
   RouteConfigProviderPtr static_config =
       route_config_provider_manager_->createStaticRouteConfigProvider(
-          parseRouteConfigurationFromV2Yaml(config_yaml), mock_factory_context_);
+          parseRouteConfigurationFromV2Yaml(config_yaml), server_factory_context_,
+          validation_visitor_);
   message_ptr =
       server_factory_context_.admin_.config_tracker_.config_tracker_callbacks_["routes"]();
   const auto& route_config_dump2 =
@@ -480,7 +493,7 @@ name: foo_route_config
 
   RouteConfigProviderSharedPtr provider2 =
       route_config_provider_manager_->createRdsRouteConfigProvider(
-          rds_, mock_factory_context_, "foo_prefix", outer_init_manager_);
+          rds_, server_factory_context_, "foo_prefix", outer_init_manager_);
 
   // provider2 should have route config immediately after create
   EXPECT_TRUE(provider2->configInfo().has_value());
@@ -497,7 +510,7 @@ name: foo_route_config
   rds2.mutable_config_source()->set_path("bar_path");
   RouteConfigProviderSharedPtr provider3 =
       route_config_provider_manager_->createRdsRouteConfigProvider(
-          rds2, mock_factory_context_, "foo_prefix", mock_factory_context_.initManager());
+          rds2, server_factory_context_, "foo_prefix", outer_init_manager_);
   EXPECT_NE(provider3, provider_);
   server_factory_context_.cluster_manager_.subscription_factory_.callbacks_->onConfigUpdate(
       route_configs, "provider3");
@@ -529,7 +542,7 @@ TEST_F(RouteConfigProviderManagerImplTest, SameProviderOnTwoInitManager) {
 
   EXPECT_FALSE(provider_->configInfo().has_value());
 
-  NiceMock<Server::Configuration::MockFactoryContext> mock_factory_context2;
+  NiceMock<Server::Configuration::MockServerFactoryContext> mock_factory_context2;
 
   Init::WatcherImpl real_watcher("real", []() {});
   Init::ManagerImpl real_init_manager("real");
diff --git a/test/common/router/scoped_rds_test.cc b/test/common/router/scoped_rds_test.cc
index e8412795e584f..6940ed7ff02bf 100644
--- a/test/common/router/scoped_rds_test.cc
+++ b/test/common/router/scoped_rds_test.cc
@@ -14,9 +14,11 @@
 
 #include "common/config/api_version.h"
 #include "common/config/grpc_mux_impl.h"
+#include "common/protobuf/message_validator_impl.h"
 #include "common/router/scoped_rds.h"
 
 #include "test/mocks/config/mocks.h"
+#include "test/mocks/protobuf/mocks.h"
 #include "test/mocks/router/mocks.h"
 #include "test/mocks/server/mocks.h"
 #include "test/test_common/simulated_time_system.h"
@@ -34,6 +36,7 @@ using testing::Invoke;
 using testing::IsNull;
 using testing::NiceMock;
 using testing::Return;
+using testing::ReturnRef;
 
 namespace Envoy {
 namespace Router {
@@ -64,20 +67,21 @@ parseHttpConnectionManagerFromYaml(const std::string& config_yaml) {
 class ScopedRoutesTestBase : public testing::Test {
 protected:
   ScopedRoutesTestBase() {
-    ON_CALL(factory_context_, initManager()).WillByDefault(ReturnRef(context_init_manager_));
-    ON_CALL(factory_context_, getServerFactoryContext())
-        .WillByDefault(ReturnRef(server_factory_context_));
+    ON_CALL(server_factory_context_, messageValidationContext())
+        .WillByDefault(ReturnRef(validation_context_));
+    EXPECT_CALL(validation_context_, dynamicValidationVisitor())
+        .WillRepeatedly(ReturnRef(ProtobufMessage::getStrictValidationVisitor()));
 
-    EXPECT_CALL(factory_context_.admin_.config_tracker_, add_("routes", _));
+    EXPECT_CALL(server_factory_context_.admin_.config_tracker_, add_("routes", _));
     route_config_provider_manager_ =
-        std::make_unique<RouteConfigProviderManagerImpl>(factory_context_.admin_);
+        std::make_unique<RouteConfigProviderManagerImpl>(server_factory_context_.admin_);
 
-    EXPECT_CALL(factory_context_.admin_.config_tracker_, add_("route_scopes", _));
+    EXPECT_CALL(server_factory_context_.admin_.config_tracker_, add_("route_scopes", _));
     config_provider_manager_ = std::make_unique<ScopedRoutesConfigProviderManager>(
-        factory_context_.admin_, *route_config_provider_manager_);
+        server_factory_context_.admin_, *route_config_provider_manager_);
   }
 
-  ~ScopedRoutesTestBase() override { factory_context_.thread_local_.shutdownThread(); }
+  ~ScopedRoutesTestBase() override { server_factory_context_.thread_local_.shutdownThread(); }
 
   // The delta style API helper.
   Protobuf::RepeatedPtrField<envoy::service::discovery::v3::Resource>
@@ -98,8 +102,7 @@ class ScopedRoutesTestBase : public testing::Test {
   Event::SimulatedTimeSystem& timeSystem() { return time_system_; }
 
   NiceMock<Init::MockManager> context_init_manager_;
-  // factory_context_ is used by srds
-  NiceMock<Server::Configuration::MockFactoryContext> factory_context_;
+  NiceMock<ProtobufMessage::MockValidationContext> validation_context_;
   // server_factory_context_ is used by rds
   NiceMock<Server::Configuration::MockServerFactoryContext> server_factory_context_;
   std::unique_ptr<RouteConfigProviderManager> route_config_provider_manager_;
@@ -111,17 +114,17 @@ class ScopedRoutesTestBase : public testing::Test {
 class ScopedRdsTest : public ScopedRoutesTestBase {
 protected:
   void setup() {
-    ON_CALL(factory_context_.cluster_manager_, adsMux())
+    ON_CALL(server_factory_context_.cluster_manager_, adsMux())
         .WillByDefault(Return(std::make_shared<::Envoy::Config::NullGrpcMuxImpl>()));
 
     InSequence s;
-    // Since factory_context_.cluster_manager_.subscription_factory_.callbacks_ is taken by the SRDS
-    // subscription. We need to return a different MockSubscription here for each RDS subscription.
-    // To build the map from RDS route_config_name to the RDS subscription, we need to get the
-    // route_config_name by mocking start() on the Config::Subscription.
+    // Since server_factory_context_.cluster_manager_.subscription_factory_.callbacks_ is taken by
+    // the SRDS subscription. We need to return a different MockSubscription here for each RDS
+    // subscription. To build the map from RDS route_config_name to the RDS subscription, we need to
+    // get the route_config_name by mocking start() on the Config::Subscription.
 
     // srds subscription
-    EXPECT_CALL(factory_context_.cluster_manager_.subscription_factory_,
+    EXPECT_CALL(server_factory_context_.cluster_manager_.subscription_factory_,
                 subscriptionFromConfigSource(_, _, _, _))
         .Times(AnyNumber());
     // rds subscription
@@ -173,11 +176,11 @@ name: foo_scoped_routes
         scoped_routes_config;
     TestUtility::loadFromYaml(config_yaml, scoped_routes_config);
     provider_ = config_provider_manager_->createXdsConfigProvider(
-        scoped_routes_config.scoped_rds(), factory_context_, "foo.",
+        scoped_routes_config.scoped_rds(), server_factory_context_, context_init_manager_, "foo.",
         ScopedRoutesConfigProviderManagerOptArg(scoped_routes_config.name(),
                                                 scoped_routes_config.rds_config_source(),
                                                 scoped_routes_config.scope_key_builder()));
-    srds_subscription_ = factory_context_.cluster_manager_.subscription_factory_.callbacks_;
+    srds_subscription_ = server_factory_context_.cluster_manager_.subscription_factory_.callbacks_;
   }
 
   // Helper function which pushes an update to given RDS subscription, the start(_) of the
@@ -235,7 +238,6 @@ route_configuration_name: foo_routes
   Protobuf::RepeatedPtrField<ProtobufWkt::Any> resources;
   parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml);
   EXPECT_THROW(srds_subscription_->onConfigUpdate(resources, "1"), ProtoValidationException);
-
   EXPECT_THROW_WITH_REGEX(
       srds_subscription_->onConfigUpdate(anyToResource(resources, "1"), {}, "1"), EnvoyException,
       "Error adding/updating scoped route\\(s\\): Proto constraint validation failed.*");
@@ -291,12 +293,12 @@ route_configuration_name: foo_routes
     - string_key: x-bar-key
 )EOF";
   parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml2);
-  EXPECT_NO_THROW(srds_subscription_->onConfigUpdate(resources, "1"));
+  init_watcher_.expectReady().Times(1); // Only the SRDS parent_init_target_.
   context_init_manager_.initialize(init_watcher_);
-  init_watcher_.expectReady().Times(3); // SRDS x2 and RDS "foo_routes"
-  EXPECT_EQ(
-      1UL,
-      factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload").value());
+  EXPECT_NO_THROW(srds_subscription_->onConfigUpdate(resources, "1"));
+  EXPECT_EQ(1UL,
+            server_factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload")
+                .value());
 
   // Verify the config is a ScopedConfigImpl instance, both scopes point to "" as RDS hasn't kicked
   // in yet(NullConfigImpl returned).
@@ -330,9 +332,9 @@ route_configuration_name: foo_routes
   EXPECT_NO_THROW(srds_subscription_->onConfigUpdate(resources, "3"));
   EXPECT_EQ(getScopedRouteMap().size(), 1);
   EXPECT_EQ(getScopedRouteMap().count("foo_scope"), 1);
-  EXPECT_EQ(
-      2UL,
-      factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload").value());
+  EXPECT_EQ(2UL,
+            server_factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload")
+                .value());
   // now scope key "x-bar-key" points to nowhere.
   EXPECT_THAT(getScopedRdsProvider()->config<ScopedConfigImpl>()->getRouteConfig(
                   TestHeaderMapImpl{{"Addr", "x-foo-key;x-bar-key"}}),
@@ -347,8 +349,7 @@ route_configuration_name: foo_routes
 // Tests that multiple uniquely named non-conflict resources are allowed in config updates.
 TEST_F(ScopedRdsTest, MultipleResourcesDelta) {
   setup();
-  init_watcher_.expectReady().Times(3); // SRDS x2 and RDS "foo_routes"
-
+  init_watcher_.expectReady().Times(1);
   const std::string config_yaml = R"EOF(
 name: foo_scope
 route_configuration_name: foo_routes
@@ -370,9 +371,9 @@ route_configuration_name: foo_routes
   // Delta API.
   EXPECT_NO_THROW(srds_subscription_->onConfigUpdate(anyToResource(resources, "2"), {}, "1"));
   context_init_manager_.initialize(init_watcher_);
-  EXPECT_EQ(
-      1UL,
-      factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload").value());
+  EXPECT_EQ(1UL,
+            server_factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload")
+                .value());
   EXPECT_EQ(getScopedRouteMap().size(), 2);
 
   // Verify the config is a ScopedConfigImpl instance, both scopes point to "" as RDS hasn't kicked
@@ -409,9 +410,9 @@ route_configuration_name: foo_routes
   EXPECT_NO_THROW(srds_subscription_->onConfigUpdate(anyToResource(resources, "4"), deletes, "2"));
   EXPECT_EQ(getScopedRouteMap().size(), 1);
   EXPECT_EQ(getScopedRouteMap().count("foo_scope"), 1);
-  EXPECT_EQ(
-      2UL,
-      factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload").value());
+  EXPECT_EQ(2UL,
+            server_factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload")
+                .value());
   // now scope key "x-bar-key" points to nowhere.
   EXPECT_THAT(getScopedRdsProvider()->config<ScopedConfigImpl>()->getRouteConfig(
                   TestHeaderMapImpl{{"Addr", "x-foo-key;x-bar-key"}}),
@@ -424,7 +425,7 @@ route_configuration_name: foo_routes
 }
 
 // Tests that conflict resources in the same push are detected.
-TEST_F(ScopedRdsTest, MultipleResourcesWithKeyConflict) {
+TEST_F(ScopedRdsTest, MultipleResourcesWithKeyConflictSotW) {
   setup();
 
   const std::string config_yaml = R"EOF(
@@ -444,34 +445,56 @@ route_configuration_name: foo_routes
     - string_key: x-foo-key
 )EOF";
   parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml2);
+  init_watcher_.expectReady().Times(0); // The onConfigUpdate will simply throw an exception.
+  context_init_manager_.initialize(init_watcher_);
   EXPECT_THROW_WITH_REGEX(
       srds_subscription_->onConfigUpdate(resources, "1"), EnvoyException,
       ".*scope key conflict found, first scope is 'foo_scope', second scope is 'foo_scope2'");
   EXPECT_EQ(
       // Fully rejected.
-      0UL,
-      factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload").value());
+      0UL, server_factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload")
+               .value());
   // Scope key "x-foo-key" points to nowhere.
   EXPECT_NE(getScopedRdsProvider(), nullptr);
   EXPECT_NE(getScopedRdsProvider()->config<ScopedConfigImpl>(), nullptr);
   EXPECT_THAT(getScopedRdsProvider()->config<ScopedConfigImpl>()->getRouteConfig(
                   TestHeaderMapImpl{{"Addr", "x-foo-key;x-foo-key"}}),
               IsNull());
-  context_init_manager_.initialize(init_watcher_);
-  init_watcher_.expectReady().Times(
-      1); // Just SRDS, RDS "foo_routes" will initialized by the noop init-manager.
   EXPECT_EQ(server_factory_context_.scope_.counter("foo.rds.foo_routes.config_reload").value(),
             0UL);
+}
+
+// Tests that conflict resources in the same push are detected in delta api form.
+TEST_F(ScopedRdsTest, MultipleResourcesWithKeyConflictDelta) {
+  setup();
+
+  const std::string config_yaml = R"EOF(
+name: foo_scope
+route_configuration_name: foo_routes
+key:
+  fragments:
+    - string_key: x-foo-key
+)EOF";
+  Protobuf::RepeatedPtrField<ProtobufWkt::Any> resources;
+  parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml);
+  const std::string config_yaml2 = R"EOF(
+name: foo_scope2
+route_configuration_name: foo_routes
+key:
+  fragments:
+    - string_key: x-foo-key
+)EOF";
+  parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml2);
+  init_watcher_.expectReady().Times(1); // Partial success gets the subscription ready.
+  context_init_manager_.initialize(init_watcher_);
 
-  // Delta API.
-  EXPECT_CALL(context_init_manager_, state()).WillOnce(Return(Init::Manager::State::Initialized));
   EXPECT_THROW_WITH_REGEX(
       srds_subscription_->onConfigUpdate(anyToResource(resources, "2"), {}, "2"), EnvoyException,
       ".*scope key conflict found, first scope is 'foo_scope', second scope is 'foo_scope2'");
   EXPECT_EQ(
       // Partially reject.
-      1UL,
-      factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload").value());
+      1UL, server_factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload")
+               .value());
   // foo_scope update is applied.
   EXPECT_EQ(getScopedRouteMap().size(), 1UL);
   EXPECT_EQ(getScopedRouteMap().count("foo_scope"), 1);
@@ -507,11 +530,10 @@ route_configuration_name: bar_routes
   Protobuf::RepeatedPtrField<ProtobufWkt::Any> resources;
   parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml1);
   parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml2);
-  EXPECT_CALL(context_init_manager_, state()).WillOnce(Return(Init::Manager::State::Uninitialized));
   EXPECT_NO_THROW(srds_subscription_->onConfigUpdate(resources, "1"));
-  EXPECT_EQ(
-      1UL,
-      factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload").value());
+  EXPECT_EQ(1UL,
+            server_factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload")
+                .value());
   // Scope key "x-foo-key" points to nowhere.
   EXPECT_NE(getScopedRdsProvider(), nullptr);
   EXPECT_NE(getScopedRdsProvider()->config<ScopedConfigImpl>(), nullptr);
@@ -521,9 +543,8 @@ route_configuration_name: bar_routes
                   ->getRouteConfig(TestHeaderMapImpl{{"Addr", "x-foo-key;x-foo-key"}})
                   ->name(),
               "");
+  init_watcher_.expectReady().Times(1);
   context_init_manager_.initialize(init_watcher_);
-  init_watcher_.expectReady().Times(
-      3); // SRDS "foo_scope1" and RDS "foo/bar_routes"(though no real push).
   pushRdsConfig({"foo_routes", "bar_routes"}, "111");
   EXPECT_EQ(server_factory_context_.scope_.counter("foo.rds.foo_routes.config_reload").value(),
             1UL);
@@ -547,13 +568,10 @@ route_configuration_name: foo_routes
   // Remove foo_scope1 and add a new scope3 reuses the same scope_key.
   parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml2);
   parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml3);
-  EXPECT_CALL(context_init_manager_, state())
-      .Times(2)
-      .WillRepeatedly(Return(Init::Manager::State::Initialized));
   EXPECT_NO_THROW(srds_subscription_->onConfigUpdate(resources, "2"));
-  EXPECT_EQ(
-      2UL,
-      factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload").value());
+  EXPECT_EQ(2UL,
+            server_factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload")
+                .value());
   // foo_scope is deleted, and foo_scope2 is added.
   EXPECT_EQ(getScopedRouteMap().size(), 2UL);
   EXPECT_EQ(getScopedRouteMap().count("foo_scope1"), 0);
@@ -597,9 +615,9 @@ route_configuration_name: foo_routes
   parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml3);
   parseScopedRouteConfigurationFromYaml(*resources.Add(), config_yaml4);
   EXPECT_NO_THROW(srds_subscription_->onConfigUpdate(resources, "4"));
-  EXPECT_EQ(
-      factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload").value(),
-      3UL);
+  EXPECT_EQ(server_factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload")
+                .value(),
+            3UL);
   EXPECT_EQ(getScopedRouteMap().size(), 2UL);
   EXPECT_EQ(getScopedRouteMap().count("foo_scope3"), 1);
   EXPECT_EQ(getScopedRouteMap().count("foo_scope4"), 1);
@@ -618,8 +636,9 @@ route_configuration_name: foo_routes
 // Tests that only one resource is provided during a config update.
 TEST_F(ScopedRdsTest, InvalidDuplicateResourceSotw) {
   setup();
+  init_watcher_.expectReady().Times(
+      0); // parent_init_target_ ready will be called by onConfigUpdateFailed
   context_init_manager_.initialize(init_watcher_);
-  init_watcher_.expectReady().Times(0);
 
   const std::string config_yaml = R"EOF(
 name: foo_scope
@@ -638,12 +657,8 @@ route_configuration_name: foo_routes
 // Tests that only one resource is provided during a config update.
 TEST_F(ScopedRdsTest, InvalidDuplicateResourceDelta) {
   setup();
+  init_watcher_.expectReady().Times(0);
   context_init_manager_.initialize(init_watcher_);
-  // After the above initialize, the default init_manager should return "Initialized".
-  EXPECT_CALL(context_init_manager_, state()).WillOnce(Return(Init::Manager::State::Initialized));
-  init_watcher_.expectReady().Times(
-      1); // SRDS onConfigUpdate breaks, but first foo_routes will
-          // kick start if it's initialized post-Server/LDS initialization.
 
   const std::string config_yaml = R"EOF(
 name: foo_scope
@@ -661,8 +676,8 @@ route_configuration_name: foo_routes
       "found");
   EXPECT_EQ(
       // Partially reject.
-      1UL,
-      factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload").value());
+      1UL, server_factory_context_.scope_.counter("foo.scoped_rds.foo_scoped_routes.config_reload")
+               .value());
   // foo_scope update is applied.
   EXPECT_EQ(getScopedRouteMap().size(), 1UL);
   EXPECT_EQ(getScopedRouteMap().count("foo_scope"), 1);
@@ -687,14 +702,10 @@ TEST_F(ScopedRdsTest, ConfigUpdateFailure) {
 // config.
 TEST_F(ScopedRdsTest, ConfigDump) {
   setup();
+  init_watcher_.expectReady().Times(1);
   context_init_manager_.initialize(init_watcher_);
-  EXPECT_CALL(context_init_manager_, state())
-      .Times(2) // There are two SRDS pushes.
-      .WillRepeatedly(Return(Init::Manager::State::Initialized));
-  init_watcher_.expectReady().Times(1); // SRDS only, no RDS push.
-
   auto message_ptr =
-      factory_context_.admin_.config_tracker_.config_tracker_callbacks_["route_scopes"]();
+      server_factory_context_.admin_.config_tracker_.config_tracker_callbacks_["route_scopes"]();
   const auto& scoped_routes_config_dump =
       TestUtility::downcastAndValidate<const envoy::admin::v3::ScopedRoutesConfigDump&>(
           *message_ptr);
@@ -741,8 +752,9 @@ stat_prefix: foo
   Envoy::Config::ConfigProviderPtr inline_config = ScopedRoutesConfigProviderUtil::create(
       parseHttpConnectionManagerFromYaml(absl::Substitute(hcm_base_config_yaml, "foo-scoped-routes",
                                                           inline_scoped_route_configs_yaml)),
-      factory_context_, "foo.", *config_provider_manager_);
-  message_ptr = factory_context_.admin_.config_tracker_.config_tracker_callbacks_["route_scopes"]();
+      server_factory_context_, context_init_manager_, "foo.", *config_provider_manager_);
+  message_ptr =
+      server_factory_context_.admin_.config_tracker_.config_tracker_callbacks_["route_scopes"]();
   const auto& scoped_routes_config_dump2 =
       TestUtility::downcastAndValidate<const envoy::admin::v3::ScopedRoutesConfigDump&>(
           *message_ptr);
@@ -811,7 +823,8 @@ route_configuration_name: dynamic-foo-route-config
     version_info: "1"
 )EOF",
                             expected_config_dump);
-  message_ptr = factory_context_.admin_.config_tracker_.config_tracker_callbacks_["route_scopes"]();
+  message_ptr =
+      server_factory_context_.admin_.config_tracker_.config_tracker_callbacks_["route_scopes"]();
   const auto& scoped_routes_config_dump3 =
       TestUtility::downcastAndValidate<const envoy::admin::v3::ScopedRoutesConfigDump&>(
           *message_ptr);
@@ -844,7 +857,8 @@ route_configuration_name: dynamic-foo-route-config
     version_info: "2"
 )EOF",
                             expected_config_dump);
-  message_ptr = factory_context_.admin_.config_tracker_.config_tracker_callbacks_["route_scopes"]();
+  message_ptr =
+      server_factory_context_.admin_.config_tracker_.config_tracker_callbacks_["route_scopes"]();
   const auto& scoped_routes_config_dump4 =
       TestUtility::downcastAndValidate<const envoy::admin::v3::ScopedRoutesConfigDump&>(
           *message_ptr);
@@ -862,7 +876,8 @@ route_configuration_name: static-foo-route-config
 key:
   fragments: { string_key: "172.30.30.10" }
 )EOF"),
-                   factory_context_, Envoy::Config::ConfigProviderManager::NullOptionalArg()),
+                   server_factory_context_,
+                   Envoy::Config::ConfigProviderManager::NullOptionalArg()),
                ".*");
 }
 
diff --git a/test/integration/BUILD b/test/integration/BUILD
index fa47c26579144..57e5487d87b56 100644
--- a/test/integration/BUILD
+++ b/test/integration/BUILD
@@ -1029,3 +1029,26 @@ envoy_cc_test(
         "@envoy_api//envoy/extensions/filters/network/http_connection_manager/v3:pkg_cc_proto",
     ],
 )
+
+envoy_cc_test(
+    name = "listener_lds_integration_test",
+    srcs = [
+        "listener_lds_integration_test.cc",
+    ],
+    deps = [
+        ":http_integration_lib",
+        "//source/common/config:api_version_lib",
+        "//source/common/config:resources_lib",
+        "//source/common/event:dispatcher_includes",
+        "//source/common/event:dispatcher_lib",
+        "//source/common/network:connection_lib",
+        "//source/common/network:utility_lib",
+        "//test/common/grpc:grpc_client_integration_lib",
+        "//test/test_common:utility_lib",
+        "@envoy_api//envoy/api/v2:pkg_cc_proto",
+        "@envoy_api//envoy/config/bootstrap/v3:pkg_cc_proto",
+        "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
+        "@envoy_api//envoy/config/route/v3:pkg_cc_proto",
+        "@envoy_api//envoy/extensions/filters/network/http_connection_manager/v3:pkg_cc_proto",
+    ],
+)
diff --git a/test/integration/listener_lds_integration_test.cc b/test/integration/listener_lds_integration_test.cc
new file mode 100644
index 0000000000000..ce6c4fdff55f2
--- /dev/null
+++ b/test/integration/listener_lds_integration_test.cc
@@ -0,0 +1,265 @@
+#include "envoy/api/v2/discovery.pb.h"
+#include "envoy/config/bootstrap/v3/bootstrap.pb.h"
+#include "envoy/config/core/v3/config_source.pb.h"
+#include "envoy/config/core/v3/grpc_service.pb.h"
+#include "envoy/config/route/v3/route.pb.h"
+#include "envoy/config/route/v3/scoped_route.pb.h"
+#include "envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.pb.h"
+
+#include "common/config/api_version.h"
+#include "common/config/resources.h"
+#include "common/config/version_converter.h"
+
+#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"
+
+namespace Envoy {
+namespace {
+
+class ListenerIntegrationTest : public HttpIntegrationTest,
+                                public Grpc::GrpcClientIntegrationParamTest {
+protected:
+  struct FakeUpstreamInfo {
+    FakeHttpConnectionPtr connection_;
+    FakeUpstream* upstream_{};
+    absl::flat_hash_map<std::string, FakeStreamPtr> stream_by_resource_name_;
+  };
+
+  ListenerIntegrationTest()
+      : HttpIntegrationTest(Http::CodecClient::Type::HTTP1, ipVersion(), realTime()) {}
+
+  ~ListenerIntegrationTest() override {
+    resetConnections();
+    cleanupUpstreamAndDownstream();
+  }
+
+  void initialize() override {
+    // We want to use the GRPC based LDS.
+    use_lds_ = false;
+    setUpstreamCount(1);
+    defer_listener_finalization_ = true;
+
+    config_helper_.addConfigModifier([](envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+      // Add the static cluster to serve LDS.
+      auto* lds_cluster = bootstrap.mutable_static_resources()->add_clusters();
+      lds_cluster->MergeFrom(bootstrap.static_resources().clusters()[0]);
+      lds_cluster->set_name("lds_cluster");
+      lds_cluster->mutable_http2_protocol_options();
+
+      // Add the static cluster to serve RDS.
+      auto* rds_cluster = bootstrap.mutable_static_resources()->add_clusters();
+      rds_cluster->MergeFrom(bootstrap.static_resources().clusters()[0]);
+      rds_cluster->set_name("rds_cluster");
+      rds_cluster->mutable_http2_protocol_options();
+    });
+
+    config_helper_.addConfigModifier(
+        [this](
+            envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
+                http_connection_manager) {
+          auto* rds_config = http_connection_manager.mutable_rds();
+          rds_config->set_route_config_name(route_table_name_);
+          envoy::config::core::v3::ApiConfigSource* rds_api_config_source =
+              rds_config->mutable_config_source()->mutable_api_config_source();
+          rds_api_config_source->set_api_type(envoy::config::core::v3::ApiConfigSource::GRPC);
+          envoy::config::core::v3::GrpcService* grpc_service =
+              rds_api_config_source->add_grpc_services();
+          setGrpcService(*grpc_service, "rds_cluster", getRdsFakeUpstream().localAddress());
+        });
+
+    // Note this has to be the last modifier as it nuke static_resource listeners.
+    setUpGrpcLds();
+    HttpIntegrationTest::initialize();
+  }
+  void setUpGrpcLds() {
+    config_helper_.addConfigModifier([this](envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+      listener_config_.Swap(bootstrap.mutable_static_resources()->mutable_listeners(0));
+      listener_config_.set_name(listener_name_);
+      ENVOY_LOG_MISC(error, "listener config: {}", listener_config_.DebugString());
+      bootstrap.mutable_static_resources()->mutable_listeners()->Clear();
+      auto* lds_api_config_source =
+          bootstrap.mutable_dynamic_resources()->mutable_lds_config()->mutable_api_config_source();
+      lds_api_config_source->set_api_type(envoy::config::core::v3::ApiConfigSource::GRPC);
+      envoy::config::core::v3::GrpcService* grpc_service =
+          lds_api_config_source->add_grpc_services();
+      setGrpcService(*grpc_service, "lds_cluster", getLdsFakeUpstream().localAddress());
+    });
+  }
+
+  void createUpstreams() override {
+    HttpIntegrationTest::createUpstreams();
+    // Create the LDS upstream (fake_upstreams_[1]).
+    fake_upstreams_.emplace_back(new FakeUpstream(0, FakeHttpConnection::Type::HTTP2, version_,
+                                                  timeSystem(), enable_half_close_));
+    // Create the RDS upstream (fake_upstreams_[2]).
+    fake_upstreams_.emplace_back(new FakeUpstream(0, FakeHttpConnection::Type::HTTP2, version_,
+                                                  timeSystem(), enable_half_close_));
+    for (auto& upstream : fake_upstreams_) {
+      upstream->set_allow_unexpected_disconnects(true);
+    }
+  }
+
+  void resetFakeUpstreamInfo(FakeUpstreamInfo* upstream_info) {
+    ASSERT(upstream_info->upstream_ != nullptr);
+
+    AssertionResult result = upstream_info->connection_->close();
+    RELEASE_ASSERT(result, result.message());
+    result = upstream_info->connection_->waitForDisconnect();
+    RELEASE_ASSERT(result, result.message());
+    upstream_info->connection_.reset();
+  }
+
+  void resetConnections() {
+    if (rds_upstream_info_.upstream_ != nullptr) {
+      resetFakeUpstreamInfo(&rds_upstream_info_);
+    }
+    resetFakeUpstreamInfo(&lds_upstream_info_);
+  }
+
+  FakeUpstream& getLdsFakeUpstream() const { return *fake_upstreams_[1]; }
+
+  FakeUpstream& getRdsFakeUpstream() const { return *fake_upstreams_[2]; }
+
+  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_by_resource_name_[resource_name]->startGrpcStream();
+  }
+
+  void createRdsStream(const std::string& resource_name) {
+    createStream(&rds_upstream_info_, getRdsFakeUpstream(), resource_name);
+  }
+
+  void createLdsStream() {
+    createStream(&lds_upstream_info_, getLdsFakeUpstream(), listener_name_);
+  }
+
+  void sendLdsResponse(const std::vector<std::string>& listener_configs,
+                       const std::string& version) {
+    API_NO_BOOST(envoy::api::v2::DiscoveryResponse) response;
+    response.set_version_info(version);
+    response.set_type_url(Config::TypeUrl::get().Listener);
+    for (const auto& listener_blob : listener_configs) {
+      const auto listener_config =
+          TestUtility::parseYaml<envoy::config::listener::v3::Listener>(listener_blob);
+      response.add_resources()->PackFrom(API_DOWNGRADE(listener_config));
+    }
+    ASSERT(lds_upstream_info_.stream_by_resource_name_[listener_name_] != nullptr);
+    lds_upstream_info_.stream_by_resource_name_[listener_name_]->sendGrpcMessage(response);
+  }
+
+  void sendRdsResponse(const std::string& route_config, const std::string& version) {
+    API_NO_BOOST(envoy::api::v2::DiscoveryResponse) response;
+    response.set_version_info(version);
+    response.set_type_url(Config::TypeUrl::get().RouteConfiguration);
+    const auto route_configuration =
+        TestUtility::parseYaml<envoy::config::route::v3::RouteConfiguration>(route_config);
+    response.add_resources()->PackFrom(API_DOWNGRADE(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);
+  }
+  envoy::config::listener::v3::Listener listener_config_;
+  std::string listener_name_{"testing-listener-0"};
+  std::string route_table_name_{"testing-route-table-0"};
+  FakeUpstreamInfo lds_upstream_info_;
+  FakeUpstreamInfo rds_upstream_info_;
+};
+
+INSTANTIATE_TEST_SUITE_P(IpVersionsAndGrpcTypes, ListenerIntegrationTest,
+                         GRPC_CLIENT_INTEGRATION_PARAMS);
+
+// Tests that a LDS deletion before Server initManager been initialized will not block the Server
+// from starting.
+TEST_P(ListenerIntegrationTest, RemoveLastUninitializedListener) {
+  on_server_init_function_ = [&]() {
+    createLdsStream();
+    sendLdsResponse({MessageUtil::getYamlStringFromMessage(listener_config_)}, "1");
+    createRdsStream(route_table_name_);
+  };
+  initialize();
+  registerTestServerPorts({listener_name_});
+  test_server_->waitForCounterGe("listener_manager.lds.update_success", 1);
+  // testing-listener-0 is not initialized as we haven't push any RDS yet.
+  EXPECT_EQ(test_server_->server().initManager().state(), Init::Manager::State::Initializing);
+  // Workers not started, the LDS added listener 0 is in active_listeners_ list.
+  EXPECT_EQ(test_server_->server().listenerManager().listeners().size(), 1);
+
+  // This actually deletes the only listener.
+  sendLdsResponse({}, "2");
+  test_server_->waitForCounterGe("listener_manager.lds.update_success", 2);
+  EXPECT_EQ(test_server_->server().listenerManager().listeners().size(), 0);
+  // Server instance is ready now because the listener's destruction marked the listener
+  // initialized.
+  EXPECT_EQ(test_server_->server().initManager().state(), Init::Manager::State::Initialized);
+}
+
+// Tests that a LDS adding listener works as expected.
+TEST_P(ListenerIntegrationTest, BasicSuccess) {
+  on_server_init_function_ = [&]() {
+    createLdsStream();
+    sendLdsResponse({MessageUtil::getYamlStringFromMessage(listener_config_)}, "1");
+    createRdsStream(route_table_name_);
+  };
+  initialize();
+  test_server_->waitForCounterGe("listener_manager.lds.update_success", 1);
+  // testing-listener-0 is not initialized as we haven't pushed any RDS yet.
+  EXPECT_EQ(test_server_->server().initManager().state(), Init::Manager::State::Initializing);
+  // Workers not started, the LDS added listener 0 is in active_listeners_ list.
+  EXPECT_EQ(test_server_->server().listenerManager().listeners().size(), 1);
+  registerTestServerPorts({listener_name_});
+
+  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, route_table_name_, "cluster_0"), "1");
+  test_server_->waitForCounterGe(
+      fmt::format("http.config_test.rds.{}.update_success", route_table_name_), 1);
+  // Now testing-listener-0 finishes initialization, Server initManager will be ready.
+  EXPECT_EQ(test_server_->server().initManager().state(), Init::Manager::State::Initialized);
+
+  test_server_->waitUntilListenersReady();
+  // NOTE: The line above doesn't tell you if listener is up and listening.
+  test_server_->waitForCounterGe("listener_manager.listener_create_success", 1);
+  // Request is sent to cluster_0.
+
+  codec_client_ = makeHttpConnection(lookupPort(listener_name_));
+  int response_size = 800;
+  int request_size = 10;
+  Http::TestHeaderMapImpl response_headers{{":status", "200"},
+                                           {"server_id", "cluster_0, backend_0"}};
+  auto response = sendRequestAndWaitForResponse(
+      Http::TestHeaderMapImpl{
+          {":method", "GET"}, {":path", "/"}, {":authority", "host"}, {":scheme", "http"}},
+      request_size, response_headers, response_size, /*cluster_0*/ 0);
+  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();
+}
+
+} // namespace
+} // namespace Envoy
diff --git a/test/mocks/config/mocks.h b/test/mocks/config/mocks.h
index 0cd19f675f1dc..e7915fd3ece02 100644
--- a/test/mocks/config/mocks.h
+++ b/test/mocks/config/mocks.h
@@ -106,16 +106,16 @@ class MockConfigProviderManager : public ConfigProviderManager {
 
   MOCK_METHOD(ConfigProviderPtr, createXdsConfigProvider,
               (const Protobuf::Message& config_source_proto,
-               Server::Configuration::FactoryContext& factory_context,
-               const std::string& stat_prefix,
+               Server::Configuration::ServerFactoryContext& factory_context,
+               Init::Manager& init_manager, const std::string& stat_prefix,
                const Envoy::Config::ConfigProviderManager::OptionalArg& optarg));
   MOCK_METHOD(ConfigProviderPtr, createStaticConfigProvider,
               (const Protobuf::Message& config_proto,
-               Server::Configuration::FactoryContext& factory_context,
+               Server::Configuration::ServerFactoryContext& factory_context,
                const Envoy::Config::ConfigProviderManager::OptionalArg& optarg));
   MOCK_METHOD(ConfigProviderPtr, createStaticConfigProvider,
               (std::vector<std::unique_ptr<const Protobuf::Message>> && config_protos,
-               Server::Configuration::FactoryContext& factory_context,
+               Server::Configuration::ServerFactoryContext& factory_context,
                const Envoy::Config::ConfigProviderManager::OptionalArg& optarg));
 };
 
diff --git a/test/mocks/router/mocks.h b/test/mocks/router/mocks.h
index 0dccccdc7498c..1c86016e6f6c6 100644
--- a/test/mocks/router/mocks.h
+++ b/test/mocks/router/mocks.h
@@ -452,11 +452,12 @@ class MockRouteConfigProviderManager : public RouteConfigProviderManager {
 
   MOCK_METHOD(RouteConfigProviderSharedPtr, createRdsRouteConfigProvider,
               (const envoy::extensions::filters::network::http_connection_manager::v3::Rds& rds,
-               Server::Configuration::FactoryContext& factory_context,
+               Server::Configuration::ServerFactoryContext& factory_context,
                const std::string& stat_prefix, Init::Manager& init_manager));
   MOCK_METHOD(RouteConfigProviderPtr, createStaticRouteConfigProvider,
               (const envoy::config::route::v3::RouteConfiguration& route_config,
-               Server::Configuration::FactoryContext& factory_context));
+               Server::Configuration::ServerFactoryContext& factory_context,
+               ProtobufMessage::ValidationVisitor& validator));
 };
 
 class MockScopedConfig : public ScopedConfig {
diff --git a/test/mocks/server/mocks.cc b/test/mocks/server/mocks.cc
index 6081292b9a20f..318fbb4de1928 100644
--- a/test/mocks/server/mocks.cc
+++ b/test/mocks/server/mocks.cc
@@ -208,6 +208,7 @@ MockServerFactoryContext::MockServerFactoryContext()
   ON_CALL(*this, admin()).WillByDefault(ReturnRef(admin_));
   ON_CALL(*this, api()).WillByDefault(ReturnRef(api_));
   ON_CALL(*this, timeSource()).WillByDefault(ReturnRef(time_system_));
+  ON_CALL(*this, messageValidationContext()).WillByDefault(ReturnRef(validation_context_));
   ON_CALL(*this, messageValidationVisitor())
       .WillByDefault(ReturnRef(ProtobufMessage::getStrictValidationVisitor()));
   ON_CALL(*this, api()).WillByDefault(ReturnRef(api_));
@@ -234,6 +235,7 @@ MockFactoryContext::MockFactoryContext()
   ON_CALL(*this, api()).WillByDefault(ReturnRef(api_));
   ON_CALL(*this, timeSource()).WillByDefault(ReturnRef(time_system_));
   ON_CALL(*this, overloadManager()).WillByDefault(ReturnRef(overload_manager_));
+  ON_CALL(*this, messageValidationContext()).WillByDefault(ReturnRef(validation_context_));
   ON_CALL(*this, messageValidationVisitor())
       .WillByDefault(ReturnRef(ProtobufMessage::getStrictValidationVisitor()));
   ON_CALL(*this, api()).WillByDefault(ReturnRef(api_));
diff --git a/test/mocks/server/mocks.h b/test/mocks/server/mocks.h
index 76ff8f853e7f4..2f9aa0f75befe 100644
--- a/test/mocks/server/mocks.h
+++ b/test/mocks/server/mocks.h
@@ -486,6 +486,7 @@ class MockServerFactoryContext : public virtual ServerFactoryContext {
   MOCK_METHOD(Server::Admin&, admin, ());
   MOCK_METHOD(TimeSource&, timeSource, ());
   Event::TestTimeSystem& timeSystem() { return time_system_; }
+  MOCK_METHOD(ProtobufMessage::ValidationContext&, messageValidationContext, ());
   MOCK_METHOD(ProtobufMessage::ValidationVisitor&, messageValidationVisitor, ());
   MOCK_METHOD(Api::Api&, api, ());
 
@@ -497,6 +498,7 @@ class MockServerFactoryContext : public virtual ServerFactoryContext {
   testing::NiceMock<Envoy::Runtime::MockLoader> runtime_loader_;
   testing::NiceMock<Stats::MockIsolatedStatsStore> scope_;
   testing::NiceMock<ThreadLocal::MockInstance> thread_local_;
+  testing::NiceMock<ProtobufMessage::MockValidationContext> validation_context_;
   Singleton::ManagerPtr singleton_manager_;
   testing::NiceMock<MockAdmin> admin_;
   Event::GlobalTimeSystem time_system_;
@@ -534,6 +536,7 @@ class MockFactoryContext : public virtual FactoryContext {
   Grpc::Context& grpcContext() override { return grpc_context_; }
   Http::Context& httpContext() override { return http_context_; }
   MOCK_METHOD(ProcessContextOptRef, processContext, ());
+  MOCK_METHOD(ProtobufMessage::ValidationContext&, messageValidationContext, ());
   MOCK_METHOD(ProtobufMessage::ValidationVisitor&, messageValidationVisitor, ());
   MOCK_METHOD(Api::Api&, api, ());
 
@@ -553,6 +556,7 @@ class MockFactoryContext : public virtual FactoryContext {
   testing::NiceMock<MockAdmin> admin_;
   Stats::IsolatedStoreImpl listener_scope_;
   Event::GlobalTimeSystem time_system_;
+  testing::NiceMock<ProtobufMessage::MockValidationContext> validation_context_;
   testing::NiceMock<MockOverloadManager> overload_manager_;
   Grpc::ContextImpl grpc_context_;
   Http::ContextImpl http_context_;
diff --git a/test/server/BUILD b/test/server/BUILD
index 68d011911b33a..d409bcf0f2f01 100644
--- a/test/server/BUILD
+++ b/test/server/BUILD
@@ -203,8 +203,10 @@ envoy_cc_test_library(
     hdrs = ["listener_manager_impl_test.h"],
     data = ["//test/extensions/transport_sockets/tls/test_data:certs"],
     deps = [
+        "//source/common/init:manager_lib",
         "//source/server:api_listener_lib",
         "//source/server:listener_lib",
+        "//test/mocks/init:init_mocks",
         "//test/mocks/network:network_mocks",
         "//test/mocks/server:server_mocks",
         "//test/test_common:environment_lib",
diff --git a/test/server/listener_manager_impl_test.cc b/test/server/listener_manager_impl_test.cc
index 63cc86ffe3f23..4cec7b4654eb1 100644
--- a/test/server/listener_manager_impl_test.cc
+++ b/test/server/listener_manager_impl_test.cc
@@ -17,6 +17,7 @@
 
 #include "common/api/os_sys_calls_impl.h"
 #include "common/config/metadata.h"
+#include "common/init/manager_impl.h"
 #include "common/network/address_impl.h"
 #include "common/network/io_socket_handle_impl.h"
 #include "common/network/utility.h"
@@ -25,6 +26,7 @@
 #include "extensions/filters/listener/original_dst/original_dst.h"
 #include "extensions/transport_sockets/tls/ssl_socket.h"
 
+#include "test/mocks/init/mocks.h"
 #include "test/server/utility.h"
 #include "test/test_common/network_utility.h"
 #include "test/test_common/registry.h"
@@ -33,14 +35,16 @@
 #include "absl/strings/escaping.h"
 #include "absl/strings/match.h"
 
-using testing::AtLeast;
-using testing::InSequence;
-using testing::Throw;
-
 namespace Envoy {
 namespace Server {
 namespace {
 
+using testing::AtLeast;
+using testing::InSequence;
+using testing::Return;
+using testing::ReturnRef;
+using testing::Throw;
+
 class ListenerManagerImplWithDispatcherStatsTest : public ListenerManagerImplTest {
 protected:
   ListenerManagerImplWithDispatcherStatsTest() { enable_dispatcher_stats_ = true; }
@@ -673,6 +677,120 @@ name: foo
   EXPECT_CALL(*listener_foo, onDestroy());
 }
 
+// Tests that when listener tears down, server's initManager is notified.
+TEST_F(ListenerManagerImplTest, ListenerTeardownNotifiesServerInitManager) {
+  time_system_.setSystemTime(std::chrono::milliseconds(1001001001001));
+
+  InSequence s;
+
+  auto* lds_api = new MockLdsApi();
+  EXPECT_CALL(listener_factory_, createLdsApi_(_)).WillOnce(Return(lds_api));
+  envoy::config::core::v3::ConfigSource lds_config;
+  manager_->createLdsApi(lds_config);
+
+  EXPECT_CALL(*lds_api, versionInfo()).WillOnce(Return(""));
+  checkConfigDump(R"EOF(
+static_listeners:
+)EOF");
+
+  const std::string listener_foo_yaml = R"EOF(
+name: "foo"
+address:
+  socket_address:
+    address: "127.0.0.1"
+    port_value: 1234
+filter_chains: {}
+  )EOF";
+
+  Init::ManagerImpl server_init_mgr("server-init-manager");
+  Init::ExpectableWatcherImpl server_init_watcher("server-init-watcher");
+  { // Add and remove a listener before starting workers.
+    ListenerHandle* listener_foo = expectListenerCreate(true, true);
+    EXPECT_CALL(server_, initManager()).WillOnce(ReturnRef(server_init_mgr));
+    EXPECT_CALL(listener_factory_, createListenSocket(_, _, _, {true}));
+    EXPECT_TRUE(manager_->addOrUpdateListener(parseListenerFromV2Yaml(listener_foo_yaml),
+                                              "version1", true));
+    checkStats(1, 0, 0, 0, 1, 0);
+
+    EXPECT_CALL(*lds_api, versionInfo()).WillOnce(Return("version1"));
+    checkConfigDump(R"EOF(
+version_info: version1
+static_listeners:
+dynamic_listeners:
+  - name: foo
+    warming_state:
+      version_info: version1
+      listener:
+        "@type": type.googleapis.com/envoy.api.v2.Listener
+        name: foo
+        address:
+          socket_address:
+            address: 127.0.0.1
+            port_value: 1234
+        filter_chains: {}
+      last_updated:
+        seconds: 1001001001
+        nanos: 1000000
+)EOF");
+    EXPECT_CALL(listener_foo->target_, initialize()).Times(1);
+    server_init_mgr.initialize(server_init_watcher);
+    // Since listener_foo->target_ is not ready, the listener's listener_init_target will not be
+    // ready until the destruction happens.
+    server_init_watcher.expectReady().Times(1);
+    EXPECT_CALL(*listener_foo, onDestroy());
+    EXPECT_TRUE(manager_->removeListener("foo"));
+  }
+  // Listener foo's listener_init_target_ is the only target added to server_init_mgr.
+  EXPECT_EQ(server_init_mgr.state(), Init::Manager::State::Initialized);
+
+  EXPECT_CALL(*lds_api, versionInfo()).WillOnce(Return(""));
+  checkConfigDump(R"EOF(
+static_listeners:
+)EOF");
+
+  EXPECT_CALL(*worker_, start(_));
+  manager_->startWorkers(guard_dog_);
+
+  // Now add new version listener foo after workers start, note it's fine that server_init_mgr is
+  // initialized, as no target will be added to it.
+  time_system_.setSystemTime(std::chrono::milliseconds(2002002002002));
+  EXPECT_CALL(server_, initManager()).Times(0); // No target added to server init manager.
+  server_init_watcher.expectReady().Times(0);
+  {
+    ListenerHandle* listener_foo2 = expectListenerCreate(true, true);
+    EXPECT_CALL(listener_factory_, createListenSocket(_, _, _, {true}));
+    // Version 2 listener will be initialized by listener manager directly.
+    EXPECT_CALL(listener_foo2->target_, initialize()).Times(1);
+    EXPECT_TRUE(manager_->addOrUpdateListener(parseListenerFromV2Yaml(listener_foo_yaml),
+                                              "version2", true));
+    // Version2 is in warming list as listener_foo2->target_ is not ready yet.
+    checkStats(/*added=*/2, 0, /*removed=*/1, /*warming=*/1, 0, 0);
+    EXPECT_CALL(*lds_api, versionInfo()).WillOnce(Return("version2"));
+    checkConfigDump(R"EOF(
+  version_info: version2
+  static_listeners:
+  dynamic_listeners:
+    - name: foo
+      warming_state:
+        version_info: version2
+        listener:
+          "@type": type.googleapis.com/envoy.api.v2.Listener
+          name: foo
+          address:
+            socket_address:
+              address: 127.0.0.1
+              port_value: 1234
+          filter_chains: {}
+        last_updated:
+          seconds: 2002002002
+          nanos: 2000000
+  )EOF");
+    // Delete foo-listener again.
+    EXPECT_CALL(*listener_foo2, onDestroy());
+    EXPECT_TRUE(manager_->removeListener("foo"));
+  }
+}
+
 TEST_F(ListenerManagerImplTest, AddOrUpdateListener) {
   time_system_.setSystemTime(std::chrono::milliseconds(1001001001001));