Repository: grpc/grpc-go

Issue:
Cleanup in google-c2p resolver after removing GRPC_EXPERIMENTAL_ENABLE_NEW_PICK_FIRST
Related: https://github.com/grpc/grpc-go/pull/8550

Currently, experiment GRPC_EXPERIMENTAL_ENABLE_NEW_PICK_FIRST defaults on.

After we remove this experiment completely, we can remove code paths in the `google-c2p` resolver which query the metadata server at `http://metadata.google.internal/computeMetadata/v1/instance/network-interfaces/0/ipv6s`

Hints from issue comments:
@arjan-bal: Removal of the old pickfirst policy is blocked on internal issue b/415354418.

Solution patch:
diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md
index 424c6a6e..deb08ca9 100644
--- a/.github/pull_request_template.md
+++ b/.github/pull_request_template.md
@@ -1,4 +1,4 @@
-Thank you for your PR.  Please read and follow
+Thank you for your PR. Please read and follow
 https://github.com/grpc/grpc-go/blob/master/CONTRIBUTING.md, especially the
 "Guidelines for Pull Requests" section, and then delete this text before
 entering your PR description.
diff --git a/internal/transport/http2_client.go b/internal/transport/http2_client.go
index 5467fe97..c31ed87b 100644
--- a/internal/transport/http2_client.go
+++ b/internal/transport/http2_client.go
@@ -1499,13 +1499,6 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 		case "grpc-message":
 			grpcMessage = decodeGrpcMessage(hf.Value)
 		case ":status":
-			if hf.Value == "200" {
-				httpStatusErr = ""
-				statusCode := 200
-				httpStatusCode = &statusCode
-				break
-			}
-
 			c, err := strconv.ParseInt(hf.Value, 10, 32)
 			if err != nil {
 				se := status.New(codes.Internal, fmt.Sprintf("transport: malformed http-status: %v", err))
@@ -1513,7 +1506,19 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 				return
 			}
 			statusCode := int(c)
+			if statusCode >= 100 && statusCode < 200 {
+				if endStream {
+					se := status.New(codes.Internal, fmt.Sprintf(
+						"protocol error: informational header with status code %d must not have END_STREAM set", statusCode))
+					t.closeStream(s, se.Err(), true, http2.ErrCodeProtocol, se, nil, endStream)
+				}
+				return
+			}
 			httpStatusCode = &statusCode
+			if statusCode == 200 {
+				httpStatusErr = ""
+				break
+			}
 
 			httpStatusErr = fmt.Sprintf(
 				"unexpected HTTP status code received from server: %d (%s)",
diff --git a/internal/transport/http2_server.go b/internal/transport/http2_server.go
index 9f725e15..83cee314 100644
--- a/internal/transport/http2_server.go
+++ b/internal/transport/http2_server.go
@@ -1353,10 +1353,10 @@ func (t *http2Server) closeStream(s *ServerStream, rst bool, rstCode http2.ErrCo
 	// called to interrupt the potential blocking on other goroutines.
 	s.cancel()
 
-	oldState := s.swapState(streamDone)
-	if oldState == streamDone {
-		return
-	}
+	// We can't return early even if the stream's state is "done" as the state
+	// might have been set by the `finishStream` method. Deleting the stream via
+	// `finishStream` can get blocked on flow control.
+	s.swapState(streamDone)
 	t.deleteStream(s, eosReceived)
 
 	t.controlBuf.put(&cleanupStream{
diff --git a/internal/xds/balancer/priority/ignore_resolve_now.go b/internal/xds/balancer/priority/ignore_resolve_now.go
index 792ee4b3..df9ec46c 100644
--- a/internal/xds/balancer/priority/ignore_resolve_now.go
+++ b/internal/xds/balancer/priority/ignore_resolve_now.go
@@ -29,29 +29,21 @@ import (
 // ResolveNow() method to ignore those calls if the ignoreResolveNow bit is set.
 type ignoreResolveNowClientConn struct {
 	balancer.ClientConn
-	ignoreResolveNow *uint32
+	ignoreResolveNow atomic.Bool
 }
 
 func newIgnoreResolveNowClientConn(cc balancer.ClientConn, ignore bool) *ignoreResolveNowClientConn {
-	ret := &ignoreResolveNowClientConn{
-		ClientConn:       cc,
-		ignoreResolveNow: new(uint32),
-	}
+	ret := &ignoreResolveNowClientConn{ClientConn: cc}
 	ret.updateIgnoreResolveNow(ignore)
 	return ret
 }
 
 func (i *ignoreResolveNowClientConn) updateIgnoreResolveNow(b bool) {
-	if b {
-		atomic.StoreUint32(i.ignoreResolveNow, 1)
-		return
-	}
-	atomic.StoreUint32(i.ignoreResolveNow, 0)
-
+	i.ignoreResolveNow.Store(b)
 }
 
-func (i ignoreResolveNowClientConn) ResolveNow(o resolver.ResolveNowOptions) {
-	if atomic.LoadUint32(i.ignoreResolveNow) != 0 {
+func (i *ignoreResolveNowClientConn) ResolveNow(o resolver.ResolveNowOptions) {
+	if i.ignoreResolveNow.Load() {
 		return
 	}
 	i.ClientConn.ResolveNow(o)
diff --git a/internal/xds/xdsclient/xdsresource/unmarshal_eds.go b/internal/xds/xdsclient/xdsresource/unmarshal_eds.go
index f0e73acd..c5b9723a 100644
--- a/internal/xds/xdsclient/xdsresource/unmarshal_eds.go
+++ b/internal/xds/xdsclient/xdsresource/unmarshal_eds.go
@@ -108,35 +108,42 @@ func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint, uniqueEndpointAddrs
 			}
 			uniqueEndpointAddrs[a] = true
 		}
-		endpointMetadata, err := validateAndConstructMetadata(lbEndpoint.GetMetadata())
-		if err != nil {
-			return nil, err
+
+		var endpointMetadata map[string]any
+		var hashKey string
+		if envconfig.XDSHTTPConnectEnabled || !envconfig.XDSEndpointHashKeyBackwardCompat {
+			var err error
+			endpointMetadata, err = validateAndConstructMetadata(lbEndpoint.GetMetadata())
+			if err != nil {
+				return nil, err
+			}
+
+			// "The xDS resolver, described in A74, will be changed to set the hash_key
+			// endpoint attribute to the value of LbEndpoint.Metadata envoy.lb hash_key
+			// field, as described in Envoy's documentation for the ring hash load
+			// balancer." - A76
+			if !envconfig.XDSEndpointHashKeyBackwardCompat {
+				hashKey = hashKeyFromMetadata(endpointMetadata)
+			}
 		}
 		endpoints = append(endpoints, Endpoint{
 			HealthStatus: EndpointHealthStatus(lbEndpoint.GetHealthStatus()),
 			Addresses:    addrs,
 			Weight:       weight,
-			HashKey:      hashKey(lbEndpoint),
+			HashKey:      hashKey,
 			Metadata:     endpointMetadata,
 		})
 	}
 	return endpoints, nil
 }
 
-// hashKey extracts and returns the hash key from the given LbEndpoint. If no
-// hash key is found, it returns an empty string.
-func hashKey(lbEndpoint *v3endpointpb.LbEndpoint) string {
-	// "The xDS resolver, described in A74, will be changed to set the hash_key
-	// endpoint attribute to the value of LbEndpoint.Metadata envoy.lb hash_key
-	// field, as described in Envoy's documentation for the ring hash load
-	// balancer." - A76
-	if envconfig.XDSEndpointHashKeyBackwardCompat {
-		return ""
-	}
-	envoyLB := lbEndpoint.GetMetadata().GetFilterMetadata()["envoy.lb"]
-	if envoyLB != nil {
-		if h := envoyLB.GetFields()["hash_key"]; h != nil {
-			return h.GetStringValue()
+// hashKey extracts and returns the hash key from the given endpoint metadata.
+// If no hash key is found, it returns an empty string.
+func hashKeyFromMetadata(metadata map[string]any) string {
+	envoyLB, ok := metadata["envoy.lb"].(StructMetadataValue)
+	if ok {
+		if h, ok := envoyLB.Data["hash_key"].(string); ok {
+			return h
 		}
 	}
 	return ""
@@ -195,9 +202,13 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment) (EndpointsUpdate,
 		if err != nil {
 			return EndpointsUpdate{}, err
 		}
-		localityMetadata, err := validateAndConstructMetadata(locality.GetMetadata())
-		if err != nil {
-			return EndpointsUpdate{}, err
+		var localityMetadata map[string]any
+		if envconfig.XDSHTTPConnectEnabled {
+			var err error
+			localityMetadata, err = validateAndConstructMetadata(locality.GetMetadata())
+			if err != nil {
+				return EndpointsUpdate{}, err
+			}
 		}
 
 		ret.Localities = append(ret.Localities, Locality{
@@ -217,9 +228,7 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment) (EndpointsUpdate,
 }
 
 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 {
+	if metadataProto == nil {
 		return nil, nil
 	}
 	metadata := make(map[string]any)
diff --git a/xds/googledirectpath/googlec2p.go b/xds/googledirectpath/googlec2p.go
index f1a02649..9ef59f1a 100644
--- a/xds/googledirectpath/googlec2p.go
+++ b/xds/googledirectpath/googlec2p.go
@@ -182,6 +182,13 @@ func newNodeConfig(zone string, ipv6Capable bool) map[string]any {
 		"id":       fmt.Sprintf("C2P-%d", randInt()),
 		"locality": map[string]any{"zone": zone},
 	}
+	if envconfig.NewPickFirstEnabled {
+		// Enable dualstack endpoints in TD.
+		// TODO(https://github.com/grpc/grpc-go/issues/8561): remove IPv6 metadata server queries entirely after old pick first is removed.
+		ipv6Capable = true
+	} else {
+		logger.Infof("GRPC_EXPERIMENTAL_ENABLE_NEW_PICK_FIRST is disabled, setting ipv6Capable node metadata based on metadata server query")
+	}
 	if ipv6Capable {
 		node["metadata"] = map[string]any{ipv6CapableMetadataName: true}
 	}



Test patch:
diff --git a/internal/transport/transport_test.go b/internal/transport/transport_test.go
index b8f97c3c..d0c8a88d 100644
--- a/internal/transport/transport_test.go
+++ b/internal/transport/transport_test.go
@@ -3125,3 +3125,82 @@ func (s) TestServerSendsRSTAfterDeadlineToMisbehavedClient(t *testing.T) {
 		t.Fatalf("RST frame received earlier than expected by duration: %v", want-got)
 	}
 }
+
+// TestClientTransport_Handle1xxHeaders validates that 1xx HTTP status headers
+// are ignored and treated as a protocol error if END_STREAM is set.
+func (s) TestClientTransport_Handle1xxHeaders(t *testing.T) {
+	testStream := func() *ClientStream {
+		return &ClientStream{
+			Stream: &Stream{
+				buf: &recvBuffer{
+					c:  make(chan recvMsg),
+					mu: sync.Mutex{},
+				},
+			},
+			done:       make(chan struct{}),
+			headerChan: make(chan struct{}),
+		}
+	}
+
+	testClient := func(ts *ClientStream) *http2Client {
+		return &http2Client{
+			mu: sync.Mutex{},
+			activeStreams: map[uint32]*ClientStream{
+				0: ts,
+			},
+			controlBuf: newControlBuffer(make(<-chan struct{})),
+		}
+	}
+
+	for _, test := range []struct {
+		name            string
+		metaHeaderFrame *http2.MetaHeadersFrame
+		httpFlags       http2.Flags
+		wantStatus      *status.Status
+	}{
+		{
+			name: "1xx with END_STREAM is error",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					{Name: ":status", Value: "100"},
+				},
+			},
+			httpFlags: http2.FlagHeadersEndStream,
+			wantStatus: status.New(
+				codes.Internal,
+				"protocol error: informational header with status code 100 must not have END_STREAM set",
+			),
+		},
+		{
+			name: "1xx without END_STREAM is ignored",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					{Name: ":status", Value: "100"},
+				},
+			},
+			httpFlags:  0,
+			wantStatus: nil,
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			ts := testStream()
+			s := testClient(ts)
+
+			test.metaHeaderFrame.HeadersFrame = &http2.HeadersFrame{
+				FrameHeader: http2.FrameHeader{
+					StreamID: 0,
+					Flags:    test.httpFlags,
+				},
+			}
+
+			s.operateHeaders(test.metaHeaderFrame)
+
+			got := ts.status
+			want := test.wantStatus
+
+			if got.Code() != want.Code() || got.Message() != want.Message() {
+				t.Fatalf("operateHeaders(%v); status = %v, want %v", test.metaHeaderFrame, got, want)
+			}
+		})
+	}
+}
diff --git a/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go b/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go
index bad0798f..3020b495 100644
--- a/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go
+++ b/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go
@@ -41,124 +41,12 @@ 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 {
@@ -699,169 +587,756 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 	}
 }
 
-func (s) TestValidateAndConstructMetadataWithXDSHTTPConnectEnabled(t *testing.T) {
+// Tests custom metadata parsing for success cases when the
+// GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT environment variable is set.
+func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T) {
 	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
 	tests := []struct {
 		name          string
-		metadataProto *v3corepb.Metadata
-		want          map[string]any
-		wantErr       bool
+		endpointProto *v3endpointpb.ClusterLoadAssignment
+		wantEndpoint  EndpointsUpdate
 	}{
 		{
-			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,
+			name: "typed_filter_metadata_in_endpoint",
+			endpointProto: 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{
+							"test-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()
+			}(),
+			wantEndpoint: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+							Metadata: map[string]any{
+								"test-key": ProxyAddressMetadataValue{
+									Address: "1.2.3.4:1111",
+								},
+							},
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
 				},
 			},
-
-			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,
+			name: "filter_metadata_in_endpoint",
+			endpointProto: 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": {
+								Fields: map[string]*structpb.Value{
+									"key": {
+										Kind: &structpb.Value_NumberValue{NumberValue: 123.0},
+									},
 								},
 							},
 						},
-					}),
-				},
-				FilterMetadata: map[string]*structpb.Struct{
-					"untyped-key": {Fields: map[string]*structpb.Value{"field": structpb.NewStringValue("value")}},
+					},
+				}}, nil)
+				return clab0.Build()
+			}(),
+			wantEndpoint: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+							Metadata: map[string]any{
+								"test-key": StructMetadataValue{Data: map[string]any{
+									"key": float64(123),
+								}},
+							},
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
 				},
 			},
-			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",
+			name: "typed_filter_metadata_in_locality",
+			endpointProto: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
+					Metadata: &v3corepb.Metadata{
+						TypedFilterMetadata: map[string]*anypb.Any{
+							"test-key": testutils.MarshalAny(t, &v3corepb.Address{
+								Address: &v3corepb.Address_SocketAddress{
+									SocketAddress: &v3corepb.SocketAddress{
+										Address: "1.2.3.4",
+										PortSpecifier: &v3corepb.SocketAddress_PortValue{
+											PortValue: 1111,
+										}},
+								},
+							}),
+						},
+					},
+				})
+				return clab0.Build()
+			}(),
+			wantEndpoint: 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": ProxyAddressMetadataValue{
+								Address: "1.2.3.4:1111",
 							},
 						},
-					}),
+					},
 				},
 			},
-			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,
+			name: "filter_metadata_in_locality",
+			endpointProto: 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": {
+								Fields: map[string]*structpb.Value{
+									"key": {
+										Kind: &structpb.Value_NumberValue{NumberValue: 123.0},
+									},
 								},
 							},
 						},
-					}),
-				},
-				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,
-								},
-							},
+					},
+				})
+				return clab0.Build()
+			}(),
+			wantEndpoint: 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{
+								"key": float64(123),
+							}},
 						},
-					}),
-				},
-				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",
+			name: "typed_filter_metadata_over_filter_metadata_in_endpoint",
+			endpointProto: 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{
+							"test-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{
+							"test-key": {
+								Fields: map[string]*structpb.Value{
+									"key": {
+										Kind: &structpb.Value_NumberValue{NumberValue: 123.0},
+									},
+								},
 							},
 						},
-					}),
-				},
-			},
-			want: nil,
-		},
-	}
-
+					},
+				}}, nil)
+				return clab0.Build()
+			}(),
+			wantEndpoint: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+							Metadata: map[string]any{
+								"test-key": ProxyAddressMetadataValue{
+									Address: "1.2.3.4:1111",
+								},
+							},
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+		{
+			name: "typed_filter_metadata_over_filter_metadata_in_locality",
+			endpointProto: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
+					Metadata: &v3corepb.Metadata{
+						TypedFilterMetadata: map[string]*anypb.Any{
+							"test-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{
+							"test-key": {
+								Fields: map[string]*structpb.Value{
+									"key": {
+										Kind: &structpb.Value_NumberValue{NumberValue: 123.0},
+									},
+								},
+							},
+						},
+					},
+				})
+				return clab0.Build()
+			}(),
+			wantEndpoint: 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": ProxyAddressMetadataValue{
+								Address: "1.2.3.4:1111",
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "both_filter_and_typed_filter_metadata_in_endpoint",
+			endpointProto: 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{
+							"test-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{
+							"another-test-key": {
+								Fields: map[string]*structpb.Value{
+									"key": {
+										Kind: &structpb.Value_NumberValue{NumberValue: 123.0},
+									},
+								},
+							},
+						},
+					},
+				}}, nil)
+				return clab0.Build()
+			}(),
+			wantEndpoint: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+							Metadata: map[string]any{
+								"test-key": ProxyAddressMetadataValue{
+									Address: "1.2.3.4:1111",
+								},
+								"another-test-key": StructMetadataValue{Data: map[string]any{
+									"key": float64(123),
+								}},
+							},
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+		{
+			name: "both_filter_and_typed_filter_metadata_in_locality",
+			endpointProto: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
+					Metadata: &v3corepb.Metadata{
+						TypedFilterMetadata: map[string]*anypb.Any{
+							"test-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{
+							"another-test-key": {
+								Fields: map[string]*structpb.Value{
+									"key": {
+										Kind: &structpb.Value_NumberValue{NumberValue: 123.0},
+									},
+								},
+							},
+						},
+					},
+				})
+				return clab0.Build()
+			}(),
+			wantEndpoint: 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": ProxyAddressMetadataValue{
+								Address: "1.2.3.4:1111",
+							},
+							"another-test-key": StructMetadataValue{Data: map[string]any{
+								"key": float64(123),
+							}},
+						},
+					},
+				},
+			},
+		},
+	}
 	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)
+			got, err := parseEDSRespProto(tt.endpointProto)
+			if err != nil {
+				t.Fatalf("parseEDSRespProto() failed: %v", err)
 			}
-			if diff := cmp.Diff(tt.want, got, cmp.AllowUnexported(ProxyAddressMetadataValue{})); diff != "" {
-				t.Errorf("validateAndConstructMetadata() returned unexpected diff (-want +got):\n%s", diff)
+			if diff := cmp.Diff(tt.wantEndpoint, got, cmpopts.EquateEmpty()); diff != "" {
+				t.Errorf("parseEDSRespProto() returned unexpected diff (-want +got):\n%s", diff)
+			}
+		})
+	}
+}
+
+// Tests custom metadata parsing for success cases when the
+// GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT environment variable is not set.
+func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, false)
+	tests := []struct {
+		name          string
+		endpointProto *v3endpointpb.ClusterLoadAssignment
+		wantEndpoint  EndpointsUpdate
+	}{
+		{
+			name: "typed_filter_metadata_in_endpoint",
+			endpointProto: 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{
+							"test-key": testutils.MarshalAny(t, &v3corepb.Address{
+								Address: &v3corepb.Address_SocketAddress{
+									SocketAddress: &v3corepb.SocketAddress{
+										Address: "1.2.3.4",
+										PortSpecifier: &v3corepb.SocketAddress_PortValue{
+											PortValue: 1111,
+										}},
+								},
+							}),
+						},
+					},
+				}}, nil)
+				return clab0.Build()
+			}(),
+			wantEndpoint: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+		{
+			name: "filter_metadata_in_endpoint",
+			endpointProto: 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": {
+								Fields: map[string]*structpb.Value{
+									"key": {
+										Kind: &structpb.Value_NumberValue{NumberValue: 123.0},
+									},
+								},
+							},
+						},
+					},
+				}}, nil)
+				return clab0.Build()
+			}(),
+			wantEndpoint: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+		{
+			name: "typed_filter_metadata_in_locality",
+			endpointProto: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
+					Metadata: &v3corepb.Metadata{
+						TypedFilterMetadata: map[string]*anypb.Any{
+							"test-key": testutils.MarshalAny(t, &v3corepb.Address{
+								Address: &v3corepb.Address_SocketAddress{
+									SocketAddress: &v3corepb.SocketAddress{
+										Address: "1.2.3.4",
+										PortSpecifier: &v3corepb.SocketAddress_PortValue{
+											PortValue: 1111,
+										}},
+								},
+							}),
+						},
+					},
+				})
+				return clab0.Build()
+			}(),
+			wantEndpoint: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+		{
+			name: "filter_metadata_in_locality",
+			endpointProto: 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": {
+								Fields: map[string]*structpb.Value{
+									"key": {
+										Kind: &structpb.Value_NumberValue{NumberValue: 123.0},
+									},
+								},
+							},
+						},
+					},
+				})
+				return clab0.Build()
+			}(),
+			wantEndpoint: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+		{
+			name: "both_filter_and_typed_filter_metadata_in_endpoint",
+			endpointProto: 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{
+							"test-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{
+							"another-test-key": {
+								Fields: map[string]*structpb.Value{
+									"key": {
+										Kind: &structpb.Value_NumberValue{NumberValue: 123.0},
+									},
+								},
+							},
+						},
+					},
+				}}, nil)
+				return clab0.Build()
+			}(),
+			wantEndpoint: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+		{
+			name: "both_filter_and_typed_filter_metadata_in_locality",
+			endpointProto: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
+					Metadata: &v3corepb.Metadata{
+						TypedFilterMetadata: map[string]*anypb.Any{
+							"test-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{
+							"another-test-key": {
+								Fields: map[string]*structpb.Value{
+									"key": {
+										Kind: &structpb.Value_NumberValue{NumberValue: 123.0},
+									},
+								},
+							},
+						},
+					},
+				})
+				return clab0.Build()
+			}(),
+			wantEndpoint: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+		{
+			name: "converter_failure_is_not_triggerred",
+			endpointProto: 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{
+							"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+								Address: &v3corepb.Address_SocketAddress{
+									SocketAddress: &v3corepb.SocketAddress{
+										Address: "invalid",
+									},
+								},
+							}),
+						},
+					},
+				}}, &addLocalityOptions{
+					Metadata: &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",
+									},
+								},
+							}),
+						},
+					},
+				})
+				return clab0.Build()
+			}(),
+			wantEndpoint: EndpointsUpdate{
+				Localities: []Locality{
+					{
+						Endpoints: []Endpoint{{
+							Addresses:    []string{"addr1:314"},
+							HealthStatus: EndpointHealthStatusUnknown,
+							Weight:       1,
+						}},
+						ID:       clients.Locality{SubZone: "locality-1"},
+						Priority: 0,
+						Weight:   1,
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := parseEDSRespProto(tt.endpointProto)
+			if err != nil {
+				t.Fatalf("parseEDSRespProto() failed: %v", err)
+			}
+			if diff := cmp.Diff(tt.wantEndpoint, got, cmpopts.EquateEmpty()); diff != "" {
+				t.Errorf("parseEDSRespProto() returned unexpected diff (-want +got):\n%s", diff)
+			}
+		})
+	}
+}
+
+// Tests custom metadata parsing for converter failure cases when the
+// GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT environment variable is set.
+func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_ConverterFailure(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
+	tests := []struct {
+		name          string
+		endpointProto *v3endpointpb.ClusterLoadAssignment
+	}{
+		{
+			name: "converter_failure_in_endpoint",
+			endpointProto: 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{
+							"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+								Address: &v3corepb.Address_SocketAddress{
+									SocketAddress: &v3corepb.SocketAddress{
+										Address: "invalid",
+									},
+								},
+							}),
+						},
+					},
+				}}, nil)
+				return clab0.Build()
+			}(),
+		},
+		{
+			name: "converter_failure_in_locality",
+			endpointProto: func() *v3endpointpb.ClusterLoadAssignment {
+				clab0 := newClaBuilder("test", nil)
+				clab0.addLocality("locality-1", 1, 0, []endpointOpts{{addrWithPort: "addr1:314"}}, &addLocalityOptions{
+					Metadata: &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",
+									},
+								},
+							}),
+						},
+					},
+				})
+				return clab0.Build()
+			}(),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if _, err := parseEDSRespProto(tt.endpointProto); err == nil {
+				t.Fatalf("parseEDSRespProto() did not return error when expected")
 			}
 		})
 	}
diff --git a/test/transport_test.go b/test/transport_test.go
index 10e9ab57..3f0a8260 100644
--- a/test/transport_test.go
+++ b/test/transport_test.go
@@ -229,3 +229,76 @@ func (s) TestRSTDuringMessageRead(t *testing.T) {
 		t.Fatalf("client.EmptyCall() returned %v; want status with code %v", err, codes.Canceled)
 	}
 }
+
+// Test verifies that a client-side cancellation correctly frees up resources on
+// the server. The test setup is designed to simulate a scenario where a server
+// is blocked from sending a large message due to a full client-side flow
+// control window. The client-side cancellation of this blocked RPC then frees
+// up the max concurrent streams quota on the server, allowing a new RPC to be
+// created successfully.
+func (s) TestCancelWhileServerWaitingForFlowControl(t *testing.T) {
+	serverDoneCh := make(chan struct{}, 2)
+	const flowControlWindowSize = 65535
+	ss := &stubserver.StubServer{
+		StreamingOutputCallF: func(_ *testpb.StreamingOutputCallRequest, stream testgrpc.TestService_StreamingOutputCallServer) error {
+			// Send a large message to exhaust the client's flow control window.
+			stream.Send(&testpb.StreamingOutputCallResponse{
+				Payload: &testpb.Payload{
+					Body: make([]byte, flowControlWindowSize+1),
+				},
+			})
+			serverDoneCh <- struct{}{}
+			return nil
+		},
+	}
+
+	// Create a server that allows only 1 stream at a time.
+	ss = stubserver.StartTestService(t, ss, grpc.MaxConcurrentStreams(1))
+	defer ss.Stop()
+	// Use a static flow control window.
+	if err := ss.StartClient(grpc.WithStaticStreamWindowSize(flowControlWindowSize)); err != nil {
+		t.Fatalf("Error while start test service client: %v", err)
+	}
+	client := ss.Client
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	streamCtx, streamCancel := context.WithCancel(ctx)
+	defer streamCancel()
+
+	if _, err := client.StreamingOutputCall(streamCtx, &testpb.StreamingOutputCallRequest{}); err != nil {
+		t.Fatalf("Failed to create server streaming RPC: %v", err)
+	}
+
+	// Wait for the server handler to return. This should cause the trailers to
+	// be buffered on the server, waiting for flow control quota to first send
+	// the data frame.
+	select {
+	case <-ctx.Done():
+		t.Fatal("Context timed out waiting for server handler to return.")
+	case <-serverDoneCh:
+	}
+
+	// Attempt to create a stream. It should fail since the previous stream is
+	// still blocked.
+	shortCtx, shortCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer shortCancel()
+	_, err := client.StreamingOutputCall(shortCtx, &testpb.StreamingOutputCallRequest{})
+	if status.Code(err) != codes.DeadlineExceeded {
+		t.Fatalf("Server stream creation returned error with unexpected status code: %v, want code: %v", err, codes.DeadlineExceeded)
+	}
+
+	// Cancel the RPC, this should free up concurrent stream quota on the
+	// server.
+	streamCancel()
+
+	// Attempt to create another stream.
+	stream, err := client.StreamingOutputCall(ctx, &testpb.StreamingOutputCallRequest{})
+	if err != nil {
+		t.Fatalf("Failed to create server streaming RPC: %v", err)
+	}
+	_, err = stream.Recv()
+	if err != nil {
+		t.Fatalf("Failed to read from the stream: %v", err)
+	}
+}
diff --git a/xds/googledirectpath/googlec2p_test.go b/xds/googledirectpath/googlec2p_test.go
index adb5c3a3..74ee65db 100644
--- a/xds/googledirectpath/googlec2p_test.go
+++ b/xds/googledirectpath/googlec2p_test.go
@@ -103,6 +103,29 @@ func useCleanUniverseDomain(t *testing.T) {
 	})
 }
 
+// TODO(https://github.com/grpc/grpc-go/issues/8561): this content can be hardcoded directly
+// in wanted bootstraps again after old pick first is removed.
+func expectedNodeJSON(ipv6Capable bool) []byte {
+	if !envconfig.NewPickFirstEnabled && !ipv6Capable {
+		return []byte(`{
+			"id": "C2P-666",
+			"locality": {
+				"zone": "test-zone"
+			}
+		}`)
+	}
+	// Otherwise, return the node metadata including the IPv6 capability flag.
+	return []byte(`{
+		"id": "C2P-666",
+		"locality": {
+			"zone": "test-zone"
+		},
+		"metadata": {
+			"TRAFFICDIRECTOR_DIRECTPATH_C2P_IPV6_CAPABLE": true
+		}
+	}`)
+}
+
 // Tests the scenario where the bootstrap env vars are set and we're running on
 // GCE. The test builds a google-c2p resolver and verifies that an xDS resolver
 // is built and that we don't fallback to DNS (because federation is enabled by
@@ -218,10 +241,7 @@ func (s) TestBuildXDS(t *testing.T) {
 							]
 						}`),
 				},
-				Node: []byte(`{
-					  "id": "C2P-666",
-					  "locality": {"zone": "test-zone"}
-					}`),
+				Node: expectedNodeJSON(false),
 			}),
 		},
 		{
@@ -244,13 +264,7 @@ func (s) TestBuildXDS(t *testing.T) {
 							]
 						}`),
 				},
-				Node: []byte(`{
-					  "id": "C2P-666",
-					  "locality": {"zone": "test-zone"},
-			  			"metadata": {
-							"TRAFFICDIRECTOR_DIRECTPATH_C2P_IPV6_CAPABLE": true
-			  			}
-					}`),
+				Node: expectedNodeJSON(true),
 			}),
 		},
 		{
@@ -274,13 +288,7 @@ func (s) TestBuildXDS(t *testing.T) {
 							]
 						}`),
 				},
-				Node: []byte(`{
-					  "id": "C2P-666",
-					  "locality": {"zone": "test-zone"},
-			  			"metadata": {
-							"TRAFFICDIRECTOR_DIRECTPATH_C2P_IPV6_CAPABLE": true
-			  			}
-					}`),
+				Node: expectedNodeJSON(true),
 			}),
 		},
 	} {
@@ -442,10 +450,7 @@ func (s) TestSetUniverseDomainNonDefault(t *testing.T) {
 							]
 						}`),
 		},
-		Node: []byte(`{
-					  "id": "C2P-666",
-					  "locality": {"zone": "test-zone"}
-					}`),
+		Node: expectedNodeJSON(false),
 	})
 	if diff := cmp.Diff(wantBootstrapConfig, gotConfig); diff != "" {
 		t.Fatalf("Unexpected diff in bootstrap config (-want +got):\n%s", diff)
@@ -513,10 +518,7 @@ func (s) TestDefaultUniverseDomain(t *testing.T) {
 							]
 						}`),
 		},
-		Node: []byte(`{
-					  "id": "C2P-666",
-					  "locality": {"zone": "test-zone"}
-					}`),
+		Node: expectedNodeJSON(false),
 	})
 	if diff := cmp.Diff(wantBootstrapConfig, gotConfig); diff != "" {
 		t.Fatalf("Unexpected diff in bootstrap config (-want +got):\n%s", diff)