diff --git a/include/envoy/config/subscription.h b/include/envoy/config/subscription.h
index badd3bad890d3..798168e7182ba 100644
--- a/include/envoy/config/subscription.h
+++ b/include/envoy/config/subscription.h
@@ -18,7 +18,8 @@ class SubscriptionCallbacks {
   virtual ~SubscriptionCallbacks() = default;
 
   /**
-   * Called when a configuration update is received.
+   * Called when a state-of-the-world configuration update is received. (State-of-the-world is
+   * everything other than delta gRPC - filesystem, HTTP, non-delta gRPC).
    * @param resources vector of fetched resources corresponding to the configuration update.
    * @param version_info supplies the version information as supplied by the xDS discovery response.
    * @throw EnvoyException with reason if the configuration is rejected. Otherwise the configuration
@@ -28,9 +29,6 @@ class SubscriptionCallbacks {
   virtual void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
                               const std::string& version_info) PURE;
 
-  // TODO(fredlas) it is a HACK that there are two of these. After delta CDS is merged,
-  //               I intend to reimplement all state-of-the-world xDSes' use of onConfigUpdate
-  //               in terms of this delta-style one (and remove the original).
   /**
    * Called when a delta configuration update is received.
    * @param added_resources resources newly added since the previous fetch.
diff --git a/source/common/router/rds_impl.cc b/source/common/router/rds_impl.cc
index a323d45511284..5bba8f9f0a990 100644
--- a/source/common/router/rds_impl.cc
+++ b/source/common/router/rds_impl.cc
@@ -92,15 +92,9 @@ RdsRouteConfigSubscription::~RdsRouteConfigSubscription() {
 void RdsRouteConfigSubscription::onConfigUpdate(
     const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
     const std::string& version_info) {
-  if (resources.empty()) {
-    ENVOY_LOG(debug, "Missing RouteConfiguration for {} in onConfigUpdate()", route_config_name_);
-    stats_.update_empty_.inc();
-    init_target_.ready();
+  if (!validateUpdateSize(resources.size())) {
     return;
   }
-  if (resources.size() != 1) {
-    throw EnvoyException(fmt::format("Unexpected RDS resource length: {}", resources.size()));
-  }
   auto route_config = MessageUtil::anyConvert<envoy::api::v2::RouteConfiguration>(resources[0]);
   MessageUtil::validate(route_config);
   if (route_config.name() != route_config_name_) {
@@ -131,12 +125,48 @@ void RdsRouteConfigSubscription::onConfigUpdate(
   init_target_.ready();
 }
 
+void RdsRouteConfigSubscription::onConfigUpdate(
+    const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& added_resources,
+    const Protobuf::RepeatedPtrField<std::string>& removed_resources,
+    const std::string& system_version_info) {
+  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.
+    ENVOY_LOG(
+        error,
+        "Server sent a delta RDS update attempting to remove a resource (name: {}). Ignoring.",
+        removed_resources[0]);
+  }
+  Protobuf::RepeatedPtrField<ProtobufWkt::Any> unwrapped_resource;
+  if (!added_resources.empty()) {
+    *unwrapped_resource.Add() = added_resources[0].resource();
+    onConfigUpdate(unwrapped_resource, added_resources[0].version());
+  } else {
+    onConfigUpdate({}, system_version_info);
+    return;
+  }
+}
+
 void RdsRouteConfigSubscription::onConfigUpdateFailed(const EnvoyException*) {
   // We need to allow server startup to continue, even if we have a bad
   // config.
   init_target_.ready();
 }
 
+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();
+    return false;
+  }
+  if (num_resources != 1) {
+    throw EnvoyException(fmt::format("Unexpected RDS resource length: {}", num_resources));
+    // (would be a return false here)
+  }
+  return true;
+}
+
 RdsRouteConfigProviderImpl::RdsRouteConfigProviderImpl(
     RdsRouteConfigSubscriptionSharedPtr&& subscription,
     Server::Configuration::FactoryContext& factory_context)
diff --git a/source/common/router/rds_impl.h b/source/common/router/rds_impl.h
index 2b59fe690fec6..015dd7e89dda2 100644
--- a/source/common/router/rds_impl.h
+++ b/source/common/router/rds_impl.h
@@ -109,13 +109,11 @@ class RdsRouteConfigSubscription : Envoy::Config::SubscriptionCallbacks,
   RouteConfigUpdatePtr& routeConfigUpdate() { return config_update_info_; }
 
   // Config::SubscriptionCallbacks
-  // TODO(fredlas) deduplicate
   void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
                       const std::string& version_info) override;
-  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>&,
-                      const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override {
-    NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
-  }
+  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& added_resources,
+                      const Protobuf::RepeatedPtrField<std::string>& removed_resources,
+                      const std::string&) override;
   void onConfigUpdateFailed(const EnvoyException* e) override;
   std::string resourceName(const ProtobufWkt::Any& resource) override {
     return MessageUtil::anyConvert<envoy::api::v2::RouteConfiguration>(resource).name();
@@ -128,6 +126,8 @@ class RdsRouteConfigSubscription : Envoy::Config::SubscriptionCallbacks,
       const std::string& stat_prefix,
       RouteConfigProviderManagerImpl& route_config_provider_manager);
 
+  bool validateUpdateSize(int num_resources);
+
   std::unique_ptr<Envoy::Config::Subscription> subscription_;
   const std::string route_config_name_;
   Server::Configuration::FactoryContext& factory_context_;
diff --git a/source/common/secret/sds_api.cc b/source/common/secret/sds_api.cc
index d5bc9addbacc2..d9160858eb471 100644
--- a/source/common/secret/sds_api.cc
+++ b/source/common/secret/sds_api.cc
@@ -31,13 +31,7 @@ SdsApi::SdsApi(const LocalInfo::LocalInfo& local_info, Event::Dispatcher& dispat
 
 void SdsApi::onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
                             const std::string&) {
-  if (resources.empty()) {
-    throw EnvoyException(
-        fmt::format("Missing SDS resources for {} in onConfigUpdate()", sds_config_name_));
-  }
-  if (resources.size() != 1) {
-    throw EnvoyException(fmt::format("Unexpected SDS secrets length: {}", resources.size()));
-  }
+  validateUpdateSize(resources.size());
   auto secret = MessageUtil::anyConvert<envoy::api::v2::auth::Secret>(resources[0]);
   MessageUtil::validate(secret);
 
@@ -57,11 +51,29 @@ void SdsApi::onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>&
   init_target_.ready();
 }
 
+void SdsApi::onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& resources,
+                            const Protobuf::RepeatedPtrField<std::string>&, const std::string&) {
+  validateUpdateSize(resources.size());
+  Protobuf::RepeatedPtrField<ProtobufWkt::Any> unwrapped_resource;
+  *unwrapped_resource.Add() = resources[0].resource();
+  onConfigUpdate(unwrapped_resource, resources[0].version());
+}
+
 void SdsApi::onConfigUpdateFailed(const EnvoyException*) {
   // We need to allow server startup to continue, even if we have a bad config.
   init_target_.ready();
 }
 
+void SdsApi::validateUpdateSize(int num_resources) {
+  if (num_resources == 0) {
+    throw EnvoyException(
+        fmt::format("Missing SDS resources for {} in onConfigUpdate()", sds_config_name_));
+  }
+  if (num_resources != 1) {
+    throw EnvoyException(fmt::format("Unexpected SDS secrets length: {}", num_resources));
+  }
+}
+
 void SdsApi::initialize() {
   subscription_ = Envoy::Config::SubscriptionFactory::subscriptionFromConfigSource(
       sds_config_, local_info_, dispatcher_, cluster_manager_, random_, stats_,
diff --git a/source/common/secret/sds_api.h b/source/common/secret/sds_api.h
index 13d7790ed157c..aabb46c5b27dc 100644
--- a/source/common/secret/sds_api.h
+++ b/source/common/secret/sds_api.h
@@ -37,13 +37,10 @@ class SdsApi : public Config::SubscriptionCallbacks {
          std::function<void()> destructor_cb, Api::Api& api);
 
   // Config::SubscriptionCallbacks
-  // TODO(fredlas) deduplicate
   void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
                       const std::string& version_info) override;
   void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>&,
-                      const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override {
-    NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
-  }
+                      const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override;
   void onConfigUpdateFailed(const EnvoyException* e) override;
   std::string resourceName(const ProtobufWkt::Any& resource) override {
     return MessageUtil::anyConvert<envoy::api::v2::auth::Secret>(resource).name();
@@ -56,6 +53,7 @@ class SdsApi : public Config::SubscriptionCallbacks {
   Common::CallbackManager<> update_callback_manager_;
 
 private:
+  void validateUpdateSize(int num_resources);
   void initialize();
   Init::TargetImpl init_target_;
   const LocalInfo::LocalInfo& local_info_;
diff --git a/source/common/upstream/cds_api_impl.cc b/source/common/upstream/cds_api_impl.cc
index 79a888a26dc6a..a4cd39664d288 100644
--- a/source/common/upstream/cds_api_impl.cc
+++ b/source/common/upstream/cds_api_impl.cc
@@ -71,12 +71,14 @@ void CdsApiImpl::onConfigUpdate(
 
   std::vector<std::string> exception_msgs;
   std::unordered_set<std::string> cluster_names;
+  bool any_applied = false;
   for (const auto& resource : added_resources) {
     envoy::api::v2::Cluster cluster;
     try {
       cluster = MessageUtil::anyConvert<envoy::api::v2::Cluster>(resource.resource());
       MessageUtil::validate(cluster);
       if (!cluster_names.insert(cluster.name()).second) {
+        // NOTE: at this point, the first of these duplicates has already been successfully applied.
         throw EnvoyException(fmt::format("duplicate cluster {} found", cluster.name()));
       }
       if (cm_.addOrUpdateCluster(
@@ -107,6 +109,7 @@ void CdsApiImpl::onConfigUpdate(
                   cm_.adsMux().resume(Config::TypeUrl::get().Cluster);
                 }
               })) {
+        any_applied = true;
         ENVOY_LOG(debug, "cds: add/update cluster '{}'", cluster.name());
       }
     } catch (const EnvoyException& e) {
@@ -119,12 +122,14 @@ void CdsApiImpl::onConfigUpdate(
     }
   }
 
+  if (any_applied) {
+    system_version_info_ = system_version_info;
+  }
   runInitializeCallbackIfAny();
   if (!exception_msgs.empty()) {
     throw EnvoyException(
         fmt::format("Error adding/updating cluster(s) {}", StringUtil::join(exception_msgs, ", ")));
   }
-  system_version_info_ = system_version_info;
 }
 
 void CdsApiImpl::onConfigUpdateFailed(const EnvoyException*) {
diff --git a/source/common/upstream/cds_api_impl.h b/source/common/upstream/cds_api_impl.h
index 2550577880e57..cdd2fe9255419 100644
--- a/source/common/upstream/cds_api_impl.h
+++ b/source/common/upstream/cds_api_impl.h
@@ -35,11 +35,11 @@ class CdsApiImpl : public CdsApi,
   const std::string versionInfo() const override { return system_version_info_; }
 
   // Config::SubscriptionCallbacks
-  // TODO(fredlas) deduplicate
   void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
                       const std::string& version_info) override;
-  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>&,
-                      const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override;
+  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& added_resources,
+                      const Protobuf::RepeatedPtrField<std::string>& removed_resources,
+                      const std::string& system_version_info) override;
   void onConfigUpdateFailed(const EnvoyException* e) override;
   std::string resourceName(const ProtobufWkt::Any& resource) override {
     return MessageUtil::anyConvert<envoy::api::v2::Cluster>(resource).name();
diff --git a/source/common/upstream/eds.cc b/source/common/upstream/eds.cc
index 80a9073a312d3..4405dbbaeafc1 100644
--- a/source/common/upstream/eds.cc
+++ b/source/common/upstream/eds.cc
@@ -101,15 +101,9 @@ void EdsClusterImpl::BatchUpdateHelper::batchUpdate(PrioritySet::HostUpdateCb& h
 
 void EdsClusterImpl::onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
                                     const std::string&) {
-  if (resources.empty()) {
-    ENVOY_LOG(debug, "Missing ClusterLoadAssignment for {} in onConfigUpdate()", cluster_name_);
-    info_->stats().update_empty_.inc();
-    onPreInitComplete();
+  if (!validateUpdateSize(resources.size())) {
     return;
   }
-  if (resources.size() != 1) {
-    throw EnvoyException(fmt::format("Unexpected EDS resource length: {}", resources.size()));
-  }
   auto cluster_load_assignment =
       MessageUtil::anyConvert<envoy::api::v2::ClusterLoadAssignment>(resources[0]);
   MessageUtil::validate(cluster_load_assignment);
@@ -135,6 +129,29 @@ void EdsClusterImpl::onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt
   priority_set_.batchHostUpdate(helper);
 }
 
+void EdsClusterImpl::onConfigUpdate(
+    const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& resources,
+    const Protobuf::RepeatedPtrField<std::string>&, const std::string&) {
+  validateUpdateSize(resources.size());
+  Protobuf::RepeatedPtrField<ProtobufWkt::Any> unwrapped_resource;
+  *unwrapped_resource.Add() = resources[0].resource();
+  onConfigUpdate(unwrapped_resource, resources[0].version());
+}
+
+bool EdsClusterImpl::validateUpdateSize(int num_resources) {
+  if (num_resources == 0) {
+    ENVOY_LOG(debug, "Missing ClusterLoadAssignment for {} in onConfigUpdate()", cluster_name_);
+    info_->stats().update_empty_.inc();
+    onPreInitComplete();
+    return false;
+  }
+  if (num_resources != 1) {
+    throw EnvoyException(fmt::format("Unexpected EDS resource length: {}", num_resources));
+    // (would be a return false here)
+  }
+  return true;
+}
+
 void EdsClusterImpl::onAssignmentTimeout() {
   // We can no longer use the assignments, remove them.
   // TODO(vishalpowar) This is not going to work for incremental updates, and we
diff --git a/source/common/upstream/eds.h b/source/common/upstream/eds.h
index ac2a95411b489..f1cae5638829c 100644
--- a/source/common/upstream/eds.h
+++ b/source/common/upstream/eds.h
@@ -30,13 +30,10 @@ class EdsClusterImpl : public BaseDynamicClusterImpl, Config::SubscriptionCallba
   InitializePhase initializePhase() const override { return InitializePhase::Secondary; }
 
   // Config::SubscriptionCallbacks
-  // TODO(fredlas) deduplicate
   void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
                       const std::string& version_info) override;
   void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>&,
-                      const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override {
-    NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
-  }
+                      const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override;
   void onConfigUpdateFailed(const EnvoyException* e) override;
   std::string resourceName(const ProtobufWkt::Any& resource) override {
     return MessageUtil::anyConvert<envoy::api::v2::ClusterLoadAssignment>(resource).cluster_name();
@@ -50,6 +47,7 @@ class EdsClusterImpl : public BaseDynamicClusterImpl, Config::SubscriptionCallba
                               LocalityWeightsMap& new_locality_weights_map,
                               PriorityStateManager& priority_state_manager,
                               std::unordered_map<std::string, HostSharedPtr>& updated_hosts);
+  bool validateUpdateSize(int num_resources);
 
   // ClusterImplBase
   void reloadHealthyHostsHelper(const HostSharedPtr& host) override;
diff --git a/source/server/lds_api.cc b/source/server/lds_api.cc
index 92c4ec6824278..e5b7a0a314a0a 100644
--- a/source/server/lds_api.cc
+++ b/source/server/lds_api.cc
@@ -31,56 +31,47 @@ LdsApiImpl::LdsApiImpl(const envoy::api::v2::core::ConfigSource& lds_config,
   init_manager.add(init_target_);
 }
 
-void LdsApiImpl::onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
-                                const std::string& version_info) {
+void LdsApiImpl::onConfigUpdate(
+    const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& added_resources,
+    const Protobuf::RepeatedPtrField<std::string>& removed_resources,
+    const std::string& system_version_info) {
   cm_.adsMux().pause(Config::TypeUrl::get().RouteConfiguration);
   Cleanup rds_resume([this] { cm_.adsMux().resume(Config::TypeUrl::get().RouteConfiguration); });
 
-  std::vector<envoy::api::v2::Listener> listeners;
-  for (const auto& listener_blob : resources) {
-    listeners.push_back(MessageUtil::anyConvert<envoy::api::v2::Listener>(listener_blob));
-    MessageUtil::validate(listeners.back());
-  }
-  std::vector<std::string> exception_msgs;
-  std::unordered_set<std::string> listener_names;
-  for (const auto& listener : listeners) {
-    if (!listener_names.insert(listener.name()).second) {
-      throw EnvoyException(fmt::format("duplicate listener {} found", listener.name()));
+  // We do all listener removals before adding the new listeners. This allows adding a new listener
+  // with the same address as a listener that is to be removed. Do not change the order.
+  for (const auto& removed_listener : removed_resources) {
+    if (listener_manager_.removeListener(removed_listener)) {
+      ENVOY_LOG(info, "lds: remove listener '{}'", removed_listener);
     }
   }
-  // We need to keep track of which listeners we might need to remove.
-  std::unordered_map<std::string, std::reference_wrapper<Network::ListenerConfig>>
-      listeners_to_remove;
 
-  // We build the list of listeners to be removed and remove them before
-  // adding new listeners. This allows adding a new listener with the same
-  // address as a listener that is to be removed. Do not change the order.
-  for (const auto& listener : listener_manager_.listeners()) {
-    listeners_to_remove.emplace(listener.get().name(), listener);
-  }
-  for (const auto& listener : listeners) {
-    listeners_to_remove.erase(listener.name());
-  }
-  for (const auto& listener : listeners_to_remove) {
-    if (listener_manager_.removeListener(listener.first)) {
-      ENVOY_LOG(info, "lds: remove listener '{}'", listener.first);
-    }
-  }
-
-  for (const auto& listener : listeners) {
-    const std::string& listener_name = listener.name();
+  std::vector<std::string> exception_msgs;
+  std::unordered_set<std::string> listener_names;
+  bool any_applied = false;
+  for (const auto& resource : added_resources) {
+    envoy::api::v2::Listener listener;
     try {
-      if (listener_manager_.addOrUpdateListener(listener, version_info, true)) {
-        ENVOY_LOG(info, "lds: add/update listener '{}'", listener_name);
+      listener = MessageUtil::anyConvert<envoy::api::v2::Listener>(resource.resource());
+      MessageUtil::validate(listener);
+      if (!listener_names.insert(listener.name()).second) {
+        // NOTE: at this point, the first of these duplicates has already been successfully applied.
+        throw EnvoyException(fmt::format("duplicate listener {} found", listener.name()));
+      }
+      if (listener_manager_.addOrUpdateListener(listener, resource.version(), true)) {
+        ENVOY_LOG(info, "lds: add/update listener '{}'", listener.name());
+        any_applied = true;
       } else {
-        ENVOY_LOG(debug, "lds: add/update listener '{}' skipped", listener_name);
+        ENVOY_LOG(debug, "lds: add/update listener '{}' skipped", listener.name());
       }
     } catch (const EnvoyException& e) {
-      exception_msgs.push_back(fmt::format("{}: {}", listener_name, e.what()));
+      exception_msgs.push_back(fmt::format("{}: {}", listener.name(), e.what()));
     }
   }
 
-  version_info_ = version_info;
+  if (any_applied) {
+    system_version_info_ = system_version_info;
+  }
   init_target_.ready();
   if (!exception_msgs.empty()) {
     throw EnvoyException(fmt::format("Error adding/updating listener(s) {}",
@@ -88,6 +79,36 @@ void LdsApiImpl::onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::An
   }
 }
 
+void LdsApiImpl::onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
+                                const std::string& version_info) {
+  // We need to keep track of which listeners need to remove.
+  // Specifically, it's [listeners we currently have] - [listeners found in the response].
+  std::unordered_set<std::string> listeners_to_remove;
+  for (const auto& listener : listener_manager_.listeners()) {
+    listeners_to_remove.insert(listener.get().name());
+  }
+
+  Protobuf::RepeatedPtrField<envoy::api::v2::Resource> to_add_repeated;
+  for (const auto& listener_blob : resources) {
+    // Add this resource to our delta added/updated pile...
+    envoy::api::v2::Resource* to_add = to_add_repeated.Add();
+    const std::string listener_name =
+        MessageUtil::anyConvert<envoy::api::v2::Listener>(listener_blob).name();
+    to_add->set_name(listener_name);
+    to_add->set_version(version_info);
+    to_add->mutable_resource()->MergeFrom(listener_blob);
+    // ...and remove its name from our delta removed pile.
+    listeners_to_remove.erase(listener_name);
+  }
+
+  // Copy our delta removed pile into the desired format.
+  Protobuf::RepeatedPtrField<std::string> to_remove_repeated;
+  for (const auto& listener : listeners_to_remove) {
+    *to_remove_repeated.Add() = listener;
+  }
+  onConfigUpdate(to_add_repeated, to_remove_repeated, version_info);
+}
+
 void LdsApiImpl::onConfigUpdateFailed(const EnvoyException*) {
   // We need to allow server startup to continue, even if we have a bad
   // config.
diff --git a/source/server/lds_api.h b/source/server/lds_api.h
index 859d26a641b71..75af148f18ae5 100644
--- a/source/server/lds_api.h
+++ b/source/server/lds_api.h
@@ -28,16 +28,14 @@ class LdsApiImpl : public LdsApi,
              Stats::Scope& scope, ListenerManager& lm, Api::Api& api);
 
   // Server::LdsApi
-  std::string versionInfo() const override { return version_info_; }
+  std::string versionInfo() const override { return system_version_info_; }
 
   // Config::SubscriptionCallbacks
-  // TODO(fredlas) deduplicate
   void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
                       const std::string& version_info) override;
-  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>&,
-                      const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override {
-    NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
-  }
+  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& added_resources,
+                      const Protobuf::RepeatedPtrField<std::string>& removed_resources,
+                      const std::string& system_version_info) override;
   void onConfigUpdateFailed(const EnvoyException* e) override;
   std::string resourceName(const ProtobufWkt::Any& resource) override {
     return MessageUtil::anyConvert<envoy::api::v2::Listener>(resource).name();
@@ -45,7 +43,7 @@ class LdsApiImpl : public LdsApi,
 
 private:
   std::unique_ptr<Config::Subscription> subscription_;
-  std::string version_info_;
+  std::string system_version_info_;
   ListenerManager& listener_manager_;
   Stats::ScopePtr scope_;
   Upstream::ClusterManager& cm_;
diff --git a/test/common/secret/sds_api_test.cc b/test/common/secret/sds_api_test.cc
index 163ad7b549f52..06500ce108100 100644
--- a/test/common/secret/sds_api_test.cc
+++ b/test/common/secret/sds_api_test.cc
@@ -115,6 +115,100 @@ TEST_F(SdsApiTest, DynamicTlsCertificateUpdateSuccess) {
   handle->remove();
 }
 
+class PartialMockSds : public SdsApi {
+public:
+  PartialMockSds(NiceMock<Server::MockInstance>& server, Api::Api& api,
+                 NiceMock<Init::MockManager>& init_manager,
+                 envoy::api::v2::core::ConfigSource& config_source)
+      : SdsApi(
+            server.localInfo(), server.dispatcher(), server.random(), server.stats(),
+            server.clusterManager(), init_manager, config_source, "abc.com", []() {}, api) {}
+
+  MOCK_METHOD2(onConfigUpdate,
+               void(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>&, const std::string&));
+  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& added,
+                      const Protobuf::RepeatedPtrField<std::string>& removed,
+                      const std::string& version) override {
+    SdsApi::onConfigUpdate(added, removed, version);
+  }
+  void setSecret(const envoy::api::v2::auth::Secret&) override {}
+  void validateConfig(const envoy::api::v2::auth::Secret&) override {}
+};
+
+// Basic test of delta's passthrough call to the state-of-the-world variant, to
+// increase coverage.
+TEST_F(SdsApiTest, Delta) {
+  Protobuf::RepeatedPtrField<envoy::api::v2::Resource> resources;
+  envoy::api::v2::auth::Secret secret;
+  secret.set_name("secret_1");
+  auto* resource = resources.Add();
+  resource->mutable_resource()->PackFrom(secret);
+  resource->set_name("secret_1");
+  resource->set_version("version1");
+
+  Protobuf::RepeatedPtrField<ProtobufWkt::Any> for_matching;
+  for_matching.Add()->PackFrom(secret);
+
+  NiceMock<Server::MockInstance> server;
+  NiceMock<Init::MockManager> init_manager;
+  envoy::api::v2::core::ConfigSource config_source;
+  PartialMockSds sds(server, *api_, init_manager, config_source);
+  EXPECT_CALL(sds, onConfigUpdate(RepeatedProtoEq(for_matching), "version1"));
+  sds.SdsApi::onConfigUpdate(resources, {}, "ignored");
+
+  // An attempt to remove a resource logs an error, but otherwise just carries on (ignoring the
+  // removal attempt).
+  resource->set_version("version2");
+  EXPECT_CALL(sds, onConfigUpdate(RepeatedProtoEq(for_matching), "version2"));
+  Protobuf::RepeatedPtrField<std::string> removals;
+  *removals.Add() = "route_0";
+  sds.SdsApi::onConfigUpdate(resources, removals, "ignored");
+}
+
+// Tests SDS's use of the delta variant of onConfigUpdate().
+TEST_F(SdsApiTest, DeltaUpdateSuccess) {
+  NiceMock<Server::MockInstance> server;
+  NiceMock<Init::MockManager> init_manager;
+  envoy::api::v2::core::ConfigSource config_source;
+  TlsCertificateSdsApi sds_api(
+      server.localInfo(), server.dispatcher(), server.random(), server.stats(),
+      server.clusterManager(), init_manager, config_source, "abc.com", []() {}, *api_);
+
+  NiceMock<Secret::MockSecretCallbacks> secret_callback;
+  auto handle =
+      sds_api.addUpdateCallback([&secret_callback]() { secret_callback.onAddOrUpdateSecret(); });
+
+  std::string yaml =
+      R"EOF(
+  name: "abc.com"
+  tls_certificate:
+    certificate_chain:
+      filename: "{{ test_rundir }}/test/extensions/transport_sockets/tls/test_data/selfsigned_cert.pem"
+    private_key:
+      filename: "{{ test_rundir }}/test/extensions/transport_sockets/tls/test_data/selfsigned_key.pem"
+    )EOF";
+  envoy::api::v2::auth::Secret typed_secret;
+  MessageUtil::loadFromYaml(TestEnvironment::substitute(yaml), typed_secret);
+  Protobuf::RepeatedPtrField<envoy::api::v2::Resource> secret_resources;
+  secret_resources.Add()->mutable_resource()->PackFrom(typed_secret);
+
+  EXPECT_CALL(secret_callback, onAddOrUpdateSecret());
+  sds_api.onConfigUpdate(secret_resources, {}, "");
+
+  Ssl::TlsCertificateConfigImpl tls_config(*sds_api.secret(), *api_);
+  const std::string cert_pem =
+      "{{ test_rundir }}/test/extensions/transport_sockets/tls/test_data/selfsigned_cert.pem";
+  EXPECT_EQ(TestEnvironment::readFileToStringForTest(TestEnvironment::substitute(cert_pem)),
+            tls_config.certificateChain());
+
+  const std::string key_pem =
+      "{{ test_rundir }}/test/extensions/transport_sockets/tls/test_data/selfsigned_key.pem";
+  EXPECT_EQ(TestEnvironment::readFileToStringForTest(TestEnvironment::substitute(key_pem)),
+            tls_config.privateKey());
+
+  handle->remove();
+}
+
 // Validate that CertificateValidationContextSdsApi updates secrets successfully if
 // a good secret is passed to onConfigUpdate().
 TEST_F(SdsApiTest, DynamicCertificateValidationContextUpdateSuccess) {
diff --git a/test/common/upstream/cds_api_impl_test.cc b/test/common/upstream/cds_api_impl_test.cc
index 83fbd093d8330..221643935422d 100644
--- a/test/common/upstream/cds_api_impl_test.cc
+++ b/test/common/upstream/cds_api_impl_test.cc
@@ -26,6 +26,7 @@ using testing::InSequence;
 using testing::Invoke;
 using testing::Return;
 using testing::ReturnRef;
+using testing::StrEq;
 using testing::Throw;
 
 namespace Envoy {
@@ -255,6 +256,55 @@ TEST_F(CdsApiImplTest, ConfigUpdateWith2ValidClusters) {
   dynamic_cast<CdsApiImpl*>(cds_.get())->onConfigUpdate(clusters, "");
 }
 
+TEST_F(CdsApiImplTest, DeltaConfigUpdate) {
+  {
+    InSequence s;
+    setup();
+  }
+  EXPECT_CALL(initialized_, ready());
+  EXPECT_CALL(request_, cancel());
+
+  {
+    Protobuf::RepeatedPtrField<envoy::api::v2::Resource> resources;
+    {
+      envoy::api::v2::Cluster cluster;
+      cluster.set_name("cluster_1");
+      cm_.expectAdd("cluster_1", "v1");
+      auto* resource = resources.Add();
+      resource->mutable_resource()->PackFrom(cluster);
+      resource->set_name("cluster_1");
+      resource->set_version("v1");
+    }
+    {
+      envoy::api::v2::Cluster cluster;
+      cluster.set_name("cluster_2");
+      cm_.expectAdd("cluster_2", "v1");
+      auto* resource = resources.Add();
+      resource->mutable_resource()->PackFrom(cluster);
+      resource->set_name("cluster_2");
+      resource->set_version("v1");
+    }
+    dynamic_cast<CdsApiImpl*>(cds_.get())->onConfigUpdate(resources, {}, "v1");
+  }
+
+  {
+    Protobuf::RepeatedPtrField<envoy::api::v2::Resource> resources;
+    {
+      envoy::api::v2::Cluster cluster;
+      cluster.set_name("cluster_3");
+      cm_.expectAdd("cluster_3", "v2");
+      auto* resource = resources.Add();
+      resource->mutable_resource()->PackFrom(cluster);
+      resource->set_name("cluster_3");
+      resource->set_version("v2");
+    }
+    Protobuf::RepeatedPtrField<std::string> removed;
+    *removed.Add() = "cluster_1";
+    EXPECT_CALL(cm_, removeCluster(StrEq("cluster_1"))).WillOnce(Return(true));
+    dynamic_cast<CdsApiImpl*>(cds_.get())->onConfigUpdate(resources, removed, "v2");
+  }
+}
+
 TEST_F(CdsApiImplTest, ConfigUpdateAddsSecondClusterEvenIfFirstThrows) {
   {
     InSequence s;
diff --git a/test/common/upstream/eds_test.cc b/test/common/upstream/eds_test.cc
index 5e2a9d7d4c5af..b3c8b8cbbd50f 100644
--- a/test/common/upstream/eds_test.cc
+++ b/test/common/upstream/eds_test.cc
@@ -251,6 +251,23 @@ TEST_F(EdsTest, OnConfigUpdateSuccess) {
   EXPECT_EQ(1UL, stats_.counter("cluster.name.update_no_rebuild").value());
 }
 
+// Validate that delta-style onConfigUpdate() with the expected cluster accepts config.
+TEST_F(EdsTest, DeltaOnConfigUpdateSuccess) {
+  envoy::api::v2::ClusterLoadAssignment cluster_load_assignment;
+  cluster_load_assignment.set_cluster_name("fare");
+  bool initialized = false;
+  cluster_->initialize([&initialized] { initialized = true; });
+
+  Protobuf::RepeatedPtrField<envoy::api::v2::Resource> resources;
+  auto* resource = resources.Add();
+  resource->mutable_resource()->PackFrom(cluster_load_assignment);
+  resource->set_version("v1");
+  VERBOSE_EXPECT_NO_THROW(cluster_->onConfigUpdate(resources, {}, "v1"));
+
+  EXPECT_TRUE(initialized);
+  EXPECT_EQ(1UL, stats_.counter("cluster.name.update_no_rebuild").value());
+}
+
 // Validate that onConfigUpdate() with no service name accepts config.
 TEST_F(EdsTest, NoServiceNameOnSuccessConfigUpdate) {
   resetCluster(R"EOF(
diff --git a/test/server/lds_api_test.cc b/test/server/lds_api_test.cc
index 9defbf9864c1d..731c0ec15cb55 100644
--- a/test/server/lds_api_test.cc
+++ b/test/server/lds_api_test.cc
@@ -152,8 +152,11 @@ TEST_F(LdsApiTest, ValidateFail) {
   Protobuf::RepeatedPtrField<ProtobufWkt::Any> listeners;
   envoy::api::v2::Listener listener;
   listeners.Add()->PackFrom(listener);
+  std::vector<std::reference_wrapper<Network::ListenerConfig>> existing_listeners;
+  EXPECT_CALL(listener_manager_, listeners()).WillOnce(Return(existing_listeners));
+  EXPECT_CALL(init_watcher_, ready());
 
-  EXPECT_THROW(lds_->onConfigUpdate(listeners, ""), ProtoValidationException);
+  EXPECT_THROW(lds_->onConfigUpdate(listeners, ""), EnvoyException);
   EXPECT_CALL(request_, cancel());
 }
 
@@ -255,6 +258,8 @@ TEST_F(LdsApiTest, ListenerCreationContinuesEvenAfterException) {
 }
 
 // Validate onConfigUpdate throws EnvoyException with duplicate listeners.
+// The first of the duplicates will be successfully applied, with the rest adding to
+// the exception message.
 TEST_F(LdsApiTest, ValidateDuplicateListeners) {
   InSequence s;
 
@@ -264,8 +269,14 @@ TEST_F(LdsApiTest, ValidateDuplicateListeners) {
   addListener(listeners, "duplicate_listener");
   addListener(listeners, "duplicate_listener");
 
+  std::vector<std::reference_wrapper<Network::ListenerConfig>> existing_listeners;
+  EXPECT_CALL(listener_manager_, listeners()).WillOnce(Return(existing_listeners));
+  EXPECT_CALL(listener_manager_, addOrUpdateListener(_, _, true)).WillOnce(Return(true));
+  EXPECT_CALL(init_watcher_, ready());
+
   EXPECT_THROW_WITH_MESSAGE(lds_->onConfigUpdate(listeners, ""), EnvoyException,
-                            "duplicate listener duplicate_listener found");
+                            "Error adding/updating listener(s) duplicate_listener: duplicate "
+                            "listener duplicate_listener found");
   EXPECT_CALL(request_, cancel());
 }
 
@@ -453,8 +464,7 @@ TEST_F(LdsApiTest, Failure) {
 
   setup();
 
-  // To test the case of valid JSON with invalid config, create 2 listeners with
-  // the same name.
+  // To test the case of valid JSON with invalid config, create a listener with no address.
   const std::string response_json = R"EOF(
 {
   "version_info": "1",
@@ -462,13 +472,6 @@ TEST_F(LdsApiTest, Failure) {
     {
       "@type": "type.googleapis.com/envoy.api.v2.Listener",
       "name": "listener1",
-      "address": { "socket_address": { "address": "tcp://0.0.0.1", "port_value": 0 } },
-      "filter_chains": [ { "filters": null } ]
-    },
-    {
-      "@type": "type.googleapis.com/envoy.api.v2.Listener",
-      "name": "listener1",
-      "address": { "socket_address": { "address": "tcp://0.0.0.3", "port_value": 0 } },
       "filter_chains": [ { "filters": null } ]
     }
   ]
@@ -479,6 +482,9 @@ TEST_F(LdsApiTest, Failure) {
       Http::HeaderMapPtr{new Http::TestHeaderMapImpl{{":status", "200"}}}));
   message->body() = std::make_unique<Buffer::OwnedImpl>(response_json);
 
+  std::vector<std::reference_wrapper<Network::ListenerConfig>> existing_listeners;
+  EXPECT_CALL(listener_manager_, listeners()).WillOnce(Return(existing_listeners));
+
   EXPECT_CALL(init_watcher_, ready());
   EXPECT_CALL(*interval_timer_, enableTimer(_));
   callbacks_->onSuccess(std::move(message));