diff --git a/docs/root/faq/api/control_plane_version_support.rst b/docs/root/faq/api/control_plane_version_support.rst
index 599ec8d7d8d8..7c6e58ffbeb3 100644
--- a/docs/root/faq/api/control_plane_version_support.rst
+++ b/docs/root/faq/api/control_plane_version_support.rst
@@ -30,6 +30,16 @@ typical rollout sequence might look like:
 4. Support for v2 is removed in the management server. The management server moves to v3 exclusively
    internally and can support newer fields.
 
+Another approach for type url version migration will be to enable the support of mixed type url 
+protected by a runtime guard *envoy.reloadable_features.enable_type_url_downgrade_and_upgrade*.
+Client can send discovery request with v2 resource type url and process discovery response with 
+v3 resource type url. Client can also send discovery request with v3 resource type url and process 
+discovery response with v2 resource type url. The upgrade and downgrade of type url is performed automatically.
+If your management server does not support both v2/v3 at the same time, you can have clients 
+with type url upgrade and downgrade feature enabled. These clients can talk to a mix of management servers
+that support either v2 or v3 exclusively. Just like the first approach, no deprecated v2 fields or new v3 fields 
+can be used at this point.
+
 If you are operating a managed control plane as-a-service, you will likely need to support a wide
 range of client versions. In this scenario, you will require long term support for multiple major
 API transport and resource versions. Strategies for managing this support are described :ref:`here
diff --git a/docs/root/version_history/current.rst b/docs/root/version_history/current.rst
index 89659d92e12e..16f3f9e3c319 100644
--- a/docs/root/version_history/current.rst
+++ b/docs/root/version_history/current.rst
@@ -127,6 +127,7 @@ New Features
 * watchdog: watchdog action extension that does cpu profiling. See ref:`Profile Action <envoy_v3_api_file_envoy/extensions/watchdog/profile_action/v3alpha/profile_action.proto>`.
 * watchdog: watchdog action extension that sends SIGABRT to the stuck thread to terminate the process. See ref:`Abort Action <envoy_v3_api_file_envoy/extensions/watchdog/abort_action/v3alpha/abort_action.proto>`.
 * xds: added :ref:`extension config discovery<envoy_v3_api_msg_config.core.v3.ExtensionConfigSource>` support for HTTP filters.
+* xds: added support for mixed v2/v3 discovery response, which enable type url downgrade and upgrade. This feature is disabled by default and is controlled by runtime guard `envoy.reloadable_features.enable_type_url_downgrade_and_upgrade`.
 * zlib: added option to use `zlib-ng <https://github.com/zlib-ng/zlib-ng>`_ as zlib library.
 
 Deprecated
diff --git a/include/envoy/config/grpc_mux.h b/include/envoy/config/grpc_mux.h
index 6c268d1076b0..43725cb30233 100644
--- a/include/envoy/config/grpc_mux.h
+++ b/include/envoy/config/grpc_mux.h
@@ -107,6 +107,9 @@ class GrpcMux {
 
   virtual void requestOnDemandUpdate(const std::string& type_url,
                                      const std::set<std::string>& for_update) PURE;
+
+  using TypeUrlMap = absl::flat_hash_map<std::string, std::string>;
+  static TypeUrlMap& typeUrlMap() { MUTABLE_CONSTRUCT_ON_FIRST_USE(TypeUrlMap, {}); }
 };
 
 using GrpcMuxPtr = std::unique_ptr<GrpcMux>;
diff --git a/source/common/config/BUILD b/source/common/config/BUILD
index d73a0f3fbeaf..61d0bf4b4445 100644
--- a/source/common/config/BUILD
+++ b/source/common/config/BUILD
@@ -14,6 +14,7 @@ envoy_cc_library(
     hdrs = ["api_type_oracle.h"],
     deps = [
         "//source/common/protobuf",
+        "//source/common/protobuf:type_util_lib",
         "@com_github_cncf_udpa//udpa/annotations:pkg_cc_proto",
     ],
 )
diff --git a/source/common/config/api_type_oracle.cc b/source/common/config/api_type_oracle.cc
index 161ecf058610..f6feba09828a 100644
--- a/source/common/config/api_type_oracle.cc
+++ b/source/common/config/api_type_oracle.cc
@@ -31,5 +31,15 @@ ApiTypeOracle::getEarlierVersionMessageTypeName(const std::string& message_type)
   }
   return absl::nullopt;
 }
+
+const absl::optional<std::string> ApiTypeOracle::getEarlierTypeUrl(const std::string& type_url) {
+  const std::string type{TypeUtil::typeUrlToDescriptorFullName(type_url)};
+  absl::optional<std::string> old_type = ApiTypeOracle::getEarlierVersionMessageTypeName(type);
+  if (old_type.has_value()) {
+    return TypeUtil::descriptorFullNameToTypeUrl(old_type.value());
+  }
+  return {};
+}
+
 } // namespace Config
 } // namespace Envoy
diff --git a/source/common/config/api_type_oracle.h b/source/common/config/api_type_oracle.h
index cd0c5971ee52..ab5e75b113d7 100644
--- a/source/common/config/api_type_oracle.h
+++ b/source/common/config/api_type_oracle.h
@@ -1,7 +1,9 @@
 #pragma once
 
 #include "common/protobuf/protobuf.h"
+#include "common/protobuf/type_util.h"
 
+#include "absl/strings/string_view.h"
 #include "absl/types/optional.h"
 
 namespace Envoy {
@@ -22,6 +24,8 @@ class ApiTypeOracle {
 
   static const absl::optional<std::string>
   getEarlierVersionMessageTypeName(const std::string& message_type);
+
+  static const absl::optional<std::string> getEarlierTypeUrl(const std::string& type_url);
 };
 
 } // namespace Config
diff --git a/source/common/config/grpc_mux_impl.cc b/source/common/config/grpc_mux_impl.cc
index 2be7cd4fddd7..b415776d6b8e 100644
--- a/source/common/config/grpc_mux_impl.cc
+++ b/source/common/config/grpc_mux_impl.cc
@@ -23,7 +23,9 @@ GrpcMuxImpl::GrpcMuxImpl(const LocalInfo::LocalInfo& local_info,
     : grpc_stream_(this, std::move(async_client), service_method, random, dispatcher, scope,
                    rate_limit_settings),
       local_info_(local_info), skip_subsequent_node_(skip_subsequent_node),
-      first_stream_request_(true), transport_api_version_(transport_api_version) {
+      first_stream_request_(true), transport_api_version_(transport_api_version),
+      enable_type_url_downgrade_and_upgrade_(Runtime::runtimeFeatureEnabled(
+          "envoy.reloadable_features.enable_type_url_downgrade_and_upgrade")) {
   Config::Utility::checkLocalInfo("ads", local_info);
 }
 
@@ -76,6 +78,9 @@ GrpcMuxWatchPtr GrpcMuxImpl::addWatch(const std::string& type_url,
     api_state_[type_url].request_.mutable_node()->MergeFrom(local_info_.node());
     api_state_[type_url].subscribed_ = true;
     subscriptions_.emplace_back(type_url);
+    if (enable_type_url_downgrade_and_upgrade_) {
+      registerVersionedTypeUrl(type_url);
+    }
   }
 
   // This will send an updated request on each subscription.
@@ -113,19 +118,41 @@ ScopedResume GrpcMuxImpl::pause(const std::vector<std::string> type_urls) {
   });
 }
 
+void GrpcMuxImpl::registerVersionedTypeUrl(const std::string& type_url) {
+  TypeUrlMap& type_url_map = typeUrlMap();
+  if (type_url_map.find(type_url) != type_url_map.end()) {
+    return;
+  }
+  // If type_url is v3, earlier_type_url will contain v2 type url.
+  const absl::optional<std::string> earlier_type_url = ApiTypeOracle::getEarlierTypeUrl(type_url);
+  // Register v2 to v3 and v3 to v2 type_url mapping in the hash map.
+  if (earlier_type_url.has_value()) {
+    type_url_map[earlier_type_url.value()] = type_url;
+    type_url_map[type_url] = earlier_type_url.value();
+  }
+}
+
 void GrpcMuxImpl::onDiscoveryResponse(
     std::unique_ptr<envoy::service::discovery::v3::DiscoveryResponse>&& message,
     ControlPlaneStats& control_plane_stats) {
-  const std::string& type_url = message->type_url();
+  std::string type_url = message->type_url();
   ENVOY_LOG(debug, "Received gRPC message for {} at version {}", type_url, message->version_info());
   if (message->has_control_plane()) {
     control_plane_stats.identifier_.set(message->control_plane().identifier());
   }
+  // If this type url is not watched(no subscriber or no watcher), try another version of type url.
+  if (enable_type_url_downgrade_and_upgrade_ && api_state_.count(type_url) == 0) {
+    registerVersionedTypeUrl(type_url);
+    TypeUrlMap& type_url_map = typeUrlMap();
+    if (type_url_map.find(type_url) != type_url_map.end()) {
+      type_url = type_url_map[type_url];
+    }
+  }
   if (api_state_.count(type_url) == 0) {
-    ENVOY_LOG(warn, "Ignoring the message for type URL {} as it has no current subscribers.",
-              type_url);
     // TODO(yuval-k): This should never happen. consider dropping the stream as this is a
     // protocol violation
+    ENVOY_LOG(warn, "Ignoring the message for type URL {} as it has no current subscribers.",
+              type_url);
     return;
   }
   if (api_state_[type_url].watches_.empty()) {
@@ -164,10 +191,10 @@ void GrpcMuxImpl::onDiscoveryResponse(
     OpaqueResourceDecoder& resource_decoder =
         api_state_[type_url].watches_.front()->resource_decoder_;
     for (const auto& resource : message->resources()) {
-      if (type_url != resource.type_url()) {
+      if (message->type_url() != resource.type_url()) {
         throw EnvoyException(
             fmt::format("{} does not match the message-wide type URL {} in DiscoveryResponse {}",
-                        resource.type_url(), type_url, message->DebugString()));
+                        resource.type_url(), message->type_url(), message->DebugString()));
       }
       resources.emplace_back(
           new DecodedResourceImpl(resource_decoder, resource, message->version_info()));
diff --git a/source/common/config/grpc_mux_impl.h b/source/common/config/grpc_mux_impl.h
index 06acf1a78ef7..1006b165c5ba 100644
--- a/source/common/config/grpc_mux_impl.h
+++ b/source/common/config/grpc_mux_impl.h
@@ -20,6 +20,7 @@
 #include "common/config/api_version.h"
 #include "common/config/grpc_stream.h"
 #include "common/config/utility.h"
+#include "common/runtime/runtime_features.h"
 
 #include "absl/container/node_hash_map.h"
 
@@ -60,6 +61,7 @@ class GrpcMuxImpl : public GrpcMux,
   // Config::GrpcStreamCallbacks
   void onStreamEstablished() override;
   void onEstablishmentFailure() override;
+  void registerVersionedTypeUrl(const std::string& type_url);
   void
   onDiscoveryResponse(std::unique_ptr<envoy::service::discovery::v3::DiscoveryResponse>&& message,
                       ControlPlaneStats& control_plane_stats) override;
@@ -147,6 +149,7 @@ class GrpcMuxImpl : public GrpcMux,
   // This string is a type URL.
   std::unique_ptr<std::queue<std::string>> request_queue_;
   const envoy::config::core::v3::ApiVersion transport_api_version_;
+  bool enable_type_url_downgrade_and_upgrade_;
 };
 
 using GrpcMuxImplPtr = std::unique_ptr<GrpcMuxImpl>;
diff --git a/source/common/config/new_grpc_mux_impl.cc b/source/common/config/new_grpc_mux_impl.cc
index 4b72f94fb8f1..0015a2689971 100644
--- a/source/common/config/new_grpc_mux_impl.cc
+++ b/source/common/config/new_grpc_mux_impl.cc
@@ -23,7 +23,9 @@ NewGrpcMuxImpl::NewGrpcMuxImpl(Grpc::RawAsyncClientPtr&& async_client,
                                const LocalInfo::LocalInfo& local_info)
     : grpc_stream_(this, std::move(async_client), service_method, random, dispatcher, scope,
                    rate_limit_settings),
-      local_info_(local_info), transport_api_version_(transport_api_version) {}
+      local_info_(local_info), transport_api_version_(transport_api_version),
+      enable_type_url_downgrade_and_upgrade_(Runtime::runtimeFeatureEnabled(
+          "envoy.reloadable_features.enable_type_url_downgrade_and_upgrade")) {}
 
 ScopedResume NewGrpcMuxImpl::pause(const std::string& type_url) {
   return pause(std::vector<std::string>{type_url});
@@ -44,12 +46,36 @@ ScopedResume NewGrpcMuxImpl::pause(const std::vector<std::string> type_urls) {
   });
 }
 
+void NewGrpcMuxImpl::registerVersionedTypeUrl(const std::string& type_url) {
+
+  TypeUrlMap& type_url_map = typeUrlMap();
+  if (type_url_map.find(type_url) != type_url_map.end()) {
+    return;
+  }
+  // If type_url is v3, earlier_type_url will contain v2 type url.
+  absl::optional<std::string> earlier_type_url = ApiTypeOracle::getEarlierTypeUrl(type_url);
+  // Register v2 to v3 and v3 to v2 type_url mapping in the hash map.
+  if (earlier_type_url.has_value()) {
+    type_url_map[earlier_type_url.value()] = type_url;
+    type_url_map[type_url] = earlier_type_url.value();
+  }
+}
+
 void NewGrpcMuxImpl::onDiscoveryResponse(
     std::unique_ptr<envoy::service::discovery::v3::DeltaDiscoveryResponse>&& message,
     ControlPlaneStats&) {
   ENVOY_LOG(debug, "Received DeltaDiscoveryResponse for {} at version {}", message->type_url(),
             message->system_version_info());
   auto sub = subscriptions_.find(message->type_url());
+  // If this type url is not watched, try another version type url.
+  if (enable_type_url_downgrade_and_upgrade_ && sub == subscriptions_.end()) {
+    const std::string& type_url = message->type_url();
+    registerVersionedTypeUrl(type_url);
+    TypeUrlMap& type_url_map = typeUrlMap();
+    if (type_url_map.find(type_url) != type_url_map.end()) {
+      sub = subscriptions_.find(type_url_map[type_url]);
+    }
+  }
   if (sub == subscriptions_.end()) {
     ENVOY_LOG(warn,
               "Dropping received DeltaDiscoveryResponse (with version {}) for non-existent "
@@ -107,6 +133,9 @@ GrpcMuxWatchPtr NewGrpcMuxImpl::addWatch(const std::string& type_url,
   auto entry = subscriptions_.find(type_url);
   if (entry == subscriptions_.end()) {
     // We don't yet have a subscription for type_url! Make one!
+    if (enable_type_url_downgrade_and_upgrade_) {
+      registerVersionedTypeUrl(type_url);
+    }
     addSubscription(type_url, use_namespace_matching);
     return addWatch(type_url, resources, callbacks, resource_decoder, use_namespace_matching);
   }
diff --git a/source/common/config/new_grpc_mux_impl.h b/source/common/config/new_grpc_mux_impl.h
index 4f549556558f..8e64f3e4399f 100644
--- a/source/common/config/new_grpc_mux_impl.h
+++ b/source/common/config/new_grpc_mux_impl.h
@@ -16,6 +16,7 @@
 #include "common/config/pausable_ack_queue.h"
 #include "common/config/watch_map.h"
 #include "common/grpc/common.h"
+#include "common/runtime/runtime_features.h"
 
 namespace Envoy {
 namespace Config {
@@ -48,6 +49,8 @@ class NewGrpcMuxImpl
   ScopedResume pause(const std::string& type_url) override;
   ScopedResume pause(const std::vector<std::string> type_urls) override;
 
+  void registerVersionedTypeUrl(const std::string& type_url);
+
   void onDiscoveryResponse(
       std::unique_ptr<envoy::service::discovery::v3::DeltaDiscoveryResponse>&& message,
       ControlPlaneStats& control_plane_stats) override;
@@ -151,6 +154,8 @@ class NewGrpcMuxImpl
   const LocalInfo::LocalInfo& local_info_;
 
   const envoy::config::core::v3::ApiVersion transport_api_version_;
+
+  const bool enable_type_url_downgrade_and_upgrade_;
 };
 
 using NewGrpcMuxImplPtr = std::unique_ptr<NewGrpcMuxImpl>;
diff --git a/source/common/config/subscription_factory_impl.cc b/source/common/config/subscription_factory_impl.cc
index f195337f2a75..467a5b04afc2 100644
--- a/source/common/config/subscription_factory_impl.cc
+++ b/source/common/config/subscription_factory_impl.cc
@@ -30,12 +30,13 @@ SubscriptionPtr SubscriptionFactoryImpl::subscriptionFromConfigSource(
   Config::Utility::checkLocalInfo(type_url, local_info_);
   std::unique_ptr<Subscription> result;
   SubscriptionStats stats = Utility::generateStats(scope);
+  auto& runtime_snapshot = runtime_.snapshot();
 
   const auto transport_api_version = config.api_config_source().transport_api_version();
   if (transport_api_version == envoy::config::core::v3::ApiVersion::V2 &&
-      runtime_.snapshot().runtimeFeatureEnabled(
+      runtime_snapshot.runtimeFeatureEnabled(
           "envoy.reloadable_features.enable_deprecated_v2_api_warning")) {
-    runtime_.snapshot().countDeprecatedFeatureUse();
+    runtime_snapshot.countDeprecatedFeatureUse();
     ENVOY_LOG(warn,
               "xDS of version v2 has been deprecated and will be removed in subsequent versions");
   }
diff --git a/source/common/protobuf/BUILD b/source/common/protobuf/BUILD
index f505161b810f..015177dcc186 100644
--- a/source/common/protobuf/BUILD
+++ b/source/common/protobuf/BUILD
@@ -61,6 +61,7 @@ envoy_cc_library(
     deps = [
         ":message_validator_lib",
         ":protobuf",
+        ":type_util_lib",
         ":well_known_lib",
         "//include/envoy/api:api_interface",
         "//include/envoy/protobuf:message_validator_interface",
@@ -78,6 +79,16 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "type_util_lib",
+    srcs = ["type_util.cc"],
+    hdrs = ["type_util.h"],
+    deps = [
+        "//source/common/protobuf",
+        "@com_github_cncf_udpa//udpa/annotations:pkg_cc_proto",
+    ],
+)
+
 envoy_cc_library(
     name = "visitor_lib",
     srcs = ["visitor.cc"],
diff --git a/source/common/protobuf/type_util.cc b/source/common/protobuf/type_util.cc
new file mode 100644
index 000000000000..03b5c1f01b34
--- /dev/null
+++ b/source/common/protobuf/type_util.cc
@@ -0,0 +1,17 @@
+#include "common/protobuf/type_util.h"
+
+namespace Envoy {
+
+absl::string_view TypeUtil::typeUrlToDescriptorFullName(absl::string_view type_url) {
+  const size_t pos = type_url.rfind('/');
+  if (pos != absl::string_view::npos) {
+    type_url = type_url.substr(pos + 1);
+  }
+  return type_url;
+}
+
+std::string TypeUtil::descriptorFullNameToTypeUrl(absl::string_view type) {
+  return "type.googleapis.com/" + std::string(type);
+}
+
+} // namespace Envoy
diff --git a/source/common/protobuf/type_util.h b/source/common/protobuf/type_util.h
new file mode 100644
index 000000000000..9bcfa0f8c2f6
--- /dev/null
+++ b/source/common/protobuf/type_util.h
@@ -0,0 +1,17 @@
+#pragma once
+
+#include "common/protobuf/protobuf.h"
+
+#include "absl/strings/string_view.h"
+#include "absl/types/optional.h"
+
+namespace Envoy {
+
+class TypeUtil {
+public:
+  static absl::string_view typeUrlToDescriptorFullName(absl::string_view type_url);
+
+  static std::string descriptorFullNameToTypeUrl(absl::string_view type);
+};
+
+} // namespace Envoy
diff --git a/source/common/protobuf/utility.cc b/source/common/protobuf/utility.cc
index a986c6732e2f..287841ce4242 100644
--- a/source/common/protobuf/utility.cc
+++ b/source/common/protobuf/utility.cc
@@ -947,12 +947,4 @@ void TimestampUtil::systemClockToTimestamp(const SystemTime system_clock_time,
           .count()));
 }
 
-absl::string_view TypeUtil::typeUrlToDescriptorFullName(absl::string_view type_url) {
-  const size_t pos = type_url.rfind('/');
-  if (pos != absl::string_view::npos) {
-    type_url = type_url.substr(pos + 1);
-  }
-  return type_url;
-}
-
 } // namespace Envoy
diff --git a/source/common/protobuf/utility.h b/source/common/protobuf/utility.h
index 3ba16b3bb910..38519e719068 100644
--- a/source/common/protobuf/utility.h
+++ b/source/common/protobuf/utility.h
@@ -136,11 +136,6 @@ class MissingFieldException : public EnvoyException {
   MissingFieldException(const std::string& field_name, const Protobuf::Message& message);
 };
 
-class TypeUtil {
-public:
-  static absl::string_view typeUrlToDescriptorFullName(absl::string_view type_url);
-};
-
 class RepeatedPtrUtil {
 public:
   static std::string join(const Protobuf::RepeatedPtrField<std::string>& source,
diff --git a/source/common/runtime/runtime_features.cc b/source/common/runtime/runtime_features.cc
index 907005680283..2bc3f46942d5 100644
--- a/source/common/runtime/runtime_features.cc
+++ b/source/common/runtime/runtime_features.cc
@@ -94,6 +94,9 @@ constexpr const char* runtime_features[] = {
 // When features are added here, there should be a tracking bug assigned to the
 // code owner to flip the default after sufficient testing.
 constexpr const char* disabled_runtime_features[] = {
+    // Allow Envoy to upgrade or downgrade version of type url, should be removed when support for
+    // v2 url is removed from codebase.
+    "envoy.reloadable_features.enable_type_url_downgrade_and_upgrade",
     // TODO(asraa) flip this feature after codec errors are handled
     "envoy.reloadable_features.new_codec_behavior",
     // TODO(alyssawilk) flip true after the release.
diff --git a/test/common/config/BUILD b/test/common/config/BUILD
index 77b7e8fa7132..f53ca9aacb69 100644
--- a/test/common/config/BUILD
+++ b/test/common/config/BUILD
@@ -132,6 +132,7 @@ envoy_cc_test(
         "//test/test_common:logging_lib",
         "//test/test_common:resources_lib",
         "//test/test_common:simulated_time_system_lib",
+        "//test/test_common:test_runtime_lib",
         "//test/test_common:utility_lib",
         "@envoy_api//envoy/api/v2:pkg_cc_proto",
         "@envoy_api//envoy/config/endpoint/v3:pkg_cc_proto",
@@ -157,6 +158,7 @@ envoy_cc_test(
         "//test/test_common:logging_lib",
         "//test/test_common:resources_lib",
         "//test/test_common:simulated_time_system_lib",
+        "//test/test_common:test_runtime_lib",
         "//test/test_common:utility_lib",
         "@envoy_api//envoy/config/endpoint/v3:pkg_cc_proto",
         "@envoy_api//envoy/service/discovery/v3:pkg_cc_proto",
diff --git a/test/common/config/api_type_oracle_test.cc b/test/common/config/api_type_oracle_test.cc
index 327d4dc32e54..a2454953c3bb 100644
--- a/test/common/config/api_type_oracle_test.cc
+++ b/test/common/config/api_type_oracle_test.cc
@@ -27,6 +27,9 @@ TEST(ApiTypeOracleTest, All) {
   EXPECT_EQ(envoy::config::filter::http::ip_tagging::v2::IPTagging::descriptor()->full_name(),
             ApiTypeOracle::getEarlierVersionMessageTypeName(v3_config.GetDescriptor()->full_name())
                 .value());
+  EXPECT_EQ("envoy.config.filter.http.ip_tagging.v2.IPTagging",
+            TypeUtil::typeUrlToDescriptorFullName(
+                "type.googleapis.com/envoy.config.filter.http.ip_tagging.v2.IPTagging"));
 }
 
 } // namespace
diff --git a/test/common/config/grpc_mux_impl_test.cc b/test/common/config/grpc_mux_impl_test.cc
index 5a8bd21840db..8c869aa44b1f 100644
--- a/test/common/config/grpc_mux_impl_test.cc
+++ b/test/common/config/grpc_mux_impl_test.cc
@@ -24,6 +24,7 @@
 #include "test/test_common/logging.h"
 #include "test/test_common/resources.h"
 #include "test/test_common/simulated_time_system.h"
+#include "test/test_common/test_runtime.h"
 #include "test/test_common/test_time.h"
 #include "test/test_common/utility.h"
 
@@ -722,6 +723,90 @@ TEST_F(GrpcMuxImplTest, BadLocalInfoEmptyNodeName) {
       "ads: node 'id' and 'cluster' are required. Set it either in 'node' config or via "
       "--service-node and --service-cluster options.");
 }
+
+// Send discovery request with v2 resource type_url, receive discovery response with v3 resource
+// type_url.
+TEST_F(GrpcMuxImplTest, WatchV2ResourceV3) {
+  TestScopedRuntime scoped_runtime;
+  Runtime::LoaderSingleton::getExisting()->mergeValues(
+      {{"envoy.reloadable_features.enable_type_url_downgrade_and_upgrade", "true"}});
+  setup();
+
+  InSequence s;
+  const std::string& v2_type_url = Config::TypeUrl::get().ClusterLoadAssignment;
+  const std::string& v3_type_url =
+      Config::getTypeUrl<envoy::config::endpoint::v3::ClusterLoadAssignment>(
+          envoy::config::core::v3::ApiVersion::V3);
+  TestUtility::TestOpaqueResourceDecoderImpl<envoy::config::endpoint::v3::ClusterLoadAssignment>
+      resource_decoder("cluster_name");
+  auto foo_sub = grpc_mux_->addWatch(v2_type_url, {}, callbacks_, resource_decoder);
+  EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(Return(&async_stream_));
+  expectSendMessage(v2_type_url, {}, "", true);
+  grpc_mux_->start();
+
+  {
+    auto response = std::make_unique<envoy::service::discovery::v3::DiscoveryResponse>();
+    response->set_type_url(v3_type_url);
+    response->set_version_info("1");
+    envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment;
+    load_assignment.set_cluster_name("x");
+    response->add_resources()->PackFrom(load_assignment);
+    EXPECT_CALL(callbacks_, onConfigUpdate(_, "1"))
+        .WillOnce(Invoke([&load_assignment](const std::vector<DecodedResourceRef>& resources,
+                                            const std::string&) {
+          EXPECT_EQ(1, resources.size());
+          const auto& expected_assignment =
+              dynamic_cast<const envoy::config::endpoint::v3::ClusterLoadAssignment&>(
+                  resources[0].get().resource());
+          EXPECT_TRUE(TestUtility::protoEqual(expected_assignment, load_assignment));
+        }));
+    expectSendMessage(v2_type_url, {}, "1");
+    grpc_mux_->grpcStreamForTest().onReceiveMessage(std::move(response));
+  }
+}
+
+// Send discovery request with v3 resource type_url, receive discovery response with v2 resource
+// type_url.
+TEST_F(GrpcMuxImplTest, WatchV3ResourceV2) {
+  TestScopedRuntime scoped_runtime;
+  Runtime::LoaderSingleton::getExisting()->mergeValues(
+      {{"envoy.reloadable_features.enable_type_url_downgrade_and_upgrade", "true"}});
+  setup();
+
+  InSequence s;
+  const std::string& v2_type_url = Config::TypeUrl::get().ClusterLoadAssignment;
+  const std::string& v3_type_url =
+      Config::getTypeUrl<envoy::config::endpoint::v3::ClusterLoadAssignment>(
+          envoy::config::core::v3::ApiVersion::V3);
+  TestUtility::TestOpaqueResourceDecoderImpl<envoy::config::endpoint::v3::ClusterLoadAssignment>
+      resource_decoder("cluster_name");
+  auto foo_sub = grpc_mux_->addWatch(v3_type_url, {}, callbacks_, resource_decoder);
+  EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(Return(&async_stream_));
+  expectSendMessage(v3_type_url, {}, "", true);
+  grpc_mux_->start();
+
+  {
+
+    auto response = std::make_unique<envoy::service::discovery::v3::DiscoveryResponse>();
+    response->set_type_url(v2_type_url);
+    response->set_version_info("1");
+    envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment;
+    load_assignment.set_cluster_name("x");
+    response->add_resources()->PackFrom(API_DOWNGRADE(load_assignment));
+    EXPECT_CALL(callbacks_, onConfigUpdate(_, "1"))
+        .WillOnce(Invoke([&load_assignment](const std::vector<DecodedResourceRef>& resources,
+                                            const std::string&) {
+          EXPECT_EQ(1, resources.size());
+          const auto& expected_assignment =
+              dynamic_cast<const envoy::config::endpoint::v3::ClusterLoadAssignment&>(
+                  resources[0].get().resource());
+          EXPECT_TRUE(TestUtility::protoEqual(expected_assignment, load_assignment));
+        }));
+    expectSendMessage(v3_type_url, {}, "1");
+    grpc_mux_->grpcStreamForTest().onReceiveMessage(std::move(response));
+  }
+}
+
 } // namespace
 } // namespace Config
 } // namespace Envoy
diff --git a/test/common/config/new_grpc_mux_impl_test.cc b/test/common/config/new_grpc_mux_impl_test.cc
index dae7c78533d9..2bcf1ecd75de 100644
--- a/test/common/config/new_grpc_mux_impl_test.cc
+++ b/test/common/config/new_grpc_mux_impl_test.cc
@@ -21,6 +21,7 @@
 #include "test/test_common/logging.h"
 #include "test/test_common/resources.h"
 #include "test/test_common/simulated_time_system.h"
+#include "test/test_common/test_runtime.h"
 #include "test/test_common/test_time.h"
 #include "test/test_common/utility.h"
 
@@ -166,14 +167,104 @@ TEST_F(NewGrpcMuxImplTest, ConfigUpdateWithNotFoundResponse) {
   response->add_resources();
   response->mutable_resources()->at(0).set_name("not-found");
   response->mutable_resources()->at(0).add_aliases("prefix/domain1.test");
+}
 
-  grpc_mux_->onDiscoveryResponse(std::move(response), control_plane_stats_);
+// Watch v2 resource type_url, receive discovery response with v3 resource type_url.
+TEST_F(NewGrpcMuxImplTest, V3ResourceResponseV2ResourceWatch) {
+  TestScopedRuntime scoped_runtime;
+  Runtime::LoaderSingleton::getExisting()->mergeValues(
+      {{"envoy.reloadable_features.enable_type_url_downgrade_and_upgrade", "true"}});
+  setup();
 
-  const auto& subscriptions = grpc_mux_->subscriptions();
-  auto sub = subscriptions.find(type_url);
+  // Watch for v2 resource type_url.
+  const std::string& v2_type_url = Config::TypeUrl::get().ClusterLoadAssignment;
+  const std::string& v3_type_url =
+      Config::getTypeUrl<envoy::config::endpoint::v3::ClusterLoadAssignment>(
+          envoy::config::core::v3::ApiVersion::V3);
+  auto watch = grpc_mux_->addWatch(v2_type_url, {}, callbacks_, resource_decoder_);
 
-  EXPECT_TRUE(sub != subscriptions.end());
-  watch->update({});
+  EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(Return(&async_stream_));
+  // Cluster is not watched, v3 resource is rejected.
+  grpc_mux_->start();
+  {
+    auto unexpected_response =
+        std::make_unique<envoy::service::discovery::v3::DeltaDiscoveryResponse>();
+    envoy::config::cluster::v3::Cluster cluster;
+    unexpected_response->set_type_url(Config::getTypeUrl<envoy::config::cluster::v3::Cluster>(
+        envoy::config::core::v3::ApiVersion::V3));
+    unexpected_response->set_system_version_info("0");
+    unexpected_response->add_resources()->mutable_resource()->PackFrom(cluster);
+    EXPECT_CALL(callbacks_, onConfigUpdate(_, _, "0")).Times(0);
+    grpc_mux_->onDiscoveryResponse(std::move(unexpected_response), control_plane_stats_);
+  }
+  // Cluster is not watched, v2 resource is rejected.
+  {
+    auto unexpected_response =
+        std::make_unique<envoy::service::discovery::v3::DeltaDiscoveryResponse>();
+    envoy::config::cluster::v3::Cluster cluster;
+    unexpected_response->set_type_url(Config::TypeUrl::get().Cluster);
+    unexpected_response->set_system_version_info("0");
+    unexpected_response->add_resources()->mutable_resource()->PackFrom(API_DOWNGRADE(cluster));
+    EXPECT_CALL(callbacks_, onConfigUpdate(_, _, "0")).Times(0);
+    grpc_mux_->onDiscoveryResponse(std::move(unexpected_response), control_plane_stats_);
+  }
+  // ClusterLoadAssignment v2 is watched, v3 resource will be accepted.
+  {
+    auto response = std::make_unique<envoy::service::discovery::v3::DeltaDiscoveryResponse>();
+    response->set_system_version_info("1");
+    envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment;
+    load_assignment.set_cluster_name("x");
+    response->add_resources()->mutable_resource()->PackFrom(load_assignment);
+    // Send response that contains resource with v3 type url.
+    response->set_type_url(v3_type_url);
+    EXPECT_CALL(callbacks_, onConfigUpdate(_, _, "1"))
+        .WillOnce(Invoke([&load_assignment](const std::vector<DecodedResourceRef>& added_resources,
+                                            const Protobuf::RepeatedPtrField<std::string>&,
+                                            const std::string&) {
+          EXPECT_EQ(1, added_resources.size());
+          EXPECT_TRUE(
+              TestUtility::protoEqual(added_resources[0].get().resource(), load_assignment));
+        }));
+    grpc_mux_->onDiscoveryResponse(std::move(response), control_plane_stats_);
+  }
+}
+
+// Watch v3 resource type_url, receive discovery response with v2 resource type_url.
+TEST_F(NewGrpcMuxImplTest, V2ResourceResponseV3ResourceWatch) {
+  TestScopedRuntime scoped_runtime;
+  Runtime::LoaderSingleton::getExisting()->mergeValues(
+      {{"envoy.reloadable_features.enable_type_url_downgrade_and_upgrade", "true"}});
+  setup();
+
+  // Watch for v3 resource type_url.
+  const std::string& v3_type_url =
+      Config::getTypeUrl<envoy::config::endpoint::v3::ClusterLoadAssignment>(
+          envoy::config::core::v3::ApiVersion::V3);
+  const std::string& v2_type_url = Config::TypeUrl::get().ClusterLoadAssignment;
+  auto watch = grpc_mux_->addWatch(v3_type_url, {}, callbacks_, resource_decoder_);
+
+  EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(Return(&async_stream_));
+
+  grpc_mux_->start();
+  // ClusterLoadAssignment v3 is watched, v2 resource will be accepted.
+  {
+    auto response = std::make_unique<envoy::service::discovery::v3::DeltaDiscoveryResponse>();
+    response->set_system_version_info("1");
+    envoy::config::endpoint::v3::ClusterLoadAssignment load_assignment;
+    load_assignment.set_cluster_name("x");
+    response->add_resources()->mutable_resource()->PackFrom(API_DOWNGRADE(load_assignment));
+    // Send response that contains resource with v3 type url.
+    response->set_type_url(v2_type_url);
+    EXPECT_CALL(callbacks_, onConfigUpdate(_, _, "1"))
+        .WillOnce(Invoke([&load_assignment](const std::vector<DecodedResourceRef>& added_resources,
+                                            const Protobuf::RepeatedPtrField<std::string>&,
+                                            const std::string&) {
+          EXPECT_EQ(1, added_resources.size());
+          EXPECT_TRUE(
+              TestUtility::protoEqual(added_resources[0].get().resource(), load_assignment));
+        }));
+    grpc_mux_->onDiscoveryResponse(std::move(response), control_plane_stats_);
+  }
 }
 
 } // namespace
diff --git a/test/common/protobuf/BUILD b/test/common/protobuf/BUILD
index bb018981c290..4aa4300922dc 100644
--- a/test/common/protobuf/BUILD
+++ b/test/common/protobuf/BUILD
@@ -46,6 +46,14 @@ envoy_cc_test(
     ],
 )
 
+envoy_cc_test(
+    name = "type_util_test",
+    srcs = ["type_util_test.cc"],
+    deps = [
+        "//source/common/protobuf:type_util_lib",
+    ],
+)
+
 envoy_cc_fuzz_test(
     name = "value_util_fuzz_test",
     srcs = ["value_util_fuzz_test.cc"],
diff --git a/test/common/protobuf/type_util_test.cc b/test/common/protobuf/type_util_test.cc
new file mode 100644
index 000000000000..d78395163637
--- /dev/null
+++ b/test/common/protobuf/type_util_test.cc
@@ -0,0 +1,18 @@
+#include "common/protobuf/type_util.h"
+
+#include "gtest/gtest.h"
+
+namespace Envoy {
+namespace Config {
+namespace {
+TEST(TypeUtilTest, TypeUrlHelperFunction) {
+  EXPECT_EQ("envoy.config.filter.http.ip_tagging.v2.IPTagging",
+            TypeUtil::typeUrlToDescriptorFullName(
+                "type.googleapis.com/envoy.config.filter.http.ip_tagging.v2.IPTagging"));
+  EXPECT_EQ(
+      "type.googleapis.com/envoy.config.filter.http.ip_tagging.v2.IPTagging",
+      TypeUtil::descriptorFullNameToTypeUrl("envoy.config.filter.http.ip_tagging.v2.IPTagging"));
+}
+} // namespace
+} // namespace Config
+} // namespace Envoy
\ No newline at end of file
diff --git a/test/common/protobuf/utility_test.cc b/test/common/protobuf/utility_test.cc
index 2132fd25e2d2..5a199800d71d 100644
--- a/test/common/protobuf/utility_test.cc
+++ b/test/common/protobuf/utility_test.cc
@@ -1772,10 +1772,4 @@ TEST(StatusCode, Strings) {
   ASSERT_EQ("OK", MessageUtil::CodeEnumToString(ProtobufUtil::error::OK));
 }
 
-TEST(TypeUtilTest, TypeUrlToDescriptorFullName) {
-  EXPECT_EQ("envoy.config.filter.http.ip_tagging.v2.IPTagging",
-            TypeUtil::typeUrlToDescriptorFullName(
-                "type.googleapis.com/envoy.config.filter.http.ip_tagging.v2.IPTagging"));
-}
-
 } // namespace Envoy
diff --git a/test/integration/ads_integration_test.cc b/test/integration/ads_integration_test.cc
index bf413b9d91d5..b49b217464bf 100644
--- a/test/integration/ads_integration_test.cc
+++ b/test/integration/ads_integration_test.cc
@@ -115,6 +115,38 @@ TEST_P(AdsIntegrationTest, Failure) {
   makeSingleRequest();
 }
 
+// Validate that xds can support a mix of v2 and v3 type url.
+TEST_P(AdsIntegrationTest, MixV2V3TypeUrlInDiscoveryResponse) {
+  config_helper_.addRuntimeOverride(
+      "envoy.reloadable_features.enable_type_url_downgrade_and_upgrade", "true");
+  initialize();
+
+  // Send initial configuration.
+  // Discovery response with v3 type url.
+  sendDiscoveryResponse<envoy::config::cluster::v3::Cluster>(
+      Config::getTypeUrl<envoy::config::cluster::v3::Cluster>(
+          envoy::config::core::v3::ApiVersion::V3),
+      {buildCluster("cluster_0")}, {buildCluster("cluster_0")}, {}, "1", false);
+  // Discovery response with v2 type url.
+  sendDiscoveryResponse<envoy::config::endpoint::v3::ClusterLoadAssignment>(
+      Config::TypeUrl::get().ClusterLoadAssignment, {buildClusterLoadAssignment("cluster_0")},
+      {buildClusterLoadAssignment("cluster_0")}, {}, "1");
+  // Discovery response with v3 type url.
+  sendDiscoveryResponse<envoy::config::listener::v3::Listener>(
+      Config::getTypeUrl<envoy::config::listener::v3::Listener>(
+          envoy::config::core::v3::ApiVersion::V3),
+      {buildListener("listener_0", "route_config_0")},
+      {buildListener("listener_0", "route_config_0")}, {}, "1", false);
+  // Discovery response with v2 type url.
+  sendDiscoveryResponse<envoy::config::route::v3::RouteConfiguration>(
+      Config::TypeUrl::get().RouteConfiguration, {buildRouteConfig("route_config_0", "cluster_0")},
+      {buildRouteConfig("route_config_0", "cluster_0")}, {}, "1");
+  test_server_->waitForCounterGe("listener_manager.listener_create_success", 1);
+
+  // Validate that we can process a request.
+  makeSingleRequest();
+}
+
 // Validate that the request with duplicate listeners is rejected.
 TEST_P(AdsIntegrationTest, DuplicateWarmingListeners) {
   initialize();