Repository: grpc/grpc-go

Issue:
xds/e2e: Use the `Metadata` field from `BackendOptions` only when it is non-nil
In the `e2e` package, we define a `BackendOptions` struct with a `Metadata` field here: https://github.com/grpc/grpc-go/blob/3074bcdd8f3c52e4b5cc9527592d475780b062a8/internal/testutils/xds/e2e/clientresources.go#L655

This `Metadata` is supposed to appear as endpoint level metadata in the `ClusterLoadAssignment` resource. The problem though is that when we build the resource in `EndpointResourceWithOptions`, we unconditionally use the `Metadata` field without checking if it is non-nil. See: https://github.com/grpc/grpc-go/blob/3074bcdd8f3c52e4b5cc9527592d475780b062a8/internal/testutils/xds/e2e/clientresources.go#L715. This results in the `ClusterLoadAssignment` resource having an empty JSON metadata.

See here for where this came up: https://github.com/grpc/grpc-go/pull/8537#discussion_r2298844760

Ideally:
- We should only use the `Metadata` field from `BackendOptions` when it is non-nil
- Once we do the above, we should be able to fix existing tests from not having to expect an empty JSON metadata at the endpoint level

Discussion:
@hugehoo: @easwars hello, i would like to take this issue. 

When `b.Metadata` is nil, `&v3endpointpb.LbEndpoint` struct's Metadata field would remain nil/unset, and it seems ok because it's configured as omitempty field. 
``` go
type LbEndpoint struct {
	state         protoimpl.MessageState
	sizeCache     protoimpl.SizeCache
	unknownFields protoimpl.UnknownFields

	...

	// The endpoint metadata specifies values that may be used by the load
	// balancer to select endpoints in a cluster for a given request. The filter
	// name should be specified as “envoy.lb“. An example boolean key-value pair
	// is “canary“, providing the optional canary status of the upstream host.
	// This may be matched against in a route's
	// :ref:`RouteAction <envoy_v3_api_msg_config.route.v3.RouteAction>` metadata_match field
	// to subset the endpoints considered in cluster load balancing.
	Metadata *v3.Metadata `protobuf:"bytes,3,opt,name=metadata,proto3" json:"metadata,omitempty"`
...
}
```

Solution patch:
diff --git a/internal/envconfig/xds.go b/internal/envconfig/xds.go
index e8755155..b1f883bc 100644
--- a/internal/envconfig/xds.go
+++ b/internal/envconfig/xds.go
@@ -68,4 +68,10 @@ var (
 	// trust.  For more details, see:
 	// https://github.com/grpc/proposal/blob/master/A87-mtls-spiffe-support.md
 	XDSSPIFFEEnabled = boolFromEnv("GRPC_EXPERIMENTAL_XDS_MTLS_SPIFFE", false)
+
+	// XDSHTTPConnectEnabled is true if gRPC should parse custom Metadata
+	// configuring use of an HTTP CONNECT proxy via xDS from cluster resources.
+	// For more details, see:
+	// https://github.com/grpc/proposal/blob/master/A86-xds-http-connect.md
+	XDSHTTPConnectEnabled = boolFromEnv("GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT", false)
 )
diff --git a/internal/xds/clients/lrsclient/lrsclient.go b/internal/xds/clients/lrsclient/lrsclient.go
index 8f7e1852..3e459495 100644
--- a/internal/xds/clients/lrsclient/lrsclient.go
+++ b/internal/xds/clients/lrsclient/lrsclient.go
@@ -61,10 +61,7 @@ type LRSClient struct {
 
 // New returns a new LRS Client configured with the provided config.
 func New(config Config) (*LRSClient, error) {
-	switch {
-	case config.Node.ID == "":
-		return nil, errors.New("lrsclient: node ID in node is empty")
-	case config.TransportBuilder == nil:
+	if config.TransportBuilder == nil {
 		return nil, errors.New("lrsclient: transport builder is nil")
 	}
 
diff --git a/internal/xds/xdsclient/clientimpl.go b/internal/xds/xdsclient/clientimpl.go
index 9a73b96c..dab08620 100644
--- a/internal/xds/xdsclient/clientimpl.go
+++ b/internal/xds/xdsclient/clientimpl.go
@@ -129,7 +129,21 @@ func newClientImpl(config *bootstrap.Config, metricsRecorder estats.MetricsRecor
 	if err != nil {
 		return nil, err
 	}
-	c := &clientImpl{XDSClient: client, xdsClientConfig: gConfig, bootstrapConfig: config, target: target, refCount: 1}
+	lrsC, err := lrsclient.New(lrsclient.Config{
+		Node:             gConfig.Node,
+		TransportBuilder: gConfig.TransportBuilder,
+	})
+	if err != nil {
+		return nil, err
+	}
+	c := &clientImpl{
+		XDSClient:       client,
+		xdsClientConfig: gConfig,
+		bootstrapConfig: config,
+		target:          target,
+		refCount:        1,
+		lrsClient:       lrsC,
+	}
 	c.logger = prefixLogger(c)
 	return c, nil
 }
diff --git a/internal/xds/xdsclient/clientimpl_loadreport.go b/internal/xds/xdsclient/clientimpl_loadreport.go
index 2cfe28d4..ffd0c90b 100644
--- a/internal/xds/xdsclient/clientimpl_loadreport.go
+++ b/internal/xds/xdsclient/clientimpl_loadreport.go
@@ -32,18 +32,6 @@ import (
 //
 // It returns a lrsclient.LoadStore for the user to report loads.
 func (c *clientImpl) ReportLoad(server *bootstrap.ServerConfig) (*lrsclient.LoadStore, func(context.Context)) {
-	if c.lrsClient == nil {
-		lrsC, err := lrsclient.New(lrsclient.Config{
-			Node:             c.xdsClientConfig.Node,
-			TransportBuilder: c.xdsClientConfig.TransportBuilder,
-		})
-		if err != nil {
-			c.logger.Warningf("Failed to create an lrs client to the management server to report load: %v", server, err)
-			return nil, func(context.Context) {}
-		}
-		c.lrsClient = lrsC
-	}
-
 	load, err := c.lrsClient.ReportLoad(clients.ServerIdentifier{
 		ServerURI: server.ServerURI(),
 		Extensions: grpctransport.ServerIdentifierExtension{
diff --git a/internal/xds/xdsclient/xdsresource/metadata.go b/internal/xds/xdsclient/xdsresource/metadata.go
new file mode 100644
index 00000000..b56c309d
--- /dev/null
+++ b/internal/xds/xdsclient/xdsresource/metadata.go
@@ -0,0 +1,93 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package xdsresource
+
+import (
+	"fmt"
+	"net/netip"
+
+	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	"google.golang.org/protobuf/types/known/anypb"
+)
+
+func init() {
+	registerMetadataConverter("type.googleapis.com/envoy.config.core.v3.Address", proxyAddressConvertor{})
+}
+
+var (
+	// metdataRegistry is a map from proto type to metadataConverter.
+	metdataRegistry = make(map[string]metadataConverter)
+)
+
+// metadataConverter converts xds metadata entries in
+// Metadata.typed_filter_metadata into an internal form with the fields relevant
+// to gRPC.
+type metadataConverter interface {
+	// convert parses the Any proto into a concrete struct.
+	convert(*anypb.Any) (any, error)
+}
+
+// registerMetadataConverter registers the converter to the map keyed on a proto
+// type_url. Must be called at init time. Not thread safe.
+func registerMetadataConverter(protoType string, c metadataConverter) {
+	metdataRegistry[protoType] = c
+}
+
+// metadataConverterForType retrieves a converter based on key given.
+func metadataConverterForType(typeURL string) metadataConverter {
+	return metdataRegistry[typeURL]
+}
+
+// StructMetadataValue stores the values in a google.protobuf.Struct from
+// FilterMetadata.
+type StructMetadataValue struct {
+	// Data stores the parsed JSON representation of a google.protobuf.Struct.
+	Data map[string]any
+}
+
+// ProxyAddressMetadataValue holds the address parsed from the
+// envoy.config.core.v3.Address proto message, as specified in gRFC A86.
+type ProxyAddressMetadataValue struct {
+	// Address stores the proxy address configured (A86). It will be in the form
+	// of host:port. It has to be either IPv6 or IPv4.
+	Address string
+}
+
+// proxyAddressConvertor implements the metadataConverter interface to handle
+// the conversion of envoy.config.core.v3.Address protobuf messages into an
+// internal representation.
+type proxyAddressConvertor struct{}
+
+func (proxyAddressConvertor) convert(anyProto *anypb.Any) (any, error) {
+	addressProto := &v3corepb.Address{}
+	if err := anyProto.UnmarshalTo(addressProto); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal resource from Any proto: %v", err)
+	}
+	socketaddress := addressProto.GetSocketAddress()
+	if socketaddress == nil {
+		return nil, fmt.Errorf("no socket_address field in metadata")
+	}
+	if _, err := netip.ParseAddr(socketaddress.GetAddress()); err != nil {
+		return nil, fmt.Errorf("address field is not a valid IPv4 or IPv6 address: %q", socketaddress.GetAddress())
+	}
+	portvalue := socketaddress.GetPortValue()
+	if portvalue == 0 {
+		return nil, fmt.Errorf("port value not set in socket_address")
+	}
+	return ProxyAddressMetadataValue{Address: parseAddress(socketaddress)}, nil
+}
diff --git a/internal/xds/xdsclient/xdsresource/type_eds.go b/internal/xds/xdsclient/xdsresource/type_eds.go
index 42eea009..a8d568e1 100644
--- a/internal/xds/xdsclient/xdsresource/type_eds.go
+++ b/internal/xds/xdsclient/xdsresource/type_eds.go
@@ -53,6 +53,7 @@ type Endpoint struct {
 	HealthStatus EndpointHealthStatus
 	Weight       uint32
 	HashKey      string
+	Metadata     map[string]any
 }
 
 // Locality contains information of a locality.
@@ -61,6 +62,7 @@ type Locality struct {
 	ID        clients.Locality
 	Priority  uint32
 	Weight    uint32
+	Metadata  map[string]any
 }
 
 // EndpointsUpdate contains an EDS update.
diff --git a/internal/xds/xdsclient/xdsresource/unmarshal_eds.go b/internal/xds/xdsclient/xdsresource/unmarshal_eds.go
index 3e1e59cd..f0e73acd 100644
--- a/internal/xds/xdsclient/xdsresource/unmarshal_eds.go
+++ b/internal/xds/xdsclient/xdsresource/unmarshal_eds.go
@@ -108,11 +108,16 @@ func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint, uniqueEndpointAddrs
 			}
 			uniqueEndpointAddrs[a] = true
 		}
+		endpointMetadata, err := validateAndConstructMetadata(lbEndpoint.GetMetadata())
+		if err != nil {
+			return nil, err
+		}
 		endpoints = append(endpoints, Endpoint{
 			HealthStatus: EndpointHealthStatus(lbEndpoint.GetHealthStatus()),
 			Addresses:    addrs,
 			Weight:       weight,
 			HashKey:      hashKey(lbEndpoint),
+			Metadata:     endpointMetadata,
 		})
 	}
 	return endpoints, nil
@@ -190,11 +195,17 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment) (EndpointsUpdate,
 		if err != nil {
 			return EndpointsUpdate{}, err
 		}
+		localityMetadata, err := validateAndConstructMetadata(locality.GetMetadata())
+		if err != nil {
+			return EndpointsUpdate{}, err
+		}
+
 		ret.Localities = append(ret.Localities, Locality{
 			ID:        lid,
 			Endpoints: endpoints,
 			Weight:    weight,
 			Priority:  priority,
+			Metadata:  localityMetadata,
 		})
 	}
 	for i := 0; i < len(priorities); i++ {
@@ -204,3 +215,36 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment) (EndpointsUpdate,
 	}
 	return ret, nil
 }
+
+func validateAndConstructMetadata(metadataProto *v3corepb.Metadata) (map[string]any, error) {
+	// TODO(easwars): Find a better place for the environment variable check
+	// once A83 is implemented.
+	if !envconfig.XDSHTTPConnectEnabled || metadataProto == nil {
+		return nil, nil
+	}
+	metadata := make(map[string]any)
+	// First go through TypedFilterMetadata.
+	for key, anyProto := range metadataProto.GetTypedFilterMetadata() {
+		converter := metadataConverterForType(anyProto.GetTypeUrl())
+		// Ignore types we don't have a converter for.
+		if converter == nil {
+			continue
+		}
+		val, err := converter.convert(anyProto)
+		if err != nil {
+			// If the converter fails, nack the whole resource.
+			return nil, fmt.Errorf("metadata conversion for key %q and type %q failed: %v", key, anyProto.GetTypeUrl(), err)
+		}
+		metadata[key] = val
+	}
+
+	// Process FilterMetadata for any keys not already handled.
+	for key, structProto := range metadataProto.GetFilterMetadata() {
+		// Skip keys already added from TyperFilterMetadata.
+		if metadata[key] != nil {
+			continue
+		}
+		metadata[key] = StructMetadataValue{Data: structProto.AsMap()}
+	}
+	return metadata, nil
+}
diff --git a/stream.go b/stream.go
index d9bbd4c5..0a0af896 100644
--- a/stream.go
+++ b/stream.go
@@ -549,6 +549,8 @@ type clientStream struct {
 
 	sentLast bool // sent an end stream
 
+	receivedFirstMsg bool // set after the first message is received
+
 	methodConfig *MethodConfig
 
 	ctx context.Context // the application's context, wrapped by stats/tracing
@@ -1144,11 +1146,16 @@ func (a *csAttempt) recvMsg(m any, payInfo *payloadInfo) (err error) {
 			if statusErr := a.transportStream.Status().Err(); statusErr != nil {
 				return statusErr
 			}
+			// Received no msg and status OK for non-server streaming rpcs.
+			if !cs.desc.ServerStreams && !cs.receivedFirstMsg {
+				return status.Error(codes.Internal, "cardinality violation: received no response message from non-server-streaming RPC")
+			}
 			return io.EOF // indicates successful end of stream.
 		}
 
 		return toRPCErr(err)
 	}
+	cs.receivedFirstMsg = true
 	if a.trInfo != nil {
 		a.mu.Lock()
 		if a.trInfo.tr != nil {
@@ -1177,7 +1184,7 @@ func (a *csAttempt) recvMsg(m any, payInfo *payloadInfo) (err error) {
 	} else if err != nil {
 		return toRPCErr(err)
 	}
-	return status.Errorf(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
+	return status.Error(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
 }
 
 func (a *csAttempt) finish(err error) {
@@ -1359,6 +1366,7 @@ type addrConnStream struct {
 	transport        transport.ClientTransport
 	ctx              context.Context
 	sentLast         bool
+	receivedFirstMsg bool
 	desc             *StreamDesc
 	codec            baseCodec
 	sendCompressorV0 Compressor
@@ -1484,10 +1492,15 @@ func (as *addrConnStream) RecvMsg(m any) (err error) {
 			if statusErr := as.transportStream.Status().Err(); statusErr != nil {
 				return statusErr
 			}
+			// Received no msg and status OK for non-server streaming rpcs.
+			if !as.desc.ServerStreams && !as.receivedFirstMsg {
+				return status.Error(codes.Internal, "cardinality violation: received no response message from non-server-streaming RPC")
+			}
 			return io.EOF // indicates successful end of stream.
 		}
 		return toRPCErr(err)
 	}
+	as.receivedFirstMsg = true
 
 	if as.desc.ServerStreams {
 		// Subsequent messages should be received by subsequent RecvMsg calls.
@@ -1501,7 +1514,7 @@ func (as *addrConnStream) RecvMsg(m any) (err error) {
 	} else if err != nil {
 		return toRPCErr(err)
 	}
-	return status.Errorf(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
+	return status.Error(codes.Internal, "cardinality violation: expected <EOF> for non server-streaming RPCs, but received another message")
 }
 
 func (as *addrConnStream) finish(err error) {



Test patch:
diff --git a/internal/xds/resolver/helpers_test.go b/internal/xds/resolver/helpers_test.go
index b15d913c..243bbfd9 100644
--- a/internal/xds/resolver/helpers_test.go
+++ b/internal/xds/resolver/helpers_test.go
@@ -40,6 +40,8 @@ import (
 	"google.golang.org/grpc/serviceconfig"
 	"google.golang.org/grpc/status"
 
+	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
+	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
 	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
 	v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
@@ -60,11 +62,17 @@ const (
 	defaultTestServiceName     = "service-name"
 	defaultTestRouteConfigName = "route-config-name"
 	defaultTestClusterName     = "cluster-name"
+	defaultTestEndpointName    = "endpoint-name"
+	defaultTestHostname        = "test-host"
 )
 
-// This is the expected service config when using default listener and route
-// configuration resources from the e2e package using the above resource names.
-var wantDefaultServiceConfig = fmt.Sprintf(`{
+var defaultTestPort = []uint32{8080}
+
+// wantServiceConfig returns a JSON representation of a service config with
+// xds_cluster_manager_experimental LB policy with a child policy of
+// cds_experimental for the provided cluster name.
+func wantServiceConfig(clusterName string) string {
+	return fmt.Sprintf(`{
    "loadBalancingConfig": [{
 	 "xds_cluster_manager_experimental": {
 	   "children": {
@@ -78,7 +86,8 @@ var wantDefaultServiceConfig = fmt.Sprintf(`{
 	   }
 	 }
    }]
- }`, defaultTestClusterName, defaultTestClusterName)
+ }`, clusterName, clusterName)
+}
 
 // buildResolverForTarget builds an xDS resolver for the given target. If
 // the bootstrap contents are provided, it build the xDS resolver using them
@@ -283,6 +292,21 @@ func configureResourcesOnManagementServer(ctx context.Context, t *testing.T, mgm
 	}
 }
 
+// Updates all the listener, route, cluster and endpoint configuration resources
+// on the given management server.
+func configureAllResourcesOnManagementServer(ctx context.Context, t *testing.T, mgmtServer *e2e.ManagementServer, nodeID string, listeners []*v3listenerpb.Listener, routes []*v3routepb.RouteConfiguration, clusters []*v3clusterpb.Cluster, endpoints []*v3endpointpb.ClusterLoadAssignment) {
+	resources := e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: listeners,
+		Routes:    routes,
+		Clusters:  clusters,
+		Endpoints: endpoints,
+	}
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+}
+
 // waitForResourceNames waits for the wantNames to be pushed on to namesCh.
 // Fails the test by calling t.Fatal if the context expires before that.
 func waitForResourceNames(ctx context.Context, t *testing.T, namesCh chan []string, wantNames []string) {
diff --git a/internal/xds/resolver/watch_service_test.go b/internal/xds/resolver/watch_service_test.go
index 7df159c6..1c68eba3 100644
--- a/internal/xds/resolver/watch_service_test.go
+++ b/internal/xds/resolver/watch_service_test.go
@@ -49,28 +49,35 @@ func (s) TestServiceWatch_ListenerPointsToNewRouteConfiguration(t *testing.T) {
 	mgmtServer, lisCh, routeCfgCh, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
-	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
-	routes := []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(defaultTestRouteConfigName, defaultTestServiceName, defaultTestClusterName)}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: defaultTestServiceName,
+		NodeID:     nodeID,
+		Host:       defaultTestHostname,
+		Port:       defaultTestPort[0],
+		SecLevel:   e2e.SecurityLevelNone,
+	})
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
 
 	stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
 	// Verify initial update from the resolver.
 	waitForResourceNames(ctx, t, lisCh, []string{defaultTestServiceName})
-	waitForResourceNames(ctx, t, routeCfgCh, []string{defaultTestRouteConfigName})
-	verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	waitForResourceNames(ctx, t, routeCfgCh, []string{resources.Routes[0].Name})
+	verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 
 	// Update the listener resource to point to a new route configuration name.
 	// Leave the old route configuration resource unchanged.
 	newTestRouteConfigName := defaultTestRouteConfigName + "-new"
-	listeners = []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, newTestRouteConfigName)}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	resources.Listeners = []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, newTestRouteConfigName)}
+	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, resources.Listeners, resources.Routes)
 
 	// Verify that the new route configuration resource is requested.
 	waitForResourceNames(ctx, t, routeCfgCh, []string{newTestRouteConfigName})
 
 	// Update the old route configuration resource by adding a new route.
-	routes[0].VirtualHosts[0].Routes = append(routes[0].VirtualHosts[0].Routes, &v3routepb.Route{
+	resources.Routes[0].VirtualHosts[0].Routes = append(resources.Routes[0].VirtualHosts[0].Routes, &v3routepb.Route{
 		Match: &v3routepb.RouteMatch{
 			PathSpecifier: &v3routepb.RouteMatch_Prefix{Prefix: "/foo/bar"},
 			CaseSensitive: &wrapperspb.BoolValue{Value: false},
@@ -81,17 +88,17 @@ func (s) TestServiceWatch_ListenerPointsToNewRouteConfiguration(t *testing.T) {
 			},
 		},
 	})
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, resources.Listeners, resources.Routes)
 
 	// Wait for no update from the resolver.
 	verifyNoUpdateFromResolver(ctx, t, stateCh)
 
 	// Update the management server with the new route configuration resource.
-	routes = append(routes, e2e.DefaultRouteConfig(newTestRouteConfigName, defaultTestServiceName, defaultTestClusterName))
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	resources.Routes = append(resources.Routes, e2e.DefaultRouteConfig(newTestRouteConfigName, defaultTestServiceName, resources.Clusters[0].Name))
+	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, resources.Listeners, resources.Routes)
 
 	// Ensure update from the resolver.
-	verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 }
 
 // Tests the case where the listener resource changes to contain an inline route
@@ -106,22 +113,28 @@ func (s) TestServiceWatch_ListenerPointsToInlineRouteConfiguration(t *testing.T)
 	mgmtServer, lisCh, routeCfgCh, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
-	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
-	routes := []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(defaultTestRouteConfigName, defaultTestServiceName, defaultTestClusterName)}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
-
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: defaultTestServiceName,
+		NodeID:     nodeID,
+		Host:       defaultTestHostname,
+		Port:       defaultTestPort[0],
+		SecLevel:   e2e.SecurityLevelNone,
+	})
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
 	stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
 	// Verify initial update from the resolver.
 	waitForResourceNames(ctx, t, lisCh, []string{defaultTestServiceName})
-	waitForResourceNames(ctx, t, routeCfgCh, []string{defaultTestRouteConfigName})
-	verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	waitForResourceNames(ctx, t, routeCfgCh, []string{resources.Routes[0].Name})
+	verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 
 	// Update listener to contain an inline route configuration.
 	hcm := testutils.MarshalAny(t, &v3httppb.HttpConnectionManager{
 		RouteSpecifier: &v3httppb.HttpConnectionManager_RouteConfig{
 			RouteConfig: &v3routepb.RouteConfiguration{
-				Name: defaultTestRouteConfigName,
+				Name: resources.Routes[0].Name,
 				VirtualHosts: []*v3routepb.VirtualHost{{
 					Domains: []string{defaultTestServiceName},
 					Routes: []*v3routepb.Route{{
@@ -130,7 +143,7 @@ func (s) TestServiceWatch_ListenerPointsToInlineRouteConfiguration(t *testing.T)
 						},
 						Action: &v3routepb.Route_Route{
 							Route: &v3routepb.RouteAction{
-								ClusterSpecifier: &v3routepb.RouteAction_Cluster{Cluster: defaultTestClusterName},
+								ClusterSpecifier: &v3routepb.RouteAction_Cluster{Cluster: resources.Clusters[0].Name},
 							},
 						},
 					}},
@@ -139,7 +152,7 @@ func (s) TestServiceWatch_ListenerPointsToInlineRouteConfiguration(t *testing.T)
 		},
 		HttpFilters: []*v3httppb.HttpFilter{e2e.HTTPFilter("router", &v3routerpb.Router{})},
 	})
-	listeners = []*v3listenerpb.Listener{{
+	resources.Listeners = []*v3listenerpb.Listener{{
 		Name:        defaultTestServiceName,
 		ApiListener: &v3listenerpb.ApiListener{ApiListener: hcm},
 		FilterChains: []*v3listenerpb.FilterChain{{
@@ -150,19 +163,19 @@ func (s) TestServiceWatch_ListenerPointsToInlineRouteConfiguration(t *testing.T)
 			}},
 		}},
 	}}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, nil)
+	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, resources.Listeners, nil)
 
 	// Verify that the old route configuration is not requested anymore.
 	waitForResourceNames(ctx, t, routeCfgCh, []string{})
-	verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 
 	// Update listener back to contain a route configuration name.
-	listeners = []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	resources.Listeners = []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, resources.Routes[0].Name)}
+	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, resources.Listeners, resources.Routes)
 
 	// Verify that that route configuration resource is requested.
-	waitForResourceNames(ctx, t, routeCfgCh, []string{defaultTestRouteConfigName})
+	waitForResourceNames(ctx, t, routeCfgCh, []string{resources.Routes[0].Name})
 
 	// Verify that appropriate SC is pushed on the channel.
-	verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 }
diff --git a/internal/xds/resolver/xds_resolver_test.go b/internal/xds/resolver/xds_resolver_test.go
index 6c6bbca1..fb436556 100644
--- a/internal/xds/resolver/xds_resolver_test.go
+++ b/internal/xds/resolver/xds_resolver_test.go
@@ -54,7 +54,9 @@ import (
 	"google.golang.org/protobuf/types/known/wrapperspb"
 
 	v3xdsxdstypepb "github.com/cncf/xds/go/xds/type/v3"
+	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
 	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
 	v3routerpb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/router/v3"
@@ -350,7 +352,7 @@ func (s) TestResolverBadServiceUpdate_NACKedWithCache(t *testing.T) {
 	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
 
 	// Expect a good update from the resolver.
-	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(defaultTestClusterName))
 
 	// "Make an RPC" by invoking the config selector.
 	_, err := cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
@@ -400,6 +402,8 @@ func (s) TestResolverGoodServiceUpdate(t *testing.T) {
 	for _, tt := range []struct {
 		name              string
 		routeConfig       *v3routepb.RouteConfiguration
+		clusterConfig     []*v3clusterpb.Cluster
+		endpointConfig    []*v3endpointpb.ClusterLoadAssignment
 		wantServiceConfig string
 		wantClusters      map[string]bool
 	}{
@@ -411,7 +415,9 @@ func (s) TestResolverGoodServiceUpdate(t *testing.T) {
 				ClusterSpecifierType: e2e.RouteConfigClusterSpecifierTypeCluster,
 				ClusterName:          defaultTestClusterName,
 			}),
-			wantServiceConfig: wantDefaultServiceConfig,
+			clusterConfig:     []*v3clusterpb.Cluster{e2e.DefaultCluster(defaultTestClusterName, defaultTestEndpointName, e2e.SecurityLevelNone)},
+			endpointConfig:    []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(defaultTestEndpointName, defaultTestHostname, defaultTestPort)},
+			wantServiceConfig: wantServiceConfig(defaultTestClusterName),
 			wantClusters:      map[string]bool{fmt.Sprintf("cluster:%s", defaultTestClusterName): true},
 		},
 		{
@@ -422,6 +428,12 @@ func (s) TestResolverGoodServiceUpdate(t *testing.T) {
 				ClusterSpecifierType: e2e.RouteConfigClusterSpecifierTypeWeightedCluster,
 				WeightedClusters:     map[string]int{"cluster_1": 75, "cluster_2": 25},
 			}),
+			clusterConfig: []*v3clusterpb.Cluster{
+				e2e.DefaultCluster("cluster_1", "endpoint_1", e2e.SecurityLevelNone),
+				e2e.DefaultCluster("cluster_2", "endpoint_2", e2e.SecurityLevelNone)},
+			endpointConfig: []*v3endpointpb.ClusterLoadAssignment{
+				e2e.DefaultEndpoint("endpoint_1", defaultTestHostname, defaultTestPort),
+				e2e.DefaultEndpoint("endpoint_2", defaultTestHostname, defaultTestPort)},
 			// This update contains the cluster from the previous update as well
 			// as this update, as the previous config selector still references
 			// the old cluster when the new one is pushed.
@@ -460,7 +472,7 @@ func (s) TestResolverGoodServiceUpdate(t *testing.T) {
 			// route configuration resource, as specified by the test case.
 			listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
 			routes := []*v3routepb.RouteConfiguration{tt.routeConfig}
-			configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+			configureAllResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes, tt.clusterConfig, tt.endpointConfig)
 
 			stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
@@ -527,7 +539,9 @@ func (s) TestResolverRequestHash(t *testing.T) {
 			}},
 		}},
 	}}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	cluster := []*v3clusterpb.Cluster{e2e.DefaultCluster(defaultTestClusterName, defaultTestEndpointName, e2e.SecurityLevelNone)}
+	endpoints := []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(defaultTestEndpointName, defaultTestHostname, defaultTestPort)}
+	configureAllResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes, cluster, endpoints)
 
 	// Build the resolver and read the config selector out of it.
 	stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
@@ -564,14 +578,19 @@ func (s) TestResolverRemovedWithRPCs(t *testing.T) {
 	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
-	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
-	routes := []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(defaultTestRouteConfigName, defaultTestServiceName, defaultTestClusterName)}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: defaultTestServiceName,
+		NodeID:     nodeID,
+		Host:       defaultTestHostname,
+		Port:       defaultTestPort[0],
+		SecLevel:   e2e.SecurityLevelNone,
+	})
+	mgmtServer.Update(ctx, resources)
 
 	stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
 	// Read the update pushed by the resolver to the ClientConn.
-	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 
 	res, err := cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
 	if err != nil {
@@ -588,7 +607,7 @@ func (s) TestResolverRemovedWithRPCs(t *testing.T) {
 	// not produce an empty service config at this point. Instead it will retain
 	// the cluster to which the RPC is ongoing in the service config, but will
 	// return an erroring config selector which will fail new RPCs.
-	cs = verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	cs = verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 	_, err = cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
 	if err := verifyResolverError(err, codes.Unavailable, "has been removed", nodeID); err != nil {
 		t.Fatal(err)
@@ -628,14 +647,14 @@ waitForStateUpdate:
 		sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
 		defer sCancel()
 
-		configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+		configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, resources.Listeners, resources.Routes)
 
 		select {
 		case state = <-stateCh:
 			if err := state.ServiceConfig.Err; err != nil {
 				t.Fatalf("Received error in service config: %v", state.ServiceConfig.Err)
 			}
-			wantSCParsed := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(wantDefaultServiceConfig)
+			wantSCParsed := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(wantServiceConfig(resources.Clusters[0].Name))
 			if !internal.EqualServiceConfigForTesting(state.ServiceConfig.Config, wantSCParsed.Config) {
 				t.Fatalf("Got service config:\n%s \nWant service config:\n%s", cmp.Diff(nil, state.ServiceConfig.Config), cmp.Diff(nil, wantSCParsed.Config))
 			}
@@ -666,14 +685,18 @@ func (s) TestResolverRemovedResource(t *testing.T) {
 	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
-	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
-	routes := []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(defaultTestRouteConfigName, defaultTestServiceName, defaultTestClusterName)}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
-
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: defaultTestServiceName,
+		NodeID:     nodeID,
+		Host:       defaultTestHostname,
+		Port:       defaultTestPort[0],
+		SecLevel:   e2e.SecurityLevelNone,
+	})
+	mgmtServer.Update(ctx, resources)
 	stateCh, errCh, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
 	// Read the update pushed by the resolver to the ClientConn.
-	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 
 	// "Make an RPC" by invoking the config selector.
 	res, err := cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
@@ -693,7 +716,7 @@ func (s) TestResolverRemovedResource(t *testing.T) {
 
 	// The channel should receive the existing service config with the original
 	// cluster but with an erroring config selector.
-	cs = verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	cs = verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 
 	// "Make another RPC" by invoking the config selector.
 	_, err = cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
@@ -821,7 +844,17 @@ func (s) TestResolverMaxStreamDuration(t *testing.T) {
 			},
 		}},
 	}}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	cluster := []*v3clusterpb.Cluster{
+		e2e.DefaultCluster("A", "endpoint_A", e2e.SecurityLevelNone),
+		e2e.DefaultCluster("B", "endpoint_B", e2e.SecurityLevelNone),
+		e2e.DefaultCluster("C", "endpoint_C", e2e.SecurityLevelNone),
+	}
+	endpoints := []*v3endpointpb.ClusterLoadAssignment{
+		e2e.DefaultEndpoint("endpoint_A", defaultTestHostname, defaultTestPort),
+		e2e.DefaultEndpoint("endpoint_B", defaultTestHostname, defaultTestPort),
+		e2e.DefaultEndpoint("endpoint_C", defaultTestHostname, defaultTestPort),
+	}
+	configureAllResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes, cluster, endpoints)
 
 	// Read the update pushed by the resolver to the ClientConn.
 	cs := verifyUpdateFromResolver(ctx, t, stateCh, "")
@@ -873,33 +906,43 @@ func (s) TestResolverDelayedOnCommitted(t *testing.T) {
 	mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
 
 	// Configure resources on the management server.
-	listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
-	routes := []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(defaultTestRouteConfigName, defaultTestServiceName, defaultTestClusterName)}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: defaultTestServiceName,
+		NodeID:     nodeID,
+		Host:       defaultTestHostname,
+		Port:       defaultTestPort[0],
+		SecLevel:   e2e.SecurityLevelNone,
+	})
+	mgmtServer.Update(ctx, resources)
 
 	stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
 	// Read the update pushed by the resolver to the ClientConn.
-	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantDefaultServiceConfig)
+	cs := verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(resources.Clusters[0].Name))
 
 	// Make an RPC, but do not commit it yet.
 	resOld, err := cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
 	if err != nil {
 		t.Fatalf("cs.SelectConfig(): %v", err)
 	}
-	wantClusterName := fmt.Sprintf("cluster:%s", defaultTestClusterName)
+	wantClusterName := fmt.Sprintf("cluster:%s", resources.Clusters[0].Name)
 	if cluster := clustermanager.GetPickedClusterForTesting(resOld.Context); cluster != wantClusterName {
 		t.Fatalf("Picked cluster is %q, want %q", cluster, wantClusterName)
 	}
 
 	// Delay resOld.OnCommitted(). As long as there are pending RPCs to removed
 	// clusters, they still appear in the service config.
-
+	oldClusterName := resources.Clusters[0].Name
 	// Update the route configuration resource on the management server to
 	// return a new cluster.
 	newClusterName := "new-" + defaultTestClusterName
-	routes = []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(defaultTestRouteConfigName, defaultTestServiceName, newClusterName)}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	newEndpointName := "new-" + defaultTestEndpointName
+	resources.Routes = []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(resources.Routes[0].Name, defaultTestServiceName, newClusterName)}
+	resources.Clusters = []*v3clusterpb.Cluster{e2e.DefaultCluster(newClusterName, newEndpointName, e2e.SecurityLevelNone)}
+	resources.Endpoints = []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint(newEndpointName, defaultTestHostname, defaultTestPort)}
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
 
 	// Read the update pushed by the resolver to the ClientConn and ensure the
 	// old cluster is present in the service config. Also ensure that the newly
@@ -932,7 +975,7 @@ func (s) TestResolverDelayedOnCommitted(t *testing.T) {
 		  }
 		}
 	  ]
-}`, defaultTestClusterName, defaultTestClusterName, newClusterName, newClusterName)
+}`, oldClusterName, oldClusterName, newClusterName, newClusterName)
 	cs = verifyUpdateFromResolver(ctx, t, stateCh, wantSC)
 
 	resNew, err := cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
@@ -1050,7 +1093,15 @@ func (s) TestResolverWRR(t *testing.T) {
 		ClusterSpecifierType: e2e.RouteConfigClusterSpecifierTypeWeightedCluster,
 		WeightedClusters:     map[string]int{"A": 75, "B": 25},
 	})}
-	configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes)
+	clusters := []*v3clusterpb.Cluster{
+		e2e.DefaultCluster("A", "endpoint_A", e2e.SecurityLevelNone),
+		e2e.DefaultCluster("B", "endpoint_B", e2e.SecurityLevelNone),
+	}
+	endpoints := []*v3endpointpb.ClusterLoadAssignment{
+		e2e.DefaultEndpoint("endpoint_A", defaultTestHostname, defaultTestPort),
+		e2e.DefaultEndpoint("endpoint_B", defaultTestHostname, defaultTestPort),
+	}
+	configureAllResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, listeners, routes, clusters, endpoints)
 
 	// Read the update pushed by the resolver to the ClientConn.
 	cs := verifyUpdateFromResolver(ctx, t, stateCh, "")
@@ -1481,9 +1532,17 @@ func (s) TestXDSResolverHTTPFilters(t *testing.T) {
 			// Build an xDS resolver.
 			stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
 
+			cluster := []*v3clusterpb.Cluster{
+				e2e.DefaultCluster("A", "endpoint_A", e2e.SecurityLevelNone),
+				e2e.DefaultCluster("B", "endpoint_B", e2e.SecurityLevelNone),
+			}
+			endpoints := []*v3endpointpb.ClusterLoadAssignment{
+				e2e.DefaultEndpoint("endpoint_A", defaultTestHostname, defaultTestPort),
+				e2e.DefaultEndpoint("endpoint_B", defaultTestHostname, defaultTestPort),
+			}
 			// Update the management server with a listener resource that
 			// contains an inline route configuration.
-			configureResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, []*v3listenerpb.Listener{tc.listener}, nil)
+			configureAllResourcesOnManagementServer(ctx, t, mgmtServer, nodeID, []*v3listenerpb.Listener{tc.listener}, nil, cluster, endpoints)
 
 			// Ensure that the resolver pushes a state update to the channel.
 			cs := verifyUpdateFromResolver(ctx, t, stateCh, "")
diff --git a/internal/xds/xdsclient/tests/loadreport_test.go b/internal/xds/xdsclient/tests/loadreport_test.go
index 97ab891a..e33f3799 100644
--- a/internal/xds/xdsclient/tests/loadreport_test.go
+++ b/internal/xds/xdsclient/tests/loadreport_test.go
@@ -23,6 +23,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net"
+	"sync"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -30,17 +31,24 @@ import (
 	"github.com/google/uuid"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
 	"google.golang.org/grpc/internal/testutils/xds/fakeserver"
 	"google.golang.org/grpc/internal/xds/bootstrap"
 	"google.golang.org/grpc/internal/xds/clients"
+	"google.golang.org/grpc/internal/xds/xdsclient"
+	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/testing/protocmp"
 
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
 	v3lrspb "github.com/envoyproxy/go-control-plane/envoy/service/load_stats/v3"
+	testgrpc "google.golang.org/grpc/interop/grpc_testing"
+	testpb "google.golang.org/grpc/interop/grpc_testing"
 	"google.golang.org/protobuf/types/known/durationpb"
 )
 
@@ -437,3 +445,113 @@ func (s) TestReportLoad_StreamCreation(t *testing.T) {
 	defer sCancel3()
 	cancel3(sCtx3)
 }
+
+// TestConcurrentReportLoad verifies that the client can safely handle concurrent
+// requests to initiate load reporting streams. It launches multiple goroutines
+// that all call client.ReportLoad simultaneously.
+func (s) TestConcurrentReportLoad(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{SupportLoadReportingService: true})
+	nodeID := uuid.New().String()
+	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
+	client := createXDSClient(t, bc)
+
+	serverConfig, err := bootstrap.ServerConfigForTesting(bootstrap.ServerConfigTestingOptions{URI: mgmtServer.Address})
+	if err != nil {
+		t.Fatalf("Failed to create server config for testing: %v", err)
+	}
+
+	// Call ReportLoad() concurrently from multiple go routines.
+	var wg sync.WaitGroup
+	const numGoroutines = 10
+	wg.Add(numGoroutines)
+	for range numGoroutines {
+		go func() {
+			defer wg.Done()
+			_, cancelStore := client.ReportLoad(serverConfig)
+			defer cancelStore(ctx)
+		}()
+	}
+	wg.Wait()
+}
+
+// TestConcurrentChannels verifies that we can create multiple gRPC channels
+// concurrently with a shared XDSClient, each of which will create a new LRS
+// stream without any race.
+func (s) TestConcurrentChannels(t *testing.T) {
+	// TODO(emchandwani) : Unskip after https://github.com/grpc/grpc-go/pull/8526 gets merged.
+	t.Skip()
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true, SupportLoadReportingService: true})
+
+	nodeID := uuid.New().String()
+	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
+
+	if internal.NewXDSResolverWithPoolForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
+	}
+
+	config, err := bootstrap.NewConfigFromContents(bc)
+	if err != nil {
+		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bc), err)
+	}
+	pool := xdsclient.NewPool(config)
+
+	resolverBuilder := internal.NewXDSResolverWithPoolForTesting.(func(*xdsclient.Pool) (resolver.Builder, error))
+	xdsResolver, err := resolverBuilder(pool)
+	if err != nil {
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
+	}
+
+	server := stubserver.StartTestService(t, nil)
+	defer server.Stop()
+
+	// Configure the management server with resources that enable LRS.
+	const serviceName = "my-service-e2e-lrs-test"
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: serviceName,
+		NodeID:     nodeID,
+		Host:       "localhost",
+		Port:       testutils.ParsePort(t, server.Address),
+		SecLevel:   e2e.SecurityLevelNone,
+	})
+	resources.Clusters[0].LrsServer = &v3corepb.ConfigSource{
+		ConfigSourceSpecifier: &v3corepb.ConfigSource_Self{
+			Self: &v3corepb.SelfConfigSource{},
+		},
+	}
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	var wg sync.WaitGroup
+	const (
+		numGoroutines = 10
+		numRPCs       = 10
+	)
+	for range numGoroutines {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for range numRPCs {
+				cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(xdsResolver))
+				if err != nil {
+					t.Errorf("grpc.NewClient() failed: %v", err)
+					return
+				}
+				defer cc.Close()
+
+				testClient := testgrpc.NewTestServiceClient(cc)
+				if _, err := testClient.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+					t.Errorf("EmptyCall() failed: %v", err)
+					return
+				}
+			}
+		}()
+	}
+	wg.Wait()
+}
diff --git a/internal/xds/xdsclient/xdsresource/metadata_test.go b/internal/xds/xdsclient/xdsresource/metadata_test.go
new file mode 100644
index 00000000..28de8b51
--- /dev/null
+++ b/internal/xds/xdsclient/xdsresource/metadata_test.go
@@ -0,0 +1,199 @@
+/*
+ *
+ * Copyright 2025 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package xdsresource
+
+import (
+	"testing"
+
+	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	"github.com/google/go-cmp/cmp"
+	"google.golang.org/grpc/internal/testutils"
+)
+
+const proxyAddressTypeURL = "type.googleapis.com/envoy.config.core.v3.Address"
+
+func (s) TestProxyAddressConverterSuccess(t *testing.T) {
+	converter := metadataConverterForType(proxyAddressTypeURL)
+	if converter == nil {
+		t.Fatalf("Converter for %q not found in registry", proxyAddressTypeURL)
+	}
+	tests := []struct {
+		name string
+		addr *v3corepb.Address
+		want ProxyAddressMetadataValue
+	}{
+		{
+			name: "valid IPv4 address and port",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "192.168.1.1",
+						PortSpecifier: &v3corepb.SocketAddress_PortValue{
+							PortValue: 8080,
+						},
+					},
+				},
+			},
+			want: ProxyAddressMetadataValue{
+				Address: "192.168.1.1:8080",
+			},
+		},
+		{
+			name: "valid full IPv6 address and port",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "2001:0db8:85a3:0000:0000:8a2e:0370:7334",
+						PortSpecifier: &v3corepb.SocketAddress_PortValue{
+							PortValue: 9090,
+						},
+					},
+				},
+			},
+			want: ProxyAddressMetadataValue{
+				Address: "[2001:0db8:85a3:0000:0000:8a2e:0370:7334]:9090",
+			},
+		},
+		{
+			name: "valid shortened IPv6 address",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "2001:db8::1",
+						PortSpecifier: &v3corepb.SocketAddress_PortValue{
+							PortValue: 9090,
+						},
+					},
+				},
+			},
+			want: ProxyAddressMetadataValue{
+				Address: "[2001:db8::1]:9090",
+			},
+		},
+		{
+			name: "valid link-local IPv6 address",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "fe80::1ff:fe23:4567:890a",
+						PortSpecifier: &v3corepb.SocketAddress_PortValue{
+							PortValue: 8888,
+						},
+					},
+				},
+			},
+			want: ProxyAddressMetadataValue{
+				Address: "[fe80::1ff:fe23:4567:890a]:8888",
+			},
+		},
+		{
+			name: "valid IPv4-mapped IPv6 address",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "::ffff:192.0.2.128",
+						PortSpecifier: &v3corepb.SocketAddress_PortValue{
+							PortValue: 1234,
+						},
+					},
+				},
+			},
+			want: ProxyAddressMetadataValue{
+				Address: "[::ffff:192.0.2.128]:1234",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			anyProto := testutils.MarshalAny(t, tt.addr)
+			got, err := converter.convert(anyProto)
+			if err != nil {
+				t.Fatalf("convert() failed with error: %v", err)
+			}
+			if diff := cmp.Diff(tt.want, got, cmp.AllowUnexported(ProxyAddressMetadataValue{})); diff != "" {
+				t.Errorf("convert() returned unexpected value (-want +got):\n%s", diff)
+			}
+		})
+	}
+}
+
+func (s) TestProxyAddressConverterFailure(t *testing.T) {
+	converter := metadataConverterForType(proxyAddressTypeURL)
+	if converter == nil {
+		t.Fatalf("Converter for %q not found in registry", proxyAddressTypeURL)
+	}
+	tests := []struct {
+		name    string
+		addr    *v3corepb.Address
+		wantErr string
+	}{
+		{
+			name: "invalid address",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address: "invalid-ip",
+					},
+				},
+			},
+			wantErr: "address field is not a valid IPv4 or IPv6 address: \"invalid-ip\"",
+		},
+		{
+			name: "missing socket_address",
+			addr: &v3corepb.Address{
+				// No SocketAddress field set.
+			},
+			wantErr: "no socket_address field in metadata",
+		},
+		{
+			name: "address is not a socket address",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_EnvoyInternalAddress{
+					EnvoyInternalAddress: &v3corepb.EnvoyInternalAddress{
+						AddressNameSpecifier: &v3corepb.EnvoyInternalAddress_ServerListenerName{
+							ServerListenerName: "some-internal-listener",
+						},
+					},
+				},
+			},
+			wantErr: "no socket_address field in metadata",
+		},
+		{
+			name: "port value not set",
+			addr: &v3corepb.Address{
+				Address: &v3corepb.Address_SocketAddress{
+					SocketAddress: &v3corepb.SocketAddress{
+						Address:       "127.0.0.1",
+						PortSpecifier: nil,
+					},
+				},
+			},
+			wantErr: "port value not set in socket_address",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			anyProto := testutils.MarshalAny(t, tt.addr)
+			_, err := converter.convert(anyProto)
+			if err == nil || err.Error() != tt.wantErr {
+				t.Errorf("convert() got error = %v, wantErr = %q", err, tt.wantErr)
+			}
+		})
+	}
+}
diff --git a/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go b/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go
index 4eb088dc..bad0798f 100644
--- a/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go
+++ b/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go
@@ -41,12 +41,124 @@ import (
 )
 
 func (s) TestEDSParseRespProto(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
 	tests := []struct {
 		name    string
 		m       *v3endpointpb.ClusterLoadAssignment
 		want    EndpointsUpdate
 		wantErr bool
 	}{
+		{
+			name: "with endpoint metadata from typed",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{
+					addrWithPort: "addr1:314",
+					metadata: &v3corepb.Metadata{
+						TypedFilterMetadata: map[string]*anypb.Any{
+							"typed.key": testutils.MarshalAny(t, &v3corepb.Address{
+								Address: &v3corepb.Address_SocketAddress{
+									SocketAddress: &v3corepb.SocketAddress{
+										Address: "1.2.3.4",
+										PortSpecifier: &v3corepb.SocketAddress_PortValue{
+											PortValue: 1111,
+										}},
+								},
+							}),
+						},
+						FilterMetadata: map[string]*structpb.Struct{
+							"some.key": {Fields: map[string]*structpb.Value{
+								"field": structpb.NewStringValue("untyped-value")}},
+						},
+					},
+				}}, nil)
+				return clab0.Build()
+			}(),
+			want: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+							Metadata: map[string]any{
+								"typed.key": ProxyAddressMetadataValue{
+									Address: "1.2.3.4:1111",
+								},
+								"some.key": StructMetadataValue{
+									Data: map[string]any{"field": "untyped-value"},
+								},
+							},
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+
+		{
+			name: "with endpoint metadata from filtered",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{
+					addrWithPort: "addr1:314",
+					metadata: &v3corepb.Metadata{
+						FilterMetadata: map[string]*structpb.Struct{
+							"test-key": {},
+						},
+					},
+				}}, nil)
+				return clab0.Build()
+			}(),
+			want: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+							Metadata: map[string]any{
+								"test-key": StructMetadataValue{Data: map[string]any{}},
+							},
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+		{
+			name: "with locality metadata from filtered",
+			m: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
+					Metadata: &v3corepb.Metadata{
+						FilterMetadata: map[string]*structpb.Struct{"test-key": {}},
+					},
+				})
+				return clab0.Build()
+			}(),
+			want: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+						Metadata: map[string]any{
+							"test-key": StructMetadataValue{Data: map[string]any{}},
+						},
+					},
+				},
+			},
+		},
 		{
 			name: "missing-priority",
 			m: func() *v3endpointpb.ClusterLoadAssignment {
@@ -214,8 +326,7 @@ func (s) TestEDSParseRespProto(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			got, err := parseEDSRespProto(tt.m)
 			if (err != nil) != tt.wantErr {
-				t.Errorf("parseEDSRespProto() error = %v, wantErr %v", err, tt.wantErr)
-				return
+				t.Fatalf("parseEDSRespProto() error = %v, wantErr %v", err, tt.wantErr)
 			}
 			if d := cmp.Diff(got, tt.want, cmpopts.EquateEmpty()); d != "" {
 				t.Errorf("parseEDSRespProto() got = %v, want %v, diff: %v", got, tt.want, d)
@@ -325,8 +436,7 @@ func (s) TestEDSParseRespProtoAdditionalAddrs(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			got, err := parseEDSRespProto(tt.m)
 			if (err != nil) != tt.wantErr {
-				t.Errorf("parseEDSRespProto() error = %v, wantErr %v", err, tt.wantErr)
-				return
+				t.Fatalf("parseEDSRespProto() error = %v, wantErr %v", err, tt.wantErr)
 			}
 			if d := cmp.Diff(got, tt.want, cmpopts.EquateEmpty()); d != "" {
 				t.Errorf("parseEDSRespProto() got = %v, want %v, diff: %v", got, tt.want, d)
@@ -465,6 +575,7 @@ func (s) TestUnmarshalEndpointHashKey(t *testing.T) {
 }
 
 func (s) TestUnmarshalEndpoints(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
 	var v3EndpointsAny = testutils.MarshalAny(t, func() *v3endpointpb.ClusterLoadAssignment {
 		clab0 := newClaBuilder("test", nil)
 		clab0.addLocality("locality-1", 1, 1, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
@@ -588,6 +699,174 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 	}
 }
 
+func (s) TestValidateAndConstructMetadataWithXDSHTTPConnectEnabled(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
+	tests := []struct {
+		name          string
+		metadataProto *v3corepb.Metadata
+		want          map[string]any
+		wantErr       bool
+	}{
+		{
+			name: "typed filter metadata over filter metadata-xDSHTTPConnect-enabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"some.key": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "1.2.3.4",
+								PortSpecifier: &v3corepb.SocketAddress_PortValue{
+									PortValue: 1111,
+								},
+							},
+						},
+					}),
+				},
+				FilterMetadata: map[string]*structpb.Struct{
+					"some.key": {Fields: map[string]*structpb.Value{
+						"field": structpb.NewStringValue("untyped-value")}},
+				},
+			},
+
+			want: map[string]any{
+				"some.key": ProxyAddressMetadataValue{Address: "1.2.3.4:1111"},
+			},
+		},
+		{
+			name: "success-case-xDSHTTPConnect-enabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "1.2.3.4",
+								PortSpecifier: &v3corepb.SocketAddress_PortValue{
+									PortValue: 8080,
+								},
+							},
+						},
+					}),
+				},
+				FilterMetadata: map[string]*structpb.Struct{
+					"untyped-key": {Fields: map[string]*structpb.Value{"field": structpb.NewStringValue("value")}},
+				},
+			},
+			want: map[string]any{
+				"envoy.http11_proxy_transport_socket.proxy_address": ProxyAddressMetadataValue{Address: "1.2.3.4:8080"},
+				"untyped-key": StructMetadataValue{Data: map[string]any{"field": string("value")}},
+			},
+		},
+		{
+			name: "failure-case-converter-error-xDSHTTPConnect-enabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "invalid",
+							},
+						},
+					}),
+				},
+			},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := validateAndConstructMetadata(tt.metadataProto)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("validateAndConstructMetadata() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if diff := cmp.Diff(tt.want, got, cmp.AllowUnexported(ProxyAddressMetadataValue{})); diff != "" {
+				t.Errorf("validateAndConstructMetadata() returned unexpected diff (-want +got):\n%s", diff)
+			}
+		})
+	}
+}
+
+func (s) TestValidateAndConstructMetadataWithXDSHTTPConnectDisabled(t *testing.T) {
+	tests := []struct {
+		name          string
+		metadataProto *v3corepb.Metadata
+		want          map[string]any
+		wantErr       bool
+	}{
+		{
+			name: "typed-filter-metadata-over-filter-metadata-disabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"some.key": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "1.2.3.4",
+								PortSpecifier: &v3corepb.SocketAddress_PortValue{
+									PortValue: 1111,
+								},
+							},
+						},
+					}),
+				},
+				FilterMetadata: map[string]*structpb.Struct{
+					"some.key": {Fields: map[string]*structpb.Value{
+						"field": structpb.NewStringValue("untyped-value")}},
+				},
+			},
+			want: nil,
+		},
+		{
+			name: "success-case-metadata-disabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "1.2.3.4",
+								PortSpecifier: &v3corepb.SocketAddress_PortValue{
+									PortValue: 8080,
+								},
+							},
+						},
+					}),
+				},
+				FilterMetadata: map[string]*structpb.Struct{
+					"untyped-key": {Fields: map[string]*structpb.Value{
+						"field": structpb.NewStringValue("value")}},
+				},
+			},
+			want: nil,
+		},
+		{
+			name: "failure-case-converter-error-metadata-disabled",
+			metadataProto: &v3corepb.Metadata{
+				TypedFilterMetadata: map[string]*anypb.Any{
+					"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+						Address: &v3corepb.Address_SocketAddress{
+							SocketAddress: &v3corepb.SocketAddress{
+								Address: "invalid",
+							},
+						},
+					}),
+				},
+			},
+			want: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := validateAndConstructMetadata(tt.metadataProto)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("validateAndConstructMetadata() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if diff := cmp.Diff(tt.want, got, cmp.AllowUnexported(ProxyAddressMetadataValue{})); diff != "" {
+				t.Errorf("validateAndConstructMetadata() returned unexpected diff (-want +got):\n%s", diff)
+			}
+		})
+	}
+}
+
 // claBuilder builds a ClusterLoadAssignment, aka EDS
 // response.
 type claBuilder struct {
@@ -619,13 +898,15 @@ func newClaBuilder(clusterName string, dropPercents []uint32) *claBuilder {
 
 // addLocalityOptions contains options when adding locality to the builder.
 type addLocalityOptions struct {
-	Health []v3corepb.HealthStatus
-	Weight []uint32
+	Health   []v3corepb.HealthStatus
+	Weight   []uint32
+	Metadata *v3corepb.Metadata
 }
 
 type endpointOpts struct {
 	addrWithPort            string
 	additionalAddrWithPorts []string
+	metadata                *v3corepb.Metadata
 }
 
 func addressFromStr(addrWithPort string) *v3corepb.Address {
@@ -666,6 +947,7 @@ func (clab *claBuilder) addLocality(subzone string, weight uint32, priority uint
 					AdditionalAddresses: additionalAddrs,
 				},
 			},
+			Metadata: e.metadata,
 		}
 		if opts != nil {
 			if i < len(opts.Health) {
@@ -687,12 +969,16 @@ func (clab *claBuilder) addLocality(subzone string, weight uint32, priority uint
 		}
 	}
 
-	clab.v.Endpoints = append(clab.v.Endpoints, &v3endpointpb.LocalityLbEndpoints{
+	locality := &v3endpointpb.LocalityLbEndpoints{
 		Locality:            localityID,
 		LbEndpoints:         lbEndPoints,
 		LoadBalancingWeight: &wrapperspb.UInt32Value{Value: weight},
 		Priority:            priority,
-	})
+	}
+	if opts != nil {
+		locality.Metadata = opts.Metadata
+	}
+	clab.v.Endpoints = append(clab.v.Endpoints, locality)
 }
 
 // Build builds ClusterLoadAssignment.
diff --git a/test/end2end_test.go b/test/end2end_test.go
index b2f50399..9157c525 100644
--- a/test/end2end_test.go
+++ b/test/end2end_test.go
@@ -3590,9 +3590,6 @@ func testClientStreamingError(t *testing.T, e env) {
 // Tests that a client receives a cardinality violation error for client-streaming
 // RPCs if the server doesn't send a message before returning status OK.
 func (s) TestClientStreamingCardinalityViolation_ServerHandlerMissingSendAndClose(t *testing.T) {
-	// TODO : https://github.com/grpc/grpc-go/issues/8119 - remove `t.Skip()`
-	// after this is fixed.
-	t.Skip()
 	ss := &stubserver.StubServer{
 		StreamingInputCallF: func(_ testgrpc.TestService_StreamingInputCallServer) error {
 			// Returning status OK without sending a response message.This is a
@@ -3741,6 +3738,165 @@ func (s) TestClientStreaming_ReturnErrorAfterSendAndClose(t *testing.T) {
 	}
 }
 
+// Tests that client receives a cardinality violation error for unary
+// RPCs if the server doesn't send a message before returning status OK.
+func (s) TestUnaryRPC_ServerSendsOnlyTrailersWithOK(t *testing.T) {
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer lis.Close()
+
+	ss := grpc.UnknownServiceHandler(func(any, grpc.ServerStream) error {
+		return nil
+	})
+
+	s := grpc.NewServer(ss)
+	go s.Serve(lis)
+	defer s.Stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) failed unexpectedly: %v", lis.Addr(), err)
+	}
+	defer cc.Close()
+
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err = client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.Internal {
+		t.Errorf("stream.RecvMsg() = %v, want error %v", status.Code(err), codes.Internal)
+	}
+}
+
+// Tests the behavior for unary RPC when client calls RecvMsg() twice.
+// Second call to RecvMsg should fail with io.EOF.
+func (s) TestUnaryRPC_ClientCallRecvMsgTwice(t *testing.T) {
+	e := tcpTLSEnv
+	te := newTest(t, e)
+	defer te.tearDown()
+
+	te.startServer(&testServer{security: e.security})
+
+	cc := te.clientConn()
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	desc := &grpc.StreamDesc{
+		StreamName:    "UnaryCall",
+		ServerStreams: false,
+		ClientStreams: false,
+	}
+	stream, err := cc.NewStream(ctx, desc, "/grpc.testing.TestService/UnaryCall")
+	if err != nil {
+		t.Fatalf("cc.NewStream() failed unexpectedly: %v", err)
+	}
+
+	if err := stream.SendMsg(&testpb.SimpleRequest{}); err != nil {
+		t.Fatalf("stream.SendMsg(_) = %v, want <nil>", err)
+	}
+
+	resp := &testpb.SimpleResponse{}
+	if err := stream.RecvMsg(resp); err != nil {
+		t.Fatalf("stream.RecvMsg() = %v , want <nil>", err)
+	}
+
+	if err = stream.RecvMsg(resp); err != io.EOF {
+		t.Errorf("stream.RecvMsg() = %v, want error %v", err, io.EOF)
+	}
+}
+
+// Tests the behavior for unary RPC when server calls SendMsg() twice.
+// Client should fail with cardinality violation error.
+func (s) TestUnaryRPC_ServerCallSendMsgTwice(t *testing.T) {
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer lis.Close()
+
+	s := grpc.NewServer()
+	serviceDesc := grpc.ServiceDesc{
+		ServiceName: "grpc.testing.TestService",
+		HandlerType: (*any)(nil),
+		Methods:     []grpc.MethodDesc{},
+		Streams: []grpc.StreamDesc{
+			{
+				StreamName: "UnaryCall",
+				Handler: func(_ any, stream grpc.ServerStream) error {
+					if err := stream.RecvMsg(&testpb.Empty{}); err != nil {
+						t.Errorf("stream.RecvMsg() = %v, want <nil>", err)
+					}
+
+					if err = stream.SendMsg(&testpb.Empty{}); err != nil {
+						t.Errorf("stream.SendMsg() = %v, want <nil>", err)
+					}
+
+					if err = stream.SendMsg(&testpb.Empty{}); err != nil {
+						t.Errorf("stream.SendMsg() = %v, want <nil>", err)
+					}
+					return nil
+				},
+				ClientStreams: false,
+				ServerStreams: false,
+			},
+		},
+	}
+	s.RegisterService(&serviceDesc, &testServer{})
+	go s.Serve(lis)
+	defer s.Stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("grpc.NewClient(%q) failed unexpectedly: %v", lis.Addr(), err)
+	}
+	defer cc.Close()
+
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err = client.UnaryCall(ctx, &testpb.SimpleRequest{}); status.Code(err) != codes.Internal {
+		t.Errorf("stream.RecvMsg() = %v, want error %v", status.Code(err), codes.Internal)
+	}
+}
+
+// Tests the behavior for client-streaming RPC when client calls RecvMsg() twice.
+// Second call to RecvMsg should fail with io.EOF.
+func (s) TestClientStreaming_ClientCallRecvMsgTwice(t *testing.T) {
+	ss := stubserver.StubServer{
+		StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {
+			if err := stream.SendAndClose(&testpb.StreamingInputCallResponse{}); err != nil {
+				t.Errorf("stream.SendAndClose(_) = %v, want <nil>", err)
+			}
+			return nil
+		},
+	}
+	if err := ss.Start(nil); err != nil {
+		t.Fatal("Error starting server:", err)
+	}
+	defer ss.Stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	stream, err := ss.Client.StreamingInputCall(ctx)
+	if err != nil {
+		t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
+	}
+	if err := stream.Send(&testpb.StreamingInputCallRequest{}); err != nil {
+		t.Fatalf("stream.Send(_) = %v, want <nil>", err)
+	}
+	if err := stream.CloseSend(); err != nil {
+		t.Fatalf("stream.CloseSend() = %v, want <nil>", err)
+	}
+	resp := new(testpb.StreamingInputCallResponse)
+	if err := stream.RecvMsg(resp); err != nil {
+		t.Fatalf("stream.RecvMsg() = %v , want <nil>", err)
+	}
+	if err = stream.RecvMsg(resp); err != io.EOF {
+		t.Errorf("stream.RecvMsg() = %v, want error %v", err, io.EOF)
+	}
+}
+
 // Tests the behavior for server-side streaming when client calls SendMsg twice.
 // Second call to SendMsg should fail with Internal error and result in closing
 // the connection with a RST_STREAM.
@@ -3802,7 +3958,6 @@ func (s) TestServerStreaming_ClientCallSendMsgTwice(t *testing.T) {
 	<-handlerDone
 }
 
-// TODO(i/7286) : Add tests to check server-side behavior for Unary RPC.
 // Tests the behavior for unary RPC when client calls SendMsg twice. Second call
 // to SendMsg should fail with Internal error.
 func (s) TestUnaryRPC_ClientCallSendMsgTwice(t *testing.T) {
@@ -3981,7 +4136,7 @@ func (s) TestServerStreaming_ClientSendsZeroRequests(t *testing.T) {
 }
 
 // Tests that a client receives a cardinality violation error for client-streaming
-// RPCs if the server call SendMsg multiple times.
+// RPCs if the server calls SendMsg() multiple times.
 func (s) TestClientStreaming_ServerHandlerSendMsgAfterSendMsg(t *testing.T) {
 	ss := stubserver.StubServer{
 		StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {