Repository: grpc/grpc-go

Issue:
streaming: should return cancel error instead of Internal error with unexpected EOF
### What version of gRPC are you using?

v1.72.0

### What version of Go are you using (`go version`)?

v1.24.2

### What operating system (Linux, Windows, …) and version?

Linux

### What did you do?


In the ETCD integration test, there is a case that verifies the client receives a context canceled error when reading data from a streaming RPC and the context is subsequently canceled. The steps would be like

* Create a context with a one-second timeout (ctx).
* Use that ctx to initiate the streaming RPC, and start a new goroutine to read the stream and write received data into a pipe.
* The client sleeps for two seconds and then attempts to read from the pipe.
    * NOTE: The background goroutine may read one message before timeout. https://github.com/etcd-io/etcd/blob/0a0fde11ace45eaf596862df35aae833b8e6d649/client/v3/maintenance.go#L284-L293
* The read should return a context canceled error.

```bash
# in etcd main branch
$ cd ./tests/integration/clientv3
$ go test -v ./ -count=10 --failfast -run TestMaintenanceSnapshotTimeout
```

### What did you expect to see?

I should receice context canceled error

### What did you see instead?

received `rpc error: code = Internal desc = unexpected EOF`

I can't reproduce this issue with v1.71.0 version. Right now, I found that https://github.com/grpc/grpc-go/blob/a43eba6fed49b81b84cfdba85c356aca22086d7e/rpc_util.go#L676 - the p.r.Read returns io.EOF and then convert it into unexpectedEOF. 

REF: https://github.com/etcd-io/etcd/pull/19833#issuecomment-2840562143

cc @ahrtr @dfawley 


Hints from issue comments:
@ahrtr: To add more context to clarify this issue from users perspective.

When etcd calls the `Recv` method (generated gRPC code, see below, also see [here](https://github.com/etcd-io/etcd/blob/8da854dc69729868531d8f3e360b77c4c046eae1/api/etcdserverpb/rpc.pb.go#L7573-L7579)), when the context (the context used when creating the gRPC stream, see [here](https://github.com/etcd-io/etcd/blob/d610b0431bdde9ff12b97ee70993137c78681592/api/etcdserverpb/rpc.pb.go#L7549) ) timeout, 
- previously (on grpc-go 1.71), the `Recv` always returned a `rpc error: code = DeadlineExceeded desc = context deadline exceeded` error. 
- After we bumped to grpc-go v1.72, the `Recv` sometimes returns the same expected error, but sometimes return unexpected error `rpc error: code = Internal desc = unexpected EOF`. This changed behaviour makes our test flaky.



```
func (x *maintenanceSnapshotClient) Recv() (*SnapshotResponse, error) {
	m := new(SnapshotResponse)
	if err := x.ClientStream.RecvMsg(m); err != nil {
		return nil, err
	}
	return m, nil
}
```

I believe this should be a generic problem for all grpc-go's users, not specific to etcd.@arjan-bal: I suspect this is a bug in the gRPC client-side stream handling that was exposed by changes in https://github.com/grpc/grpc-go/pull/8071. This PR made gRPC Go servers cancel the stream on seeing the deadline being exceeded by sending an RST_STREAM with status code CANCELLED. The client monitors the deadline too, so it's a race between the client cancelling the stream and the server cancelling the stream. Usually clients would cancel the RPC before receiving the cancellation from the server. 

On the client side, the [handleRSTStream method](https://github.com/grpc/grpc-go/blob/080f9563df8b6a1eb4230cfcd500bc9d7930ecf8/internal/transport/http2_client.go#L1241) which converts the CANCELLED status to a DEADLINE_EXCEEDED and calls `closeStream()` with an `io.EOF` error.
https://github.com/grpc/grpc-go/blob/080f9563df8b6a1eb4230cfcd500bc9d7930ecf8/internal/transport/http2_client.go#L1257-L1265

`closeStream()` writes this `io.EOF` error for the reading goroutine to stop gracefully.
https://github.com/grpc/grpc-go/blob/080f9563df8b6a1eb4230cfcd500bc9d7930ecf8/internal/transport/http2_client.go#L942-L945

As pointed out in the issue description, this `io.EOF` gets converted to an `io.ErrUnexpectedEOF` if the reader was in the middle of reading a gRPC message. This could happen when a gRPC message is split across multiple HTTP/2 data frames.
https://github.com/grpc/grpc-go/blob/a43eba6fed49b81b84cfdba85c356aca22086d7e/rpc_util.go#L674-L680 

In stream.go, we have multiple places where we return the RPC status when we encounter an `io.EOF` while reading the stream, example
https://github.com/grpc/grpc-go/blob/080f9563df8b6a1eb4230cfcd500bc9d7930ecf8/stream.go#L1134-L1139

As the `io.EOF` is converted to an `io.ErrUnexpectedEOF`, the RPC status is not returned.@dfawley: Good find!  That sounds like exactly what's happening.

Is there any value in the `io.UnexpectedEOF`?  Or can we use the stream's status (if there is one, which I think there always should be) in this case?@arjan-bal: > Is there any value in the io.UnexpectedEOF?

~~An `io.UnexpectedEOF` may be an internal error when we're trying to read  for example the 5 byte gRPC message header, but receive an io.EOF before reading the required number of bytes.~~
On re-checking, an `io.UnexpectedEOF` could be due to different status codes. For example when a incorrectly structured HTTP/2 frame is read by the framer, it would result in `http2Client.closeStream()` being called with an `Unavailable` gRPC status. 

> Or can we use the stream's status (if there is one, which I think there always should be) in this case?

This seems like the safer change to make. I'm not a 100% sure if there will always be an RPC status available when we encounter  and `io.UnexpectedEOF`, I couldn't think of a case when the stream's status is not available. I've checked for a non-nil status in [the fix](https://github.com/grpc/grpc-go/pull/8289), keeping the existing behaviour in case of a nil status.

Solution patch:
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index d9bfa6e1..43b64bc6 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -1,73 +1,102 @@
 # How to contribute
 
-We definitely welcome your patches and contributions to gRPC! Please read the gRPC
-organization's [governance rules](https://github.com/grpc/grpc-community/blob/master/governance.md)
-and [contribution guidelines](https://github.com/grpc/grpc-community/blob/master/CONTRIBUTING.md) before proceeding.
+We welcome your patches and contributions to gRPC! Please read the gRPC
+organization's [governance
+rules](https://github.com/grpc/grpc-community/blob/master/governance.md) before
+proceeding.
 
 If you are new to GitHub, please start by reading [Pull Request howto](https://help.github.com/articles/about-pull-requests/)
 
 ## Legal requirements
 
 In order to protect both you and ourselves, you will need to sign the
-[Contributor License Agreement](https://identity.linuxfoundation.org/projects/cncf).
+[Contributor License
+Agreement](https://identity.linuxfoundation.org/projects/cncf). When you create
+your first PR, a link will be added as a comment that contains the steps needed
+to complete this process.
 
-## Guidelines for Pull Requests
-How to get your contributions merged smoothly and quickly.
+## Getting Started
 
-- Create **small PRs** that are narrowly focused on **addressing a single
-  concern**. We often times receive PRs that are trying to fix several things at
-  a time, but only one fix is considered acceptable, nothing gets merged and
-  both author's & review's time is wasted. Create more PRs to address different
-  concerns and everyone will be happy.
+A great way to start is by searching through our open issues. [Unassigned issues
+labeled as "help
+wanted"](https://github.com/grpc/grpc-go/issues?q=sort%3Aupdated-desc%20is%3Aissue%20is%3Aopen%20label%3A%22Status%3A%20Help%20Wanted%22%20no%3Aassignee)
+are especially nice for first-time contributors, as they should be well-defined
+problems that already have agreed-upon solutions.
 
-- If you are searching for features to work on, issues labeled [Status: Help
-  Wanted](https://github.com/grpc/grpc-go/issues?q=is%3Aissue+is%3Aopen+sort%3Aupdated-desc+label%3A%22Status%3A+Help+Wanted%22)
-  is a great place to start. These issues are well-documented and usually can be
-  resolved with a single pull request.
+## Code Style
 
-- If you are adding a new file, make sure it has the copyright message template
-  at the top as a comment. You can copy over the message from an existing file
-  and update the year.
+We follow [Google's published Go style
+guide](https://google.github.io/styleguide/go/). Note that there are three
+primary documents that make up this style guide; please follow them as closely
+as possible. If a reviewer recommends something that contradicts those
+guidelines, there may be valid reasons to do so, but it should be rare.
 
-- The grpc package should only depend on standard Go packages and a small number
-  of exceptions. If your contribution introduces new dependencies which are NOT
-  in the [list](https://godoc.org/google.golang.org/grpc?imports), you need a
-  discussion with gRPC-Go authors and consultants.
+## Guidelines for Pull Requests
 
-- For speculative changes, consider opening an issue and discussing it first. If
-  you are suggesting a behavioral or API change, consider starting with a [gRFC
-  proposal](https://github.com/grpc/proposal).
+How to get your contributions merged smoothly and quickly:
+
+- Create **small PRs** that are narrowly focused on **addressing a single
+  concern**. We often receive PRs that attempt to fix several things at the same
+  time, and if one part of the PR has a problem, that will hold up the entire
+  PR.
+
+- For **speculative changes**, consider opening an issue and discussing it
+  first. If you are suggesting a behavioral or API change, consider starting
+  with a [gRFC proposal](https://github.com/grpc/proposal). Many new features
+  that are not bug fixes will require cross-language agreement.
+
+- If you want to fix **formatting or style**, consider whether your changes are
+  an obvious improvement or might be considered a personal preference. If a
+  style change is based on preference, it likely will not be accepted. If it
+  corrects widely agreed-upon anti-patterns, then please do create a PR and
+  explain the benefits of the change.
+
+- For correcting **misspellings**, please be aware that we use some terms that
+  are sometimes flagged by spell checkers. As an example, "if an only if" is
+  often written as "iff". Please do not make spelling correction changes unless
+  you are certain they are misspellings.
 
 - Provide a good **PR description** as a record of **what** change is being made
   and **why** it was made. Link to a GitHub issue if it exists.
 
-- If you want to fix formatting or style, consider whether your changes are an
-  obvious improvement or might be considered a personal preference. If a style
-  change is based on preference, it likely will not be accepted. If it corrects
-  widely agreed-upon anti-patterns, then please do create a PR and explain the
-  benefits of the change.
+- Maintain a **clean commit history** and use **meaningful commit messages**.
+  PRs with messy commit histories are difficult to review and won't be merged.
+  Before sending your PR, ensure your changes are based on top of the latest
+  `upstream/master` commits, and avoid rebasing in the middle of a code review.
+  You should **never use `git push -f`** unless absolutely necessary during a
+  review, as it can interfere with GitHub's tracking of comments.
 
-- Unless your PR is trivial, you should expect there will be reviewer comments
-  that you'll need to address before merging. We'll mark it as `Status: Requires
-  Reporter Clarification` if we expect you to respond to these comments in a
-  timely manner. If the PR remains inactive for 6 days, it will be marked as
-  `stale` and automatically close 7 days after that if we don't hear back from
-  you.
+- **All tests need to be passing** before your change can be merged. We
+  recommend you run tests locally before creating your PR to catch breakages
+  early on:
 
-- Maintain **clean commit history** and use **meaningful commit messages**. PRs
-  with messy commit history are difficult to review and won't be merged. Use
-  `rebase -i upstream/master` to curate your commit history and/or to bring in
-  latest changes from master (but avoid rebasing in the middle of a code
-  review).
+  - `./scripts/vet.sh` to catch vet errors.
+  - `go test -cpu 1,4 -timeout 7m ./...` to run the tests.
+  - `go test -race -cpu 1,4 -timeout 7m ./...` to run tests in race mode.
 
-- Keep your PR up to date with upstream/master (if there are merge conflicts, we
-  can't really merge your change).
+  Note that we have a multi-module repo, so `go test` commands may need to be
+  run from the root of each module in order to cause all tests to run.
 
-- **All tests need to be passing** before your change can be merged. We
-  recommend you **run tests locally** before creating your PR to catch breakages
-  early on.
-  - `./scripts/vet.sh` to catch vet errors
-  - `go test -cpu 1,4 -timeout 7m ./...` to run the tests
-  - `go test -race -cpu 1,4 -timeout 7m ./...` to run tests in race mode
+  *Alternatively*, you may find it easier to push your changes to your fork on
+  GitHub, which will trigger a GitHub Actions run that you can use to verify
+  everything is passing.
+
+- If you are adding a new file, make sure it has the **copyright message**
+  template at the top as a comment. You can copy the message from an existing
+  file and update the year.
+
+- The grpc package should only depend on standard Go packages and a small number
+  of exceptions. **If your contribution introduces new dependencies**, you will
+  need a discussion with gRPC-Go maintainers. A github action check will run on
+  every PR, and will flag any transitive dependency changes from any public
+  package.
+
+- Unless your PR is trivial, you should **expect reviewer comments** that you
+  will need to address before merging. We'll label the PR as `Status: Requires
+  Reporter Clarification` if we expect you to respond to these comments in a
+  timely manner. If the PR remains inactive for 6 days, it will be marked as
+  `stale`, and we will automatically close it after 7 days if we don't hear back
+  from you. Please feel free to ping issues or bugs if you do not get a response
+  within a week.
 
-- Exceptions to the rules can be made if there's a compelling reason for doing so.
+- Exceptions to the rules can be made if there's a compelling reason to do so.
diff --git a/Documentation/encoding.md b/Documentation/encoding.md
index cce9384f..00e19cc8 100644
--- a/Documentation/encoding.md
+++ b/Documentation/encoding.md
@@ -57,7 +57,7 @@ As a reminder, all `CallOption`s may be converted into `DialOption`s that become
 the default for all RPCs sent through a client using `grpc.WithDefaultCallOptions`:
 
 ```go
-    myclient := grpc.Dial(ctx, target, grpc.WithDefaultCallOptions(grpc.CallContentSubtype("mycodec")))
+    myclient := grpc.NewClient(target, grpc.WithDefaultCallOptions(grpc.CallContentSubtype("mycodec")))
 ```
 
 When specified in either of these ways, messages will be encoded using this
@@ -132,7 +132,7 @@ As a reminder, all `CallOption`s may be converted into `DialOption`s that become
 the default for all RPCs sent through a client using `grpc.WithDefaultCallOptions`:
 
 ```go
-    myclient := grpc.Dial(ctx, target, grpc.WithDefaultCallOptions(grpc.UseCompressor("gzip")))
+    myclient := grpc.NewClient(target, grpc.WithDefaultCallOptions(grpc.UseCompressor("gzip")))
 ```
 
 When specified in either of these ways, messages will be compressed using this
diff --git a/Documentation/grpc-auth-support.md b/Documentation/grpc-auth-support.md
index cedcaa6f..c8c2a556 100644
--- a/Documentation/grpc-auth-support.md
+++ b/Documentation/grpc-auth-support.md
@@ -5,7 +5,7 @@ As outlined in the [gRPC authentication guide](https://grpc.io/docs/guides/auth.
 # Enabling TLS on a gRPC client
 
 ```Go
-conn, err := grpc.Dial(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")))
+conn, err := grpc.NewClient(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")))
 ```
 
 # Enabling TLS on a gRPC server
@@ -63,7 +63,7 @@ to prevent any insecure transmission of tokens.
 ## Google Compute Engine (GCE)
 
 ```Go
-conn, err := grpc.Dial(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(oauth.NewComputeEngine()))
+conn, err := grpc.NewClient(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(oauth.NewComputeEngine()))
 ```
 
 ## JWT
@@ -73,6 +73,6 @@ jwtCreds, err := oauth.NewServiceAccountFromFile(*serviceAccountKeyFile, *oauthS
 if err != nil {
   log.Fatalf("Failed to create JWT credentials: %v", err)
 }
-conn, err := grpc.Dial(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(jwtCreds))
+conn, err := grpc.NewClient(serverAddr, grpc.WithTransportCredentials(credentials.NewClientTLSFromCert(nil, "")), grpc.WithPerRPCCredentials(jwtCreds))
 ```
 
diff --git a/README.md b/README.md
index b572707c..f9a88d59 100644
--- a/README.md
+++ b/README.md
@@ -32,6 +32,7 @@ import "google.golang.org/grpc"
 - [Low-level technical docs](Documentation) from this repository
 - [Performance benchmark][]
 - [Examples](examples)
+- [Contribution guidelines](CONTRIBUTING.md)
 
 ## FAQ
 
diff --git a/balancer/weightedroundrobin/balancer.go b/balancer/weightedroundrobin/balancer.go
index 0ee70760..e6fe5a37 100644
--- a/balancer/weightedroundrobin/balancer.go
+++ b/balancer/weightedroundrobin/balancer.go
@@ -497,7 +497,6 @@ func (p *picker) start(stopPicker *grpcsync.Event) {
 // that listener.
 type endpointWeight struct {
 	// The following fields are immutable.
-	balancer.SubConn
 	logger          *grpclog.PrefixLogger
 	target          string
 	metricsRecorder estats.MetricsRecorder
@@ -527,7 +526,7 @@ type endpointWeight struct {
 
 func (w *endpointWeight) OnLoadReport(load *v3orcapb.OrcaLoadReport) {
 	if w.logger.V(2) {
-		w.logger.Infof("Received load report for subchannel %v: %v", w.SubConn, load)
+		w.logger.Infof("Received load report for subchannel %v: %v", w.pickedSC, load)
 	}
 	// Update weights of this endpoint according to the reported load.
 	utilization := load.ApplicationUtilization
@@ -536,7 +535,7 @@ func (w *endpointWeight) OnLoadReport(load *v3orcapb.OrcaLoadReport) {
 	}
 	if utilization == 0 || load.RpsFractional == 0 {
 		if w.logger.V(2) {
-			w.logger.Infof("Ignoring empty load report for subchannel %v", w.SubConn)
+			w.logger.Infof("Ignoring empty load report for subchannel %v", w.pickedSC)
 		}
 		return
 	}
@@ -547,7 +546,7 @@ func (w *endpointWeight) OnLoadReport(load *v3orcapb.OrcaLoadReport) {
 	errorRate := load.Eps / load.RpsFractional
 	w.weightVal = load.RpsFractional / (utilization + errorRate*w.cfg.ErrorUtilizationPenalty)
 	if w.logger.V(2) {
-		w.logger.Infof("New weight for subchannel %v: %v", w.SubConn, w.weightVal)
+		w.logger.Infof("New weight for subchannel %v: %v", w.pickedSC, w.weightVal)
 	}
 
 	w.lastUpdated = internal.TimeNow()
diff --git a/credentials/alts/internal/handshaker/service/service.go b/credentials/alts/internal/handshaker/service/service.go
index e0a1afc1..2580995d 100644
--- a/credentials/alts/internal/handshaker/service/service.go
+++ b/credentials/alts/internal/handshaker/service/service.go
@@ -22,9 +22,12 @@ package service
 
 import (
 	"sync"
+	"time"
 
 	grpc "google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal/envconfig"
+	"google.golang.org/grpc/keepalive"
 )
 
 var (
@@ -50,7 +53,17 @@ func Dial(hsAddress string) (*grpc.ClientConn, error) {
 		// Disable the service config to avoid unnecessary TXT record lookups that
 		// cause timeouts with some versions of systemd-resolved.
 		var err error
-		hsConn, err = grpc.NewClient(hsAddress, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDisableServiceConfig())
+		opts := []grpc.DialOption{
+			grpc.WithTransportCredentials(insecure.NewCredentials()),
+			grpc.WithDisableServiceConfig(),
+		}
+		if envconfig.ALTSHandshakerKeepaliveParams {
+			opts = append(opts, grpc.WithKeepaliveParams(keepalive.ClientParameters{
+				Timeout: 10 * time.Second,
+				Time:    10 * time.Minute,
+			}))
+		}
+		hsConn, err = grpc.NewClient(hsAddress, opts...)
 		if err != nil {
 			return nil, err
 		}
diff --git a/credentials/insecure/insecure.go b/credentials/insecure/insecure.go
index f45c04f7..93156c0f 100644
--- a/credentials/insecure/insecure.go
+++ b/credentials/insecure/insecure.go
@@ -30,7 +30,7 @@ import (
 // NewCredentials returns a credentials which disables transport security.
 //
 // Note that using this credentials with per-RPC credentials which require
-// transport security is incompatible and will cause grpc.Dial() to fail.
+// transport security is incompatible and will cause RPCs to fail.
 func NewCredentials() credentials.TransportCredentials {
 	return insecureTC{}
 }
diff --git a/credentials/local/local.go b/credentials/local/local.go
index 5991d0ad..2fe0215d 100644
--- a/credentials/local/local.go
+++ b/credentials/local/local.go
@@ -49,6 +49,12 @@ func (info) AuthType() string {
 	return "local"
 }
 
+// ValidateAuthority allows any value to be overridden for the :authority
+// header.
+func (info) ValidateAuthority(string) error {
+	return nil
+}
+
 // localTC is the credentials required to establish a local connection.
 type localTC struct {
 	info credentials.ProtocolInfo
diff --git a/dialoptions.go b/dialoptions.go
index 405a2ffe..050ba0f1 100644
--- a/dialoptions.go
+++ b/dialoptions.go
@@ -360,7 +360,7 @@ func WithReturnConnectionError() DialOption {
 //
 // Note that using this DialOption with per-RPC credentials (through
 // WithCredentialsBundle or WithPerRPCCredentials) which require transport
-// security is incompatible and will cause grpc.Dial() to fail.
+// security is incompatible and will cause RPCs to fail.
 //
 // Deprecated: use WithTransportCredentials and insecure.NewCredentials()
 // instead. Will be supported throughout 1.x.
diff --git a/health/server.go b/health/server.go
index d4b4b708..d8eebe99 100644
--- a/health/server.go
+++ b/health/server.go
@@ -30,6 +30,13 @@ import (
 	"google.golang.org/grpc/status"
 )
 
+const (
+	// maxAllowedServices defines the maximum number of resources a List
+	// operation can return. An error is returned if the number of services
+	// exceeds this limit.
+	maxAllowedServices = 100
+)
+
 // Server implements `service Health`.
 type Server struct {
 	healthgrpc.UnimplementedHealthServer
@@ -62,6 +69,23 @@ func (s *Server) Check(_ context.Context, in *healthpb.HealthCheckRequest) (*hea
 	return nil, status.Error(codes.NotFound, "unknown service")
 }
 
+// List implements `service Health`.
+func (s *Server) List(_ context.Context, _ *healthpb.HealthListRequest) (*healthpb.HealthListResponse, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	if len(s.statusMap) > maxAllowedServices {
+		return nil, status.Errorf(codes.ResourceExhausted, "server health list exceeds maximum capacity: %d", maxAllowedServices)
+	}
+
+	statusMap := make(map[string]*healthpb.HealthCheckResponse, len(s.statusMap))
+	for k, v := range s.statusMap {
+		statusMap[k] = &healthpb.HealthCheckResponse{Status: v}
+	}
+
+	return &healthpb.HealthListResponse{Statuses: statusMap}, nil
+}
+
 // Watch implements `service Health`.
 func (s *Server) Watch(in *healthpb.HealthCheckRequest, stream healthgrpc.Health_WatchServer) error {
 	service := in.Service
diff --git a/internal/envconfig/envconfig.go b/internal/envconfig/envconfig.go
index f122a041..f5f2bdeb 100644
--- a/internal/envconfig/envconfig.go
+++ b/internal/envconfig/envconfig.go
@@ -69,6 +69,10 @@ var (
 	// to gRFC A76. It can be enabled by setting the environment variable
 	// "GRPC_EXPERIMENTAL_RING_HASH_SET_REQUEST_HASH_KEY" to "true".
 	RingHashSetRequestHashKey = boolFromEnv("GRPC_EXPERIMENTAL_RING_HASH_SET_REQUEST_HASH_KEY", false)
+
+	// ALTSHandshakerKeepaliveParams is set if we should add the
+	// KeepaliveParams when dial the ALTS handshaker service.
+	ALTSHandshakerKeepaliveParams = boolFromEnv("GRPC_EXPERIMENTAL_ALTS_HANDSHAKER_KEEPALIVE_PARAMS", false)
 )
 
 func boolFromEnv(envVar string, def bool) bool {
diff --git a/internal/resolver/delegatingresolver/delegatingresolver.go b/internal/resolver/delegatingresolver/delegatingresolver.go
index 7b93f692..c0e22757 100644
--- a/internal/resolver/delegatingresolver/delegatingresolver.go
+++ b/internal/resolver/delegatingresolver/delegatingresolver.go
@@ -28,6 +28,8 @@ import (
 
 	"google.golang.org/grpc/grpclog"
 	"google.golang.org/grpc/internal/proxyattributes"
+	"google.golang.org/grpc/internal/transport"
+	"google.golang.org/grpc/internal/transport/networktype"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/serviceconfig"
 )
@@ -40,7 +42,7 @@ var (
 
 // delegatingResolver manages both target URI and proxy address resolution by
 // delegating these tasks to separate child resolvers. Essentially, it acts as
-// a intermediary between the gRPC ClientConn and the child resolvers.
+// an intermediary between the gRPC ClientConn and the child resolvers.
 //
 // It implements the [resolver.Resolver] interface.
 type delegatingResolver struct {
@@ -48,6 +50,9 @@ type delegatingResolver struct {
 	cc       resolver.ClientConn // gRPC ClientConn
 	proxyURL *url.URL            // proxy URL, derived from proxy environment and target
 
+	// We do not hold both mu and childMu in the same goroutine. Avoid holding
+	// both locks when calling into the child, as the child resolver may
+	// synchronously callback into the channel.
 	mu                  sync.Mutex         // protects all the fields below
 	targetResolverState *resolver.State    // state of the target resolver
 	proxyAddrs          []resolver.Address // resolved proxy addresses; empty if no proxy is configured
@@ -66,8 +71,8 @@ func (nopResolver) ResolveNow(resolver.ResolveNowOptions) {}
 
 func (nopResolver) Close() {}
 
-// proxyURLForTarget determines the proxy URL for the given address based on
-// the environment. It can return the following:
+// proxyURLForTarget determines the proxy URL for the given address based on the
+// environment. It can return the following:
 //   - nil URL, nil error: No proxy is configured or the address is excluded
 //     using the `NO_PROXY` environment variable or if req.URL.Host is
 //     "localhost" (with or without // a port number)
@@ -86,7 +91,8 @@ func proxyURLForTarget(address string) (*url.URL, error) {
 // resolvers:
 //   - one to resolve the proxy address specified using the supported
 //     environment variables. This uses the registered resolver for the "dns"
-//     scheme.
+//     scheme. It is lazily built when a target resolver update contains at least
+//     one TCP address.
 //   - one to resolve the target URI using the resolver specified by the scheme
 //     in the target URI or specified by the user using the WithResolvers dial
 //     option. As a special case, if the target URI's scheme is "dns" and a
@@ -95,8 +101,10 @@ func proxyURLForTarget(address string) (*url.URL, error) {
 //     resolution is enabled using the dial option.
 func New(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions, targetResolverBuilder resolver.Builder, targetResolutionEnabled bool) (resolver.Resolver, error) {
 	r := &delegatingResolver{
-		target: target,
-		cc:     cc,
+		target:         target,
+		cc:             cc,
+		proxyResolver:  nopResolver{},
+		targetResolver: nopResolver{},
 	}
 
 	var err error
@@ -123,37 +131,26 @@ func New(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOpti
 	// resolution should be handled by the proxy, not the client. Therefore, we
 	// bypass the target resolver and store the unresolved target address.
 	if target.URL.Scheme == "dns" && !targetResolutionEnabled {
-		state := resolver.State{
+		r.targetResolverState = &resolver.State{
 			Addresses: []resolver.Address{{Addr: target.Endpoint()}},
 			Endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: target.Endpoint()}}}},
 		}
-		r.targetResolverState = &state
-	} else {
-		wcc := &wrappingClientConn{
-			stateListener: r.updateTargetResolverState,
-			parent:        r,
-		}
-		if r.targetResolver, err = targetResolverBuilder.Build(target, wcc, opts); err != nil {
-			return nil, fmt.Errorf("delegating_resolver: unable to build the resolver for target %s: %v", target, err)
-		}
-	}
-
-	if r.proxyResolver, err = r.proxyURIResolver(opts); err != nil {
-		return nil, fmt.Errorf("delegating_resolver: failed to build resolver for proxy URL %q: %v", r.proxyURL, err)
+		r.updateTargetResolverState(*r.targetResolverState)
+		return r, nil
 	}
-
-	if r.targetResolver == nil {
-		r.targetResolver = nopResolver{}
+	wcc := &wrappingClientConn{
+		stateListener: r.updateTargetResolverState,
+		parent:        r,
 	}
-	if r.proxyResolver == nil {
-		r.proxyResolver = nopResolver{}
+	if r.targetResolver, err = targetResolverBuilder.Build(target, wcc, opts); err != nil {
+		return nil, fmt.Errorf("delegating_resolver: unable to build the resolver for target %s: %v", target, err)
 	}
 	return r, nil
 }
 
-// proxyURIResolver creates a resolver for resolving proxy URIs using the
-// "dns" scheme. It adjusts the proxyURL to conform to the "dns:///" format and
-// builds a resolver with a wrappingClientConn to capture resolved addresses.
+// proxyURIResolver creates a resolver for resolving proxy URIs using the "dns"
+// scheme. It adjusts the proxyURL to conform to the "dns:///" format and builds
+// a resolver with a wrappingClientConn to capture resolved addresses.
 func (r *delegatingResolver) proxyURIResolver(opts resolver.BuildOptions) (resolver.Resolver, error) {
 	proxyBuilder := resolver.Get("dns")
 	if proxyBuilder == nil {
@@ -189,18 +186,43 @@ func (r *delegatingResolver) Close() {
 	r.proxyResolver = nil
 }
 
-// updateClientConnStateLocked creates a list of combined addresses by
-// pairing each proxy address with every target address. For each pair, it
-// generates a new [resolver.Address] using the proxy address, and adding the
-// target address as the attribute along with user info. It returns nil if
-// either resolver has not sent update even once and returns the error from
-// ClientConn update once both resolvers have sent update atleast once.
+func networkTypeFromAddr(addr resolver.Address) string {
+	networkType, ok := networktype.Get(addr)
+	if !ok {
+		networkType, _ = transport.ParseDialTarget(addr.Addr)
+	}
+	return networkType
+}
+
+func isTCPAddressPresent(state *resolver.State) bool {
+	for _, addr := range state.Addresses {
+		if networkType := networkTypeFromAddr(addr); networkType == "tcp" {
+			return true
+		}
+	}
+	for _, endpoint := range state.Endpoints {
+		for _, addr := range endpoint.Addresses {
+			if networktype := networkTypeFromAddr(addr); networktype == "tcp" {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// updateClientConnStateLocked constructs a combined list of addresses by
+// pairing each proxy address with every target address of type TCP. For each
+// pair, it creates a new [resolver.Address] using the proxy address and
+// attaches the corresponding target address and user info as attributes. Target
+// addresses that are not of type TCP are appended to the list as-is. The
+// function returns nil if either resolver has not yet provided an update, and
+// returns the result of ClientConn.UpdateState once both resolvers have
+// provided at least one update.
 func (r *delegatingResolver) updateClientConnStateLocked() error {
 	if r.targetResolverState == nil || r.proxyAddrs == nil {
 		return nil
 	}
 
-	curState := *r.targetResolverState
 	// If multiple resolved proxy addresses are present, we send only the
 	// unresolved proxy host and let net.Dial handle the proxy host name
 	// resolution when creating the transport. Sending all resolved addresses
@@ -218,24 +240,30 @@ func (r *delegatingResolver) updateClientConnStateLocked() error {
 	}
 	var addresses []resolver.Address
 	for _, targetAddr := range (*r.targetResolverState).Addresses {
+		// Avoid proxy when network is not tcp.
+		if networkType := networkTypeFromAddr(targetAddr); networkType != "tcp" {
+			addresses = append(addresses, targetAddr)
+			continue
+		}
 		addresses = append(addresses, proxyattributes.Set(proxyAddr, proxyattributes.Options{
 			User:        r.proxyURL.User,
 			ConnectAddr: targetAddr.Addr,
 		}))
 	}
 
-	// Create a list of combined endpoints by pairing all proxy endpoints
-	// with every target endpoint. Each time, it constructs a new
-	// [resolver.Endpoint] using the all addresses from all the proxy endpoint
-	// and the target addresses from one endpoint. The target address and user
-	// information from the proxy URL are added as attributes to the proxy
-	// address.The resulting list of addresses is then grouped into endpoints,
-	// covering all combinations of proxy and target endpoints.
+	// For each target endpoint, construct a new [resolver.Endpoint] that
+	// includes all addresses from all proxy endpoints and the addresses from
+	// that target endpoint, preserving the number of target endpoints.
 	var endpoints []resolver.Endpoint
 	for _, endpt := range (*r.targetResolverState).Endpoints {
 		var addrs []resolver.Address
-		for _, proxyAddr := range r.proxyAddrs {
-			for _, targetAddr := range endpt.Addresses {
+		for _, targetAddr := range endpt.Addresses {
+			// Avoid proxy when network is not tcp.
+			if networkType := networkTypeFromAddr(targetAddr); networkType != "tcp" {
+				addrs = append(addrs, targetAddr)
+				continue
+			}
+			for _, proxyAddr := range r.proxyAddrs {
 				addrs = append(addrs, proxyattributes.Set(proxyAddr, proxyattributes.Options{
 					User:        r.proxyURL.User,
 					ConnectAddr: targetAddr.Addr,
@@ -246,8 +274,9 @@ func (r *delegatingResolver) updateClientConnStateLocked() error {
 	}
 	// Use the targetResolverState for its service config and attributes
 	// contents. The state update is only sent after both the target and proxy
-	// resolvers have sent their updates, and curState has been updated with
-	// the combined addresses.
+	// resolvers have sent their updates, and curState has been updated with the
+	// combined addresses.
+	curState := *r.targetResolverState
 	curState.Addresses = addresses
 	curState.Endpoints = endpoints
 	return r.cc.UpdateState(curState)
@@ -257,7 +286,8 @@ func (r *delegatingResolver) updateClientConnStateLocked() error {
 // addresses and endpoints, marking the resolver as ready, and triggering a
 // state update if both proxy and target resolvers are ready. If the ClientConn
 // returns a non-nil error, it calls `ResolveNow()` on the target resolver.  It
-// is a StateListener function of wrappingClientConn passed to the proxy resolver.
+// is a StateListener function of wrappingClientConn passed to the proxy
+// resolver.
 func (r *delegatingResolver) updateProxyResolverState(state resolver.State) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()
@@ -265,8 +295,8 @@ func (r *delegatingResolver) updateProxyResolverState(state resolver.State) erro
 		logger.Infof("Addresses received from proxy resolver: %s", state.Addresses)
 	}
 	if len(state.Endpoints) > 0 {
-		// We expect exactly one address per endpoint because the proxy
-		// resolver uses "dns" resolution.
+		// We expect exactly one address per endpoint because the proxy resolver
+		// uses "dns" resolution.
 		r.proxyAddrs = make([]resolver.Address, 0, len(state.Endpoints))
 		for _, endpoint := range state.Endpoints {
 			r.proxyAddrs = append(r.proxyAddrs, endpoint.Addresses...)
@@ -294,11 +324,14 @@ func (r *delegatingResolver) updateProxyResolverState(state resolver.State) erro
 	return err
 }
 
-// updateTargetResolverState updates the target resolver state by storing target
-// addresses, endpoints, and service config, marking the resolver as ready, and
-// triggering a state update if both resolvers are ready. If the ClientConn
-// returns a non-nil error, it calls `ResolveNow()` on the proxy resolver. It
-// is a StateListener function of wrappingClientConn passed to the target resolver.
+// updateTargetResolverState is the StateListener function provided to the
+// target resolver via wrappingClientConn. It updates the resolver state and
+// marks the target resolver as ready. If the update includes at least one TCP
+// address and the proxy resolver has not yet been constructed, it initializes
+// the proxy resolver. A combined state update is triggered once both resolvers
+// are ready. If all addresses are non-TCP, it proceeds without waiting for the
+// proxy resolver. If ClientConn.UpdateState returns a non-nil error,
+// ResolveNow() is called on the proxy resolver.
 func (r *delegatingResolver) updateTargetResolverState(state resolver.State) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()
@@ -307,6 +340,31 @@ func (r *delegatingResolver) updateTargetResolverState(state resolver.State) err
 		logger.Infof("Addresses received from target resolver: %v", state.Addresses)
 	}
 	r.targetResolverState = &state
+	// If no addresses returned by resolver have network type as tcp , do not
+	// wait for proxy update.
+	if !isTCPAddressPresent(r.targetResolverState) {
+		return r.cc.UpdateState(*r.targetResolverState)
+	}
+
+	// The proxy resolver may be rebuilt multiple times, specifically each time
+	// the target resolver sends an update, even if the target resolver is built
+	// successfully but building the proxy resolver fails.
+	if len(r.proxyAddrs) == 0 {
+		go func() {
+			r.childMu.Lock()
+			defer r.childMu.Unlock()
+			if _, ok := r.proxyResolver.(nopResolver); !ok {
+				return
+			}
+			proxyResolver, err := r.proxyURIResolver(resolver.BuildOptions{})
+			if err != nil {
+				r.cc.ReportError(fmt.Errorf("delegating_resolver: unable to build the proxy resolver: %v", err))
+				return
+			}
+			r.proxyResolver = proxyResolver
+		}()
+	}
+
 	err := r.updateClientConnStateLocked()
 	if err != nil {
 		go func() {
@@ -335,7 +393,8 @@ func (wcc *wrappingClientConn) UpdateState(state resolver.State) error {
 	return wcc.stateListener(state)
 }
 
-// ReportError intercepts errors from the child resolvers and passes them to ClientConn.
+// ReportError intercepts errors from the child resolvers and passes them to
+// ClientConn.
 func (wcc *wrappingClientConn) ReportError(err error) {
 	wcc.parent.cc.ReportError(err)
 }
@@ -346,8 +405,8 @@ func (wcc *wrappingClientConn) NewAddress(addrs []resolver.Address) {
 	wcc.UpdateState(resolver.State{Addresses: addrs})
 }
 
-// ParseServiceConfig parses the provided service config and returns an
-// object that provides the parsed config.
+// ParseServiceConfig parses the provided service config and returns an object
+// that provides the parsed config.
 func (wcc *wrappingClientConn) ParseServiceConfig(serviceConfigJSON string) *serviceconfig.ParseResult {
 	return wcc.parent.cc.ParseServiceConfig(serviceConfigJSON)
 }
diff --git a/internal/status/status.go b/internal/status/status.go
index 1186f1e9..aad171cd 100644
--- a/internal/status/status.go
+++ b/internal/status/status.go
@@ -236,3 +236,11 @@ func IsRestrictedControlPlaneCode(s *Status) bool {
 	}
 	return false
 }
+
+// RawStatusProto returns the internal protobuf message for use by gRPC itself.
+func RawStatusProto(s *Status) *spb.Status {
+	if s == nil {
+		return nil
+	}
+	return s.s
+}
diff --git a/internal/transport/http2_client.go b/internal/transport/http2_client.go
index 32047128..ed2d93af 100644
--- a/internal/transport/http2_client.go
+++ b/internal/transport/http2_client.go
@@ -176,7 +176,7 @@ func dial(ctx context.Context, fn func(context.Context, string) (net.Conn, error
 		return fn(ctx, address)
 	}
 	if !ok {
-		networkType, address = parseDialTarget(address)
+		networkType, address = ParseDialTarget(address)
 	}
 	if opts, present := proxyattributes.Get(addr); present {
 		return proxyDial(ctx, addr, grpcUA, opts)
@@ -1261,7 +1261,8 @@ func (t *http2Client) handleRSTStream(f *http2.RSTStreamFrame) {
 			statusCode = codes.DeadlineExceeded
 		}
 	}
-	t.closeStream(s, io.EOF, false, http2.ErrCodeNo, status.Newf(statusCode, "stream terminated by RST_STREAM with error code: %v", f.ErrCode), nil, false)
+	st := status.Newf(statusCode, "stream terminated by RST_STREAM with error code: %v", f.ErrCode)
+	t.closeStream(s, st.Err(), false, http2.ErrCodeNo, st, nil, false)
 }
 
 func (t *http2Client) handleSettings(f *http2.SettingsFrame, isFirst bool) {
diff --git a/internal/transport/http2_server.go b/internal/transport/http2_server.go
index 7e53eb17..e4c3731b 100644
--- a/internal/transport/http2_server.go
+++ b/internal/transport/http2_server.go
@@ -39,6 +39,7 @@ import (
 	"google.golang.org/grpc/internal/grpclog"
 	"google.golang.org/grpc/internal/grpcutil"
 	"google.golang.org/grpc/internal/pretty"
+	istatus "google.golang.org/grpc/internal/status"
 	"google.golang.org/grpc/internal/syscall"
 	"google.golang.org/grpc/mem"
 	"google.golang.org/protobuf/proto"
@@ -1055,7 +1056,7 @@ func (t *http2Server) writeHeaderLocked(s *ServerStream) error {
 	return nil
 }
 
-// WriteStatus sends stream status to the client and terminates the stream.
+// writeStatus sends stream status to the client and terminates the stream.
 // There is no further I/O operations being able to perform on this stream.
 // TODO(zhaoq): Now it indicates the end of entire stream. Revisit if early
 // OK is adopted.
@@ -1083,7 +1084,7 @@ func (t *http2Server) writeStatus(s *ServerStream, st *status.Status) error {
 	headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-status", Value: strconv.Itoa(int(st.Code()))})
 	headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-message", Value: encodeGrpcMessage(st.Message())})
 
-	if p := st.Proto(); p != nil && len(p.Details) > 0 {
+	if p := istatus.RawStatusProto(st); len(p.GetDetails()) > 0 {
 		// Do not use the user's grpc-status-details-bin (if present) if we are
 		// even attempting to set our own.
 		delete(s.trailer, grpcStatusDetailsBinHeader)
diff --git a/internal/transport/http_util.go b/internal/transport/http_util.go
index 3613d7b6..607d2c4c 100644
--- a/internal/transport/http_util.go
+++ b/internal/transport/http_util.go
@@ -196,11 +196,14 @@ func decodeTimeout(s string) (time.Duration, error) {
 	if !ok {
 		return 0, fmt.Errorf("transport: timeout unit is not recognized: %q", s)
 	}
-	t, err := strconv.ParseInt(s[:size-1], 10, 64)
+	t, err := strconv.ParseUint(s[:size-1], 10, 64)
 	if err != nil {
 		return 0, err
 	}
-	const maxHours = math.MaxInt64 / int64(time.Hour)
+	if t == 0 {
+		return 0, fmt.Errorf("transport: timeout must be positive: %q", s)
+	}
+	const maxHours = math.MaxInt64 / uint64(time.Hour)
 	if d == time.Hour && t > maxHours {
 		// This timeout would overflow math.MaxInt64; clamp it.
 		return time.Duration(math.MaxInt64), nil
@@ -439,8 +442,8 @@ func getWriteBufferPool(size int) *sync.Pool {
 	return pool
 }
 
-// parseDialTarget returns the network and address to pass to dialer.
-func parseDialTarget(target string) (string, string) {
+// ParseDialTarget returns the network and address to pass to dialer.
+func ParseDialTarget(target string) (string, string) {
 	net := "tcp"
 	m1 := strings.Index(target, ":")
 	m2 := strings.Index(target, ":/")
diff --git a/stream.go b/stream.go
index bbbcbc72..d58bb647 100644
--- a/stream.go
+++ b/stream.go
@@ -101,9 +101,9 @@ type ClientStream interface {
 	// It must only be called after stream.CloseAndRecv has returned, or
 	// stream.Recv has returned a non-nil error (including io.EOF).
 	Trailer() metadata.MD
-	// CloseSend closes the send direction of the stream. It closes the stream
-	// when non-nil error is met. It is also not safe to call CloseSend
-	// concurrently with SendMsg.
+	// CloseSend closes the send direction of the stream. This method always
+	// returns a nil error. The status of the stream may be discovered using
+	// RecvMsg. It is also not safe to call CloseSend concurrently with SendMsg.
 	CloseSend() error
 	// Context returns the context for this stream.
 	//
@@ -993,7 +993,7 @@ func (cs *clientStream) RecvMsg(m any) error {
 
 func (cs *clientStream) CloseSend() error {
 	if cs.sentLast {
-		// TODO: return an error and finish the stream instead, due to API misuse?
+		// Return a nil error on repeated calls to this method.
 		return nil
 	}
 	cs.sentLast = true
@@ -1014,7 +1014,10 @@ func (cs *clientStream) CloseSend() error {
 			binlog.Log(cs.ctx, chc)
 		}
 	}
-	// We never returned an error here for reasons.
+	// We don't return an error here as we expect users to read all messages
+	// from the stream and get the RPC status from RecvMsg().  Note that
+	// SendMsg() must return an error when one occurs so the application
+	// knows to stop sending messages, but that does not apply here.
 	return nil
 }
 
@@ -1378,7 +1381,7 @@ func (as *addrConnStream) Trailer() metadata.MD {
 
 func (as *addrConnStream) CloseSend() error {
 	if as.sentLast {
-		// TODO: return an error and finish the stream instead, due to API misuse?
+		// Return a nil error on repeated calls to this method.
 		return nil
 	}
 	as.sentLast = true
diff --git a/xds/internal/clients/config.go b/xds/internal/clients/config.go
index 0ededfae..f106465f 100644
--- a/xds/internal/clients/config.go
+++ b/xds/internal/clients/config.go
@@ -100,3 +100,13 @@ type Locality struct {
 	// SubZone is the further subdivision within a zone.
 	SubZone string
 }
+
+// MetricsReporter is used by the XDSClient to report metrics.
+type MetricsReporter interface {
+	// ReportMetric reports a metric. The metric will be one of the predefined
+	// set of types depending on the client (XDSClient or LRSClient).
+	//
+	// Each client will produce different metrics. Please see the client's
+	// documentation for a list of possible metrics events.
+	ReportMetric(metric any)
+}
diff --git a/xds/internal/clients/grpctransport/grpc_transport.go b/xds/internal/clients/grpctransport/grpc_transport.go
index e13c982a..af39349e 100644
--- a/xds/internal/clients/grpctransport/grpc_transport.go
+++ b/xds/internal/clients/grpctransport/grpc_transport.go
@@ -43,31 +43,41 @@ var (
 // It must be set by value (not pointer) in the
 // clients.ServerIdentifier.Extensions field (See Example).
 type ServerIdentifierExtension struct {
-	// Credentials is name of the credentials to use for this transport to the
-	// server. It must be present in the map passed to NewBuilder.
-	Credentials string
+	// ConfigName is the name of the configuration to use for this transport.
+	// It must be present as a key in the map of configs passed to NewBuilder.
+	ConfigName string
 }
 
 // Builder creates gRPC-based Transports. It must be paired with ServerIdentifiers
 // that contain an Extension field of type ServerIdentifierExtension.
 type Builder struct {
-	// credentials is a map of credentials names to credentials.Bundle which
-	// can be used to connect to the server.
-	credentials map[string]credentials.Bundle
+	// configs is a map of configuration names to their respective Config.
+	configs map[string]Config
 
 	mu sync.Mutex
 	// connections is a map of clients.ServerIdentifiers in use by the Builder
 	// to connect to different servers.
 	connections map[clients.ServerIdentifier]*grpc.ClientConn
-	refs        map[clients.ServerIdentifier]int
+	// refs tracks the number of active references to each connection.
+	refs map[clients.ServerIdentifier]int
+}
+
+// Config defines the configuration for connecting to a gRPC server, including
+// credentials and an optional custom new client function.
+type Config struct {
+	// Credentials is the credentials bundle to be used for the connection.
+	Credentials credentials.Bundle
+	// GRPCNewClient is an optional custom function to establish a gRPC connection.
+	// If nil, grpc.NewClient will be used as the default.
+	GRPCNewClient func(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error)
 }
 
 // NewBuilder provides a builder for creating gRPC-based Transports using
 // the credentials from provided map of credentials names to
 // credentials.Bundle.
-func NewBuilder(credentials map[string]credentials.Bundle) *Builder {
+func NewBuilder(configs map[string]Config) *Builder {
 	return &Builder{
-		credentials: credentials,
+		configs:     configs,
 		connections: make(map[clients.ServerIdentifier]*grpc.ClientConn),
 		refs:        make(map[clients.ServerIdentifier]int),
 	}
@@ -88,9 +98,12 @@ func (b *Builder) Build(si clients.ServerIdentifier) (clients.Transport, error)
 		return nil, fmt.Errorf("grpctransport: Extensions field is %T, but must be %T in ServerIdentifier", si.Extensions, ServerIdentifierExtension{})
 	}
 
-	creds, ok := b.credentials[sce.Credentials]
+	config, ok := b.configs[sce.ConfigName]
 	if !ok {
-		return nil, fmt.Errorf("grpctransport: unknown credentials type %q specified in extensions", sce.Credentials)
+		return nil, fmt.Errorf("grpctransport: unknown config name %q specified in ServerIdentifierExtension", sce.ConfigName)
+	}
+	if config.Credentials == nil {
+		return nil, fmt.Errorf("grpctransport: config %q has nil credentials bundle", sce.ConfigName)
 	}
 
 	b.mu.Lock()
@@ -114,7 +127,12 @@ func (b *Builder) Build(si clients.ServerIdentifier) (clients.Transport, error)
 		Time:    5 * time.Minute,
 		Timeout: 20 * time.Second,
 	})
-	cc, err := grpc.NewClient(si.ServerURI, kpCfg, grpc.WithCredentialsBundle(creds), grpc.WithDefaultCallOptions(grpc.ForceCodec(&byteCodec{})))
+	dopts := []grpc.DialOption{kpCfg, grpc.WithCredentialsBundle(config.Credentials), grpc.WithDefaultCallOptions(grpc.ForceCodec(&byteCodec{}))}
+	newClientFunc := grpc.NewClient
+	if config.GRPCNewClient != nil {
+		newClientFunc = config.GRPCNewClient
+	}
+	cc, err := newClientFunc(si.ServerURI, dopts...)
 	if err != nil {
 		return nil, fmt.Errorf("grpctransport: failed to create connection to server %q: %v", si.ServerURI, err)
 	}
diff --git a/xds/internal/clients/xdsclient/authority.go b/xds/internal/clients/xdsclient/authority.go
index 3ada1465..7a3a2969 100644
--- a/xds/internal/clients/xdsclient/authority.go
+++ b/xds/internal/clients/xdsclient/authority.go
@@ -26,8 +26,10 @@ import (
 
 	"google.golang.org/grpc/grpclog"
 	igrpclog "google.golang.org/grpc/internal/grpclog"
+	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/internal/syncutil"
 	"google.golang.org/grpc/xds/internal/clients/xdsclient/internal/xdsresource"
+	"google.golang.org/grpc/xds/internal/clients/xdsclient/metrics"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
@@ -87,6 +89,7 @@ type authority struct {
 	xdsClientSerializerClose  func()                       // Function to close the above serializer.
 	logger                    *igrpclog.PrefixLogger       // Logger for this authority.
 	target                    string                       // The gRPC Channel target.
+	metricsReporter           clients.MetricsReporter
 
 	// The below defined fields must only be accessed in the context of the
 	// serializer callback, owned by this authority.
@@ -121,6 +124,7 @@ type authorityBuildOptions struct {
 	getChannelForADS xdsChannelForADS             // Function to acquire a reference to an xdsChannel
 	logPrefix        string                       // Prefix for logging
 	target           string                       // Target for the gRPC Channel that owns xDS Client/Authority
+	metricsReporter  clients.MetricsReporter      // Metrics reporter for the authority
 }
 
 // newAuthority creates a new authority instance with the provided
@@ -145,6 +149,7 @@ func newAuthority(args authorityBuildOptions) *authority {
 		logger:                    igrpclog.NewPrefixLogger(l, logPrefix),
 		resources:                 make(map[ResourceType]map[string]*resourceState),
 		target:                    args.target,
+		metricsReporter:           args.metricsReporter,
 	}
 
 	// Create an ordered list of xdsChannels with their server configs. The
@@ -363,6 +368,11 @@ func (a *authority) handleADSResourceUpdate(serverConfig *ServerConfig, rType Re
 		// On error, keep previous version of the resource. But update status
 		// and error.
 		if uErr.Err != nil {
+			if a.metricsReporter != nil {
+				a.metricsReporter.ReportMetric(&metrics.ResourceUpdateInvalid{
+					ServerURI: serverConfig.ServerIdentifier.ServerURI, ResourceType: rType.TypeName,
+				})
+			}
 			state.md.ErrState = md.ErrState
 			state.md.Status = md.Status
 			for watcher := range state.watchers {
@@ -378,6 +388,12 @@ func (a *authority) handleADSResourceUpdate(serverConfig *ServerConfig, rType Re
 			continue
 		}
 
+		if a.metricsReporter != nil {
+			a.metricsReporter.ReportMetric(&metrics.ResourceUpdateValid{
+				ServerURI: serverConfig.ServerIdentifier.ServerURI, ResourceType: rType.TypeName,
+			})
+		}
+
 		if state.deletionIgnored {
 			state.deletionIgnored = false
 			a.logger.Infof("A valid update was received for resource %q of type %q after previously ignoring a deletion", name, rType.TypeName)
diff --git a/xds/internal/clients/xdsclient/metrics/metrics.go b/xds/internal/clients/xdsclient/metrics/metrics.go
new file mode 100644
index 00000000..2e14efb1
--- /dev/null
+++ b/xds/internal/clients/xdsclient/metrics/metrics.go
@@ -0,0 +1,42 @@
+/*
+ *
+ * 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 metrics defines all metrics that can be produced by an xDS client.
+// All calls to the MetricsRecorder by the xDS client will contain a struct
+// from this package passed by pointer.
+package metrics
+
+// ResourceUpdateValid is a metric to report a valid resource update from
+// the xDS management server for a given resource type.
+type ResourceUpdateValid struct {
+	ServerURI    string
+	ResourceType string
+}
+
+// ResourceUpdateInvalid is a metric to report an invalid resource update
+// from the xDS management server for a given resource type.
+type ResourceUpdateInvalid struct {
+	ServerURI    string
+	ResourceType string
+}
+
+// ServerFailure is a metric to report a server failure of the xDS
+// management server.
+type ServerFailure struct {
+	ServerURI string
+}
diff --git a/xds/internal/clients/xdsclient/xdsclient.go b/xds/internal/clients/xdsclient/xdsclient.go
index f0f4a678..415a2a1e 100644
--- a/xds/internal/clients/xdsclient/xdsclient.go
+++ b/xds/internal/clients/xdsclient/xdsclient.go
@@ -44,6 +44,7 @@ import (
 	"google.golang.org/grpc/xds/internal/clients/internal/backoff"
 	"google.golang.org/grpc/xds/internal/clients/internal/syncutil"
 	"google.golang.org/grpc/xds/internal/clients/xdsclient/internal/xdsresource"
+	"google.golang.org/grpc/xds/internal/clients/xdsclient/metrics"
 	"google.golang.org/protobuf/proto"
 
 	v3statuspb "github.com/envoyproxy/go-control-plane/envoy/service/status/v3"
@@ -82,8 +83,9 @@ type XDSClient struct {
 	resourceTypes      map[string]ResourceType      // Registry of resource types, for parsing incoming ADS responses.
 	serializer         *syncutil.CallbackSerializer // Serializer for invoking resource watcher callbacks.
 	serializerClose    func()                       // Function to close the serializer.
-	logger             *grpclog.PrefixLogger        // Logger for this client.
-	target             string                       // The gRPC target for this client.
+	logger             *grpclog.PrefixLogger
+	target             string
+	metricsReporter    clients.MetricsReporter
 
 	// The XDSClient owns a bunch of channels to individual xDS servers
 	// specified in the xDS client configuration. Authorities acquire references
@@ -138,6 +140,7 @@ func newClient(config *Config, watchExpiryTimeout time.Duration, streamBackoff f
 		transportBuilder:   config.TransportBuilder,
 		resourceTypes:      config.ResourceTypes,
 		xdsActiveChannels:  make(map[ServerConfig]*channelState),
+		metricsReporter:    config.MetricsReporter,
 	}
 
 	for name, cfg := range config.Authorities {
@@ -154,6 +157,7 @@ func newClient(config *Config, watchExpiryTimeout time.Duration, streamBackoff f
 			getChannelForADS: c.getChannelForADS,
 			logPrefix:        clientPrefix(c),
 			target:           target,
+			metricsReporter:  c.metricsReporter,
 		})
 	}
 	c.topLevelAuthority = newAuthority(authorityBuildOptions{
@@ -163,8 +167,10 @@ func newClient(config *Config, watchExpiryTimeout time.Duration, streamBackoff f
 		getChannelForADS: c.getChannelForADS,
 		logPrefix:        clientPrefix(c),
 		target:           target,
+		metricsReporter:  c.metricsReporter,
 	})
 	c.logger = prefixLogger(c)
+
 	return c, nil
 }
 
@@ -384,6 +390,12 @@ func (cs *channelState) adsStreamFailure(err error) {
 		return
 	}
 
+	if xdsresource.ErrType(err) != xdsresource.ErrTypeStreamFailedAfterRecv && cs.parent.metricsReporter != nil {
+		cs.parent.metricsReporter.ReportMetric(&metrics.ServerFailure{
+			ServerURI: cs.serverConfig.ServerIdentifier.ServerURI,
+		})
+	}
+
 	cs.parent.channelsMu.Lock()
 	defer cs.parent.channelsMu.Unlock()
 	for authority := range cs.interestedAuthorities {
diff --git a/xds/internal/clients/xdsclient/xdsconfig.go b/xds/internal/clients/xdsclient/xdsconfig.go
index 4c7d090d..a276309d 100644
--- a/xds/internal/clients/xdsclient/xdsconfig.go
+++ b/xds/internal/clients/xdsclient/xdsconfig.go
@@ -56,6 +56,10 @@ type Config struct {
 	//
 	// For example: "type.googleapis.com/envoy.config.listener.v3.Listener"
 	ResourceTypes map[string]ResourceType
+
+	// MetricsReporter is used to report registered metrics. If unset, no
+	// metrics will be reported.
+	MetricsReporter clients.MetricsReporter
 }
 
 // ServerConfig contains configuration for an xDS management server.



Test patch:
diff --git a/balancer/endpointsharding/endpointsharding_test.go b/balancer/endpointsharding/endpointsharding_test.go
index ebad977e..5567b4c9 100644
--- a/balancer/endpointsharding/endpointsharding_test.go
+++ b/balancer/endpointsharding/endpointsharding_test.go
@@ -23,7 +23,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"log"
 	"strings"
 	"testing"
 	"time"
@@ -166,7 +165,7 @@ func (s) TestEndpointShardingBasic(t *testing.T) {
 	}
 	cc, err := grpc.NewClient(mr.Scheme()+":///", dOpts...)
 	if err != nil {
-		log.Fatalf("Failed to create new client: %v", err)
+		t.Fatalf("Failed to create new client: %v", err)
 	}
 	defer cc.Close()
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
@@ -246,7 +245,7 @@ func (s) TestEndpointShardingReconnectDisabled(t *testing.T) {
 
 	cc, err := grpc.NewClient(mr.Scheme()+":///", grpc.WithResolvers(mr), grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
-		log.Fatalf("Failed to create new client: %v", err)
+		t.Fatalf("Failed to create new client: %v", err)
 	}
 	defer cc.Close()
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
diff --git a/credentials/credentials_ext_test.go b/credentials/credentials_ext_test.go
index 27da3a26..f1733358 100644
--- a/credentials/credentials_ext_test.go
+++ b/credentials/credentials_ext_test.go
@@ -30,6 +30,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/credentials/local"
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
@@ -57,45 +58,88 @@ func authorityChecker(ctx context.Context, wantAuthority string) error {
 	return nil
 }
 
-// Tests the `grpc.CallAuthority` option with TLS credentials. This test verifies
-// that the provided authority is correctly propagated to the server when a
-// correct authority is used.
-func (s) TestCorrectAuthorityWithTLSCreds(t *testing.T) {
+func loadTLSCreds(t *testing.T) (grpc.ServerOption, grpc.DialOption) {
+	t.Helper()
 	cert, err := tls.LoadX509KeyPair(testdata.Path("x509/server1_cert.pem"), testdata.Path("x509/server1_key.pem"))
 	if err != nil {
-		t.Fatalf("Failed to load key pair: %s", err)
+		t.Fatalf("Failed to load key pair: %v", err)
+		return nil, nil
 	}
-	creds, err := credentials.NewClientTLSFromFile(testdata.Path("x509/server_ca_cert.pem"), "x.test.example.com")
+	serverCreds := grpc.Creds(credentials.NewServerTLSFromCert(&cert))
+
+	clientCreds, err := credentials.NewClientTLSFromFile(testdata.Path("x509/server_ca_cert.pem"), "x.test.example.com")
 	if err != nil {
-		t.Fatalf("Failed to create credentials %v", err)
+		t.Fatalf("Failed to create client credentials: %v", err)
 	}
+	return serverCreds, grpc.WithTransportCredentials(clientCreds)
+}
+
+// Tests the scenario where the `grpc.CallAuthority` call option is used with
+// different transport credentials. The test verifies that the specified
+// authority is correctly propagated to the serve when a correct authority is
+// used.
+func (s) TestCorrectAuthorityWithCreds(t *testing.T) {
 	const authority = "auth.test.example.com"
-	ss := &stubserver.StubServer{
-		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
-			if err := authorityChecker(ctx, authority); err != nil {
-				return nil, err
-			}
-			return &testpb.Empty{}, nil
+
+	tests := []struct {
+		name         string
+		creds        func(t *testing.T) (grpc.ServerOption, grpc.DialOption)
+		expectedAuth string
+	}{
+		{
+			name: "Insecure",
+			creds: func(t *testing.T) (grpc.ServerOption, grpc.DialOption) {
+				c := insecure.NewCredentials()
+				return grpc.Creds(c), grpc.WithTransportCredentials(c)
+			},
+			expectedAuth: authority,
+		},
+		{
+			name: "Local",
+			creds: func(t *testing.T) (grpc.ServerOption, grpc.DialOption) {
+				c := local.NewCredentials()
+				return grpc.Creds(c), grpc.WithTransportCredentials(c)
+			},
+			expectedAuth: authority,
+		},
+		{
+			name: "TLS",
+			creds: func(t *testing.T) (grpc.ServerOption, grpc.DialOption) {
+				return loadTLSCreds(t)
+			},
+			expectedAuth: authority,
 		},
 	}
-	if err := ss.StartServer(grpc.Creds(credentials.NewServerTLSFromCert(&cert))); err != nil {
-		t.Fatalf("Error starting endpoint server: %v", err)
-	}
-	defer ss.Stop()
 
-	cc, err := grpc.NewClient(ss.Address, grpc.WithTransportCredentials(creds))
-	if err != nil {
-		t.Fatalf("grpc.NewClient(%q) = %v", ss.Address, err)
-	}
-	defer cc.Close()
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ss := &stubserver.StubServer{
+				EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+					if err := authorityChecker(ctx, tt.expectedAuth); err != nil {
+						return nil, err
+					}
+					return &testpb.Empty{}, nil
+				},
+			}
+			serverOpt, dialOpt := tt.creds(t)
+			if err := ss.StartServer(serverOpt); err != nil {
+				t.Fatalf("Error starting endpoint server: %v", err)
+			}
+			defer ss.Stop()
 
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
+			cc, err := grpc.NewClient(ss.Address, dialOpt)
+			if err != nil {
+				t.Fatalf("grpc.NewClient(%q) = %v", ss.Address, err)
+			}
+			defer cc.Close()
 
-	if _, err = testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(authority)); status.Code(err) != codes.OK {
-		t.Fatalf("EmptyCall() returned status %v, want %v", status.Code(err), codes.OK)
+			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+			defer cancel()
+			if _, err = testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(tt.expectedAuth)); err != nil {
+				t.Fatalf("EmptyCall() rpc failed: %v", err)
+			}
+		})
 	}
-
 }
 
 // Tests the `grpc.CallAuthority` option with TLS credentials. This test verifies
@@ -143,38 +187,6 @@ func (s) TestIncorrectAuthorityWithTLS(t *testing.T) {
 	}
 }
 
-// Tests the scenario where the `grpc.CallAuthority` call option is used with
-// insecure transport credentials. The test verifies that the specified
-// authority is correctly propagated to the server.
-func (s) TestAuthorityCallOptionWithInsecureCreds(t *testing.T) {
-	const authority = "test.server.name"
-
-	ss := &stubserver.StubServer{
-		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
-			if err := authorityChecker(ctx, authority); err != nil {
-				return nil, err
-			}
-			return &testpb.Empty{}, nil
-		},
-	}
-	if err := ss.Start(nil); err != nil {
-		t.Fatalf("Error starting endpoint server: %v", err)
-	}
-	defer ss.Stop()
-
-	cc, err := grpc.NewClient(ss.Address, grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		t.Fatalf("grpc.NewClient(%q) = %v", ss.Address, err)
-	}
-	defer cc.Close()
-
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	if _, err = testgrpc.NewTestServiceClient(cc).EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(authority)); err != nil {
-		t.Fatalf("EmptyCall() rpc failed: %v", err)
-	}
-}
-
 // testAuthInfoNoValidator implements only credentials.AuthInfo and not
 // credentials.AuthorityValidator.
 type testAuthInfoNoValidator struct{}
diff --git a/health/server_internal_test.go b/health/server_internal_test.go
index 349eae23..d31c917e 100644
--- a/health/server_internal_test.go
+++ b/health/server_internal_test.go
@@ -19,10 +19,18 @@
 package health
 
 import (
+	"context"
+	"errors"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/testing/protocmp"
+
 	healthpb "google.golang.org/grpc/health/grpc_health_v1"
 	"google.golang.org/grpc/internal/grpctest"
 )
@@ -81,3 +89,65 @@ func (s) TestShutdown(t *testing.T) {
 		t.Fatalf("status for %s is %v, want %v", testService, status, healthpb.HealthCheckResponse_NOT_SERVING)
 	}
 }
+
+// TestList verifies that List() returns the health status of all the services if no. of services are within
+// maxAllowedLimits.
+func (s) TestList(t *testing.T) {
+	s := NewServer()
+
+	// Fill out status map with information
+	const length = 3
+	for i := 0; i < length; i++ {
+		s.SetServingStatus(fmt.Sprintf("%d", i),
+			healthpb.HealthCheckResponse_ServingStatus(i))
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	var in healthpb.HealthListRequest
+	got, err := s.List(ctx, &in)
+
+	if err != nil {
+		t.Fatalf("s.List(ctx, &in) returned err %v, want nil", err)
+	}
+	if len(got.GetStatuses()) != length+1 {
+		t.Fatalf("len(out.GetStatuses()) = %d, want %d",
+			len(got.GetStatuses()), length+1)
+	}
+	want := &healthpb.HealthListResponse{
+		Statuses: map[string]*healthpb.HealthCheckResponse{
+			"":  {Status: healthpb.HealthCheckResponse_SERVING},
+			"0": {Status: healthpb.HealthCheckResponse_UNKNOWN},
+			"1": {Status: healthpb.HealthCheckResponse_SERVING},
+			"2": {Status: healthpb.HealthCheckResponse_NOT_SERVING},
+		},
+	}
+	if diff := cmp.Diff(got, want, protocmp.Transform()); diff != "" {
+		t.Fatalf("Health response did not match expectation.  Diff (-got, +want): %s", diff)
+	}
+}
+
+// TestListResourceExhausted verifies that List()
+// returns a ResourceExhausted error if no. of services are more than
+// maxAllowedServices.
+func (s) TestListResourceExhausted(t *testing.T) {
+	s := NewServer()
+
+	// Fill out status map with service information,
+	// 101 (100 + 1 existing) elements will trigger an error.
+	for i := 1; i <= maxAllowedServices; i++ {
+		s.SetServingStatus(fmt.Sprintf("%d", i),
+			healthpb.HealthCheckResponse_SERVING)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	var in healthpb.HealthListRequest
+	_, err := s.List(ctx, &in)
+
+	want := status.Errorf(codes.ResourceExhausted,
+		"server health list exceeds maximum capacity: %d", maxAllowedServices)
+	if !errors.Is(err, want) {
+		t.Fatalf("s.List(ctx, &in) returned %v, want %v", err, want)
+	}
+}
diff --git a/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go b/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go
index 148b4481..b7176d5f 100644
--- a/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go
+++ b/internal/resolver/delegatingresolver/delegatingresolver_ext_test.go
@@ -31,6 +31,7 @@ import (
 	"google.golang.org/grpc/internal/proxyattributes"
 	"google.golang.org/grpc/internal/resolver/delegatingresolver"
 	"google.golang.org/grpc/internal/testutils"
+	"google.golang.org/grpc/internal/transport/networktype"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
 	"google.golang.org/grpc/serviceconfig"
@@ -130,15 +131,29 @@ func (s) TestDelegatingResolverNoProxyEnvVarsSet(t *testing.T) {
 // overwriting the previously registered DNS resolver. This allows the test to
 // mock the DNS resolution for the proxy resolver. It also registers the
 // original DNS resolver after the test is done.
-func setupDNS(t *testing.T) *manual.Resolver {
+func setupDNS(t *testing.T) (*manual.Resolver, chan struct{}) {
 	t.Helper()
 	mr := manual.NewBuilderWithScheme("dns")
 
 	dnsResolverBuilder := resolver.Get("dns")
 	resolver.Register(mr)
 
+	resolverBuilt := make(chan struct{})
+	mr.BuildCallback = func(resolver.Target, resolver.ClientConn, resolver.BuildOptions) {
+		close(resolverBuilt)
+	}
+
 	t.Cleanup(func() { resolver.Register(dnsResolverBuilder) })
-	return mr
+	return mr, resolverBuilt
+}
+
+func mustBuildResolver(ctx context.Context, t *testing.T, buildCh chan struct{}) {
+	t.Helper()
+	select {
+	case <-buildCh:
+	case <-ctx.Done():
+		t.Fatalf("Context timed out waiting for resolver to be built.")
+	}
 }
 
 // proxyAddressWithTargetAttribute creates a resolver.Address for the proxy,
@@ -180,15 +195,18 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithTargetResolution(t *testing.T)
 	targetResolver := manual.NewBuilderWithScheme("dns")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, true); err != nil {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
-	proxyResolver.UpdateState(resolver.State{
-		Addresses:     []resolver.Address{{Addr: resolvedProxyTestAddr1}},
+	targetResolver.UpdateState(resolver.State{
+		Addresses: []resolver.Address{
+			{Addr: resolvedTargetTestAddr1},
+			{Addr: resolvedTargetTestAddr2},
+		},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
 
@@ -198,11 +216,13 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithTargetResolution(t *testing.T)
 	case <-time.After(defaultTestShortTimeout):
 	}
 
-	targetResolver.UpdateState(resolver.State{
-		Addresses: []resolver.Address{
-			{Addr: resolvedTargetTestAddr1},
-			{Addr: resolvedTargetTestAddr2},
-		},
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
+	proxyResolver.UpdateState(resolver.State{
+		Addresses:     []resolver.Address{{Addr: resolvedProxyTestAddr1}},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
 
@@ -217,8 +237,8 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithTargetResolution(t *testing.T)
 	var gotState resolver.State
 	select {
 	case gotState = <-stateCh:
-	case <-time.After(defaultTestTimeout):
-		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	case <-ctx.Done():
+		t.Fatal("Context timeed out when waiting for a state update from the delegating resolver")
 	}
 
 	if diff := cmp.Diff(gotState, wantState); diff != "" {
@@ -256,13 +276,18 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithNoTargetResolution(t *testing.
 	targetResolver := manual.NewBuilderWithScheme("dns")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
 	proxyResolver.UpdateState(resolver.State{
 		Addresses: []resolver.Address{
 			{Addr: resolvedProxyTestAddr1},
@@ -277,8 +302,8 @@ func (s) TestDelegatingResolverwithDNSAndProxyWithNoTargetResolution(t *testing.
 	var gotState resolver.State
 	select {
 	case gotState = <-stateCh:
-	case <-time.After(defaultTestTimeout):
-		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	case <-ctx.Done():
+		t.Fatal("Context timed out when waiting for a state update from the delegating resolver")
 	}
 
 	if diff := cmp.Diff(gotState, wantState); diff != "" {
@@ -318,15 +343,18 @@ func (s) TestDelegatingResolverwithCustomResolverAndProxy(t *testing.T) {
 	targetResolver := manual.NewBuilderWithScheme("test")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
-	proxyResolver.UpdateState(resolver.State{
-		Addresses:     []resolver.Address{{Addr: resolvedProxyTestAddr1}},
+	targetResolver.UpdateState(resolver.State{
+		Addresses: []resolver.Address{
+			{Addr: resolvedTargetTestAddr1},
+			{Addr: resolvedTargetTestAddr2},
+		},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
 
@@ -336,11 +364,13 @@ func (s) TestDelegatingResolverwithCustomResolverAndProxy(t *testing.T) {
 	case <-time.After(defaultTestShortTimeout):
 	}
 
-	targetResolver.UpdateState(resolver.State{
-		Addresses: []resolver.Address{
-			{Addr: resolvedTargetTestAddr1},
-			{Addr: resolvedTargetTestAddr2},
-		},
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
+	proxyResolver.UpdateState(resolver.State{
+		Addresses:     []resolver.Address{{Addr: resolvedProxyTestAddr1}},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
 
@@ -354,8 +384,8 @@ func (s) TestDelegatingResolverwithCustomResolverAndProxy(t *testing.T) {
 	var gotState resolver.State
 	select {
 	case gotState = <-stateCh:
-	case <-time.After(defaultTestTimeout):
-		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	case <-ctx.Done():
+		t.Fatal("Context timed out when waiting for a state update from the delegating resolver")
 	}
 
 	if diff := cmp.Diff(gotState, wantState); diff != "" {
@@ -400,26 +430,13 @@ func (s) TestDelegatingResolverForEndpointsWithProxy(t *testing.T) {
 	targetResolver := manual.NewBuilderWithScheme("test")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
-	proxyResolver.UpdateState(resolver.State{
-		Endpoints: []resolver.Endpoint{
-			{Addresses: []resolver.Address{{Addr: resolvedProxyTestAddr1}}},
-			{Addresses: []resolver.Address{{Addr: resolvedProxyTestAddr2}}},
-		},
-		ServiceConfig: &serviceconfig.ParseResult{},
-	})
-
-	select {
-	case <-stateCh:
-		t.Fatalf("Delegating resolver invoked UpdateState before both the proxy and target resolvers had updated their states.")
-	case <-time.After(defaultTestShortTimeout):
-	}
 	targetResolver.UpdateState(resolver.State{
 		Endpoints: []resolver.Endpoint{
 			{
@@ -435,22 +452,39 @@ func (s) TestDelegatingResolverForEndpointsWithProxy(t *testing.T) {
 		},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
+	select {
+	case <-stateCh:
+		t.Fatalf("Delegating resolver invoked UpdateState before both the proxy and target resolvers had updated their states.")
+	case <-time.After(defaultTestShortTimeout):
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
 
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
+	proxyResolver.UpdateState(resolver.State{
+		Endpoints: []resolver.Endpoint{
+			{Addresses: []resolver.Address{{Addr: resolvedProxyTestAddr1}}},
+			{Addresses: []resolver.Address{{Addr: resolvedProxyTestAddr2}}},
+		},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	})
 	wantState := resolver.State{
 		Endpoints: []resolver.Endpoint{
 			{
 				Addresses: []resolver.Address{
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr1),
-					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr2),
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr2, resolvedTargetTestAddr1),
+					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr2),
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr2, resolvedTargetTestAddr2),
 				},
 			},
 			{
 				Addresses: []resolver.Address{
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr3),
-					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr4),
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr2, resolvedTargetTestAddr3),
+					proxyAddressWithTargetAttribute(resolvedProxyTestAddr1, resolvedTargetTestAddr4),
 					proxyAddressWithTargetAttribute(resolvedProxyTestAddr2, resolvedTargetTestAddr4),
 				},
 			},
@@ -460,8 +494,8 @@ func (s) TestDelegatingResolverForEndpointsWithProxy(t *testing.T) {
 	var gotState resolver.State
 	select {
 	case gotState = <-stateCh:
-	case <-time.After(defaultTestTimeout):
-		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	case <-ctx.Done():
+		t.Fatal("Contex timed out when waiting for a state update from the delegating resolver")
 	}
 
 	if diff := cmp.Diff(gotState, wantState); diff != "" {
@@ -474,7 +508,7 @@ func (s) TestDelegatingResolverForEndpointsWithProxy(t *testing.T) {
 // The test verifies that the delegating resolver combines unresolved proxy
 // host and target addresses correctly, returning addresses with the proxy host
 // populated and the target address included as an attribute.
-func (s) TestDelegatingResolverForMutipleProxyAddress(t *testing.T) {
+func (s) TestDelegatingResolverForMultipleProxyAddress(t *testing.T) {
 	const (
 		targetTestAddr          = "test.com"
 		resolvedTargetTestAddr1 = "1.1.1.1:8080"
@@ -503,17 +537,16 @@ func (s) TestDelegatingResolverForMutipleProxyAddress(t *testing.T) {
 	targetResolver := manual.NewBuilderWithScheme("test")
 	target := targetResolver.Scheme() + ":///" + targetTestAddr
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
-
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 	tcc, stateCh, _ := createTestResolverClientConn(t)
 	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
-	proxyResolver.UpdateState(resolver.State{
+	targetResolver.UpdateState(resolver.State{
 		Addresses: []resolver.Address{
-			{Addr: resolvedProxyTestAddr1},
-			{Addr: resolvedProxyTestAddr2},
+			{Addr: resolvedTargetTestAddr1},
+			{Addr: resolvedTargetTestAddr2},
 		},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
@@ -524,10 +557,15 @@ func (s) TestDelegatingResolverForMutipleProxyAddress(t *testing.T) {
 	case <-time.After(defaultTestShortTimeout):
 	}
 
-	targetResolver.UpdateState(resolver.State{
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
+	proxyResolver.UpdateState(resolver.State{
 		Addresses: []resolver.Address{
-			{Addr: resolvedTargetTestAddr1},
-			{Addr: resolvedTargetTestAddr2},
+			{Addr: resolvedProxyTestAddr1},
+			{Addr: resolvedProxyTestAddr2},
 		},
 		ServiceConfig: &serviceconfig.ParseResult{},
 	})
@@ -542,8 +580,8 @@ func (s) TestDelegatingResolverForMutipleProxyAddress(t *testing.T) {
 	var gotState resolver.State
 	select {
 	case gotState = <-stateCh:
-	case <-time.After(defaultTestTimeout):
-		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	case <-ctx.Done():
+		t.Fatal("Context timed out when waiting for a state update from the delegating resolver")
 	}
 
 	if diff := cmp.Diff(gotState, wantState); diff != "" {
@@ -586,9 +624,9 @@ func (s) TestDelegatingResolverUpdateStateDuringClose(t *testing.T) {
 
 	target := targetResolver.Scheme() + ":///" + "ignored"
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
-	unblockProxyResolverClose := make(chan struct{})
+	unblockProxyResolverClose := make(chan struct{}, 1)
 	proxyResolver.CloseCallback = func() {
 		<-unblockProxyResolverClose
 		t.Log("Proxy resolver is closed.")
@@ -607,11 +645,13 @@ func (s) TestDelegatingResolverUpdateStateDuringClose(t *testing.T) {
 		Endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "1.1.1.1"}}}},
 	})
 
+	// Wait for the proxy resolver to be built before calling Close.
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
 	// Closing the delegating resolver will block until the test writes to the
 	// unblockProxyResolverClose channel.
 	go dr.Close()
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	select {
 	case <-targetResolverCloseCalled:
 	case <-ctx.Done():
@@ -667,7 +707,7 @@ func (s) TestDelegatingResolverUpdateStateFromResolveNow(t *testing.T) {
 
 	target := targetResolver.Scheme() + ":///" + "ignored"
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
 
 	tcc, _, _ := createTestResolverClientConn(t)
 	tcc.UpdateStateF = func(resolver.State) error {
@@ -682,14 +722,18 @@ func (s) TestDelegatingResolverUpdateStateFromResolveNow(t *testing.T) {
 		Endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "1.1.1.1"}}}},
 	})
 
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
+
 	// Updating the channel will result in an error being returned. The
 	// delegating resolver should call call "ResolveNow" on the target resolver.
 	proxyResolver.UpdateState(resolver.State{
 		Endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "1.1.1.1"}}}},
 	})
 
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	select {
 	case <-targetResolverCalled:
 	case <-ctx.Done():
@@ -716,18 +760,19 @@ func (s) TestDelegatingResolverResolveNow(t *testing.T) {
 
 	// Manual resolver to control the target resolution.
 	targetResolver := manual.NewBuilderWithScheme("test")
-	targetResolverCalled := make(chan struct{})
+	targetResolverCalled := make(chan struct{}, 1)
 	targetResolver.ResolveNowCallback = func(resolver.ResolveNowOptions) {
 		// Updating the resolver state should not deadlock.
 		targetResolver.CC().UpdateState(resolver.State{
 			Endpoints: []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: "1.1.1.1"}}}},
 		})
-		close(targetResolverCalled)
+		targetResolverCalled <- struct{}{}
 	}
 
 	target := targetResolver.Scheme() + ":///" + "ignored"
 	// Set up a manual DNS resolver to control the proxy address resolution.
-	proxyResolver := setupDNS(t)
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
+
 	proxyResolverCalled := make(chan struct{})
 	proxyResolver.ResolveNowCallback = func(resolver.ResolveNowOptions) {
 		// Updating the resolver state should not deadlock.
@@ -743,12 +788,23 @@ func (s) TestDelegatingResolverResolveNow(t *testing.T) {
 		t.Fatalf("Failed to create delegating resolver: %v", err)
 	}
 
-	// Call ResolveNow on the delegatingResolver and verify both children
-	// receive the ResolveNow call.
+	// ResolveNow of manual proxy resolver will not be called. Proxy resolver is
+	// only built when we get the first update from target resolver. Therefore
+	// in the first ResolveNow, proxy resolver will be a no-op resolver and only
+	// target resolver's ResolveNow will be called.
 	dr.ResolveNow(resolver.ResolveNowOptions{})
-
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
+	select {
+	case <-targetResolverCalled:
+	case <-ctx.Done():
+		t.Fatalf("context timed out waiting for targetResolver.ResolveNow() to be called.")
+	}
+
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
+
+	dr.ResolveNow(resolver.ResolveNowOptions{})
+
 	select {
 	case <-targetResolverCalled:
 	case <-ctx.Done():
@@ -760,3 +816,155 @@ func (s) TestDelegatingResolverResolveNow(t *testing.T) {
 		t.Fatalf("context timed out waiting for proxyResolver.ResolveNow() to be called.")
 	}
 }
+
+// Tests the scenario where a proxy is configured, and the resolver returns a
+// network type other than tcp for all addresses. The test verifies that the
+// delegating resolver avoids the proxy build and directly sends the update
+// from target resolver to clientconn.
+func (s) TestDelegatingResolverForNonTCPTarget(t *testing.T) {
+	const (
+		targetTestAddr          = "test.target"
+		resolvedTargetTestAddr1 = "1.1.1.1:8080"
+		envProxyAddr            = "proxytest.com"
+	)
+	hpfe := func(req *http.Request) (*url.URL, error) {
+		if req.URL.Host == targetTestAddr {
+			return &url.URL{
+				Scheme: "https",
+				Host:   envProxyAddr,
+			}, nil
+		}
+		t.Errorf("Unexpected request host to proxy: %s want %s", req.URL.Host, targetTestAddr)
+		return nil, nil
+	}
+	originalhpfe := delegatingresolver.HTTPSProxyFromEnvironment
+	delegatingresolver.HTTPSProxyFromEnvironment = hpfe
+	defer func() {
+		delegatingresolver.HTTPSProxyFromEnvironment = originalhpfe
+	}()
+
+	// Manual resolver to control the target resolution.
+	targetResolver := manual.NewBuilderWithScheme("test")
+	target := targetResolver.Scheme() + ":///" + targetTestAddr
+	// Set up a manual DNS resolver to control the proxy address resolution.
+	_, proxyResolverBuilt := setupDNS(t)
+
+	tcc, stateCh, _ := createTestResolverClientConn(t)
+	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
+		t.Fatalf("Failed to create delegating resolver: %v", err)
+	}
+
+	// Set network to anything other than tcp.
+	nonTCPAddr := networktype.Set(resolver.Address{Addr: resolvedTargetTestAddr1}, "unix")
+	targetResolver.UpdateState(resolver.State{
+		Addresses:     []resolver.Address{nonTCPAddr},
+		Endpoints:     []resolver.Endpoint{{Addresses: []resolver.Address{nonTCPAddr}}},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	})
+
+	var gotState resolver.State
+	select {
+	case gotState = <-stateCh:
+	case <-time.After(defaultTestTimeout):
+		t.Fatal("Timeout when waiting for a state update from the delegating resolver")
+	}
+
+	// Verify that the delegating resolver doesn't call proxy resolver's
+	// UpdateState since we have no tcp address
+	select {
+	case <-proxyResolverBuilt:
+		t.Fatal("Unexpected call to proxy resolver update state")
+	case <-time.After(defaultTestShortTimeout):
+	}
+
+	wantState := resolver.State{
+		Addresses:     []resolver.Address{nonTCPAddr},
+		Endpoints:     []resolver.Endpoint{{Addresses: []resolver.Address{nonTCPAddr}}},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	}
+
+	// Verify that the state clientconn receives is same as updated by target
+	// resolver, since we want to avoid proxy for any network type apart from
+	// tcp.
+	if diff := cmp.Diff(gotState, wantState); diff != "" {
+		t.Fatalf("Unexpected state from delegating resolver. Diff (-got +want):\n%s", diff)
+	}
+}
+
+// Tests the scenario where a proxy is configured, and the resolver returns tcp
+// and non-tcp addresses. The test verifies that the delegating resolver doesn't
+// add proxyatrribute to adresses with network type other than tcp , but adds
+// the proxyattribute to addresses with network type tcp.
+func (s) TestDelegatingResolverForMixNetworkType(t *testing.T) {
+	const (
+		targetTestAddr          = "test.target"
+		resolvedTargetTestAddr1 = "1.1.1.1:8080"
+		resolvedTargetTestAddr2 = "2.2.2.2:8080"
+		envProxyAddr            = "proxytest.com"
+	)
+	hpfe := func(req *http.Request) (*url.URL, error) {
+		if req.URL.Host == targetTestAddr {
+			return &url.URL{
+				Scheme: "https",
+				Host:   envProxyAddr,
+			}, nil
+		}
+		t.Errorf("Unexpected request host to proxy: %s want %s", req.URL.Host, targetTestAddr)
+		return nil, nil
+	}
+	originalhpfe := delegatingresolver.HTTPSProxyFromEnvironment
+	delegatingresolver.HTTPSProxyFromEnvironment = hpfe
+	defer func() {
+		delegatingresolver.HTTPSProxyFromEnvironment = originalhpfe
+	}()
+
+	// Manual resolver to control the target resolution.
+	targetResolver := manual.NewBuilderWithScheme("test")
+	target := targetResolver.Scheme() + ":///" + targetTestAddr
+	// Set up a manual DNS resolver to control the proxy address resolution.
+	proxyResolver, proxyResolverBuilt := setupDNS(t)
+
+	tcc, stateCh, _ := createTestResolverClientConn(t)
+	if _, err := delegatingresolver.New(resolver.Target{URL: *testutils.MustParseURL(target)}, tcc, resolver.BuildOptions{}, targetResolver, false); err != nil {
+		t.Fatalf("Failed to create delegating resolver: %v", err)
+	}
+	// Set network to anything other than tcp.
+	nonTCPAddr := networktype.Set(resolver.Address{Addr: resolvedTargetTestAddr1}, "unix")
+	targetResolver.UpdateState(resolver.State{
+		Addresses:     []resolver.Address{nonTCPAddr, {Addr: resolvedTargetTestAddr2}},
+		Endpoints:     []resolver.Endpoint{{Addresses: []resolver.Address{nonTCPAddr, {Addr: resolvedTargetTestAddr2}}}},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	})
+
+	select {
+	case <-stateCh:
+		t.Fatalf("Delegating resolver invoked UpdateState before both the proxy and target resolvers had updated their states.")
+	case <-time.After(defaultTestShortTimeout):
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Wait for the proxy resolver to be built before calling UpdateState.
+	mustBuildResolver(ctx, t, proxyResolverBuilt)
+	proxyResolver.UpdateState(resolver.State{
+		Addresses:     []resolver.Address{{Addr: envProxyAddr}},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	})
+
+	var gotState resolver.State
+	select {
+	case gotState = <-stateCh:
+	case <-ctx.Done():
+		t.Fatal("Context timed out when waiting for a state update from the delegating resolver")
+	}
+	wantState := resolver.State{
+		Addresses:     []resolver.Address{nonTCPAddr, proxyAddressWithTargetAttribute(envProxyAddr, resolvedTargetTestAddr2)},
+		Endpoints:     []resolver.Endpoint{{Addresses: []resolver.Address{nonTCPAddr, proxyAddressWithTargetAttribute(envProxyAddr, resolvedTargetTestAddr2)}}},
+		ServiceConfig: &serviceconfig.ParseResult{},
+	}
+
+	if diff := cmp.Diff(gotState, wantState); diff != "" {
+		t.Fatalf("Unexpected state from delegating resolver. Diff (-got +want):\n%v", diff)
+	}
+}
diff --git a/internal/stats/metrics_recorder_list_test.go b/internal/stats/metrics_recorder_list_test.go
index fef3d1df..38f9472f 100644
--- a/internal/stats/metrics_recorder_list_test.go
+++ b/internal/stats/metrics_recorder_list_test.go
@@ -23,7 +23,6 @@ package stats_test
 import (
 	"context"
 	"fmt"
-	"log"
 	"strings"
 	"testing"
 	"time"
@@ -154,7 +153,7 @@ func (s) TestMetricsRecorderList(t *testing.T) {
 
 	cc, err := grpc.NewClient(mr.Scheme()+":///", grpc.WithResolvers(mr), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithStatsHandler(mr2))
 	if err != nil {
-		log.Fatalf("Failed to dial: %v", err)
+		t.Fatalf("grpc.NewClient() failed: %v", err)
 	}
 	defer cc.Close()
 
diff --git a/internal/status/status_test.go b/internal/status/status_test.go
new file mode 100644
index 00000000..1d3fc4c4
--- /dev/null
+++ b/internal/status/status_test.go
@@ -0,0 +1,40 @@
+/*
+ *
+ * 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 status_test
+
+import (
+	"testing"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/internal/status"
+)
+
+func TestRawStatusProto(t *testing.T) {
+	spb := status.RawStatusProto(nil)
+	if spb != nil {
+		t.Errorf("RawStatusProto(nil) = %v; must return nil", spb)
+	}
+	s := status.New(codes.Internal, "test internal error")
+	spb1 := status.RawStatusProto(s)
+	spb2 := status.RawStatusProto(s)
+	// spb1 and spb2 should be the same pointer: no copies
+	if spb1 != spb2 {
+		t.Errorf("RawStatusProto(s)=%p then %p; must return the same pointer", spb1, spb2)
+	}
+}
diff --git a/internal/transport/http_util_test.go b/internal/transport/http_util_test.go
index 5a259d43..eaf17d0c 100644
--- a/internal/transport/http_util_test.go
+++ b/internal/transport/http_util_test.go
@@ -22,28 +22,49 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"reflect"
 	"testing"
 	"time"
 )
 
-func (s) TestTimeoutDecode(t *testing.T) {
+func (s) TestDecodeTimeout(t *testing.T) {
 	for _, test := range []struct {
 		// input
 		s string
 		// output
-		d   time.Duration
-		err error
+		d       time.Duration
+		wantErr bool
 	}{
-		{"1234S", time.Second * 1234, nil},
-		{"1234x", 0, fmt.Errorf("transport: timeout unit is not recognized: %q", "1234x")},
-		{"1", 0, fmt.Errorf("transport: timeout string is too short: %q", "1")},
-		{"", 0, fmt.Errorf("transport: timeout string is too short: %q", "")},
+
+		{"00000001n", time.Nanosecond, false},
+		{"10u", time.Microsecond * 10, false},
+		{"00000010m", time.Millisecond * 10, false},
+		{"1234S", time.Second * 1234, false},
+		{"00000001M", time.Minute, false},
+		{"09999999S", time.Second * 9999999, false},
+		{"99999999S", time.Second * 99999999, false},
+		{"99999999M", time.Minute * 99999999, false},
+		{"2562047H", time.Hour * 2562047, false},
+		{"2562048H", time.Duration(math.MaxInt64), false},
+		{"99999999H", time.Duration(math.MaxInt64), false},
+		{"-1S", 0, true},
+		{"1234x", 0, true},
+		{"1234s", 0, true},
+		{"1234", 0, true},
+		{"1", 0, true},
+		{"", 0, true},
+		{"9a1S", 0, true},
+		{"0S", 0, true}, // PROTOCOL-HTTP2.md requires positive integers
+		{"00000000S", 0, true},
+		{"000000000S", 0, true},
 	} {
 		d, err := decodeTimeout(test.s)
-		if d != test.d || fmt.Sprint(err) != fmt.Sprint(test.err) {
-			t.Fatalf("timeoutDecode(%q) = %d, %v, want %d, %v", test.s, int64(d), err, int64(test.d), test.err)
+		gotErr := err != nil
+		if d != test.d || gotErr != test.wantErr {
+			t.Errorf("timeoutDecode(%q) = %d, %v, want %d, wantErr=%v",
+				test.s, int64(d), err, int64(test.d), test.wantErr)
 		}
 	}
 }
@@ -211,9 +232,9 @@ func (s) TestParseDialTarget(t *testing.T) {
 		{"dns:///google.com", "tcp", "dns:///google.com"},
 		{"/unix/socket/address", "tcp", "/unix/socket/address"},
 	} {
-		gotNet, gotAddr := parseDialTarget(test.target)
+		gotNet, gotAddr := ParseDialTarget(test.target)
 		if gotNet != test.wantNet || gotAddr != test.wantAddr {
-			t.Errorf("parseDialTarget(%q) = %s, %s want %s, %s", test.target, gotNet, gotAddr, test.wantNet, test.wantAddr)
+			t.Errorf("ParseDialTarget(%q) = %s, %s want %s, %s", test.target, gotNet, gotAddr, test.wantNet, test.wantAddr)
 		}
 	}
 }
diff --git a/internal/transport/transport_test.go b/internal/transport/transport_test.go
index 8b121959..d2c4fac3 100644
--- a/internal/transport/transport_test.go
+++ b/internal/transport/transport_test.go
@@ -919,8 +919,9 @@ func (s) TestLargeMessageSuspension(t *testing.T) {
 	}
 	// The server will send an RST stream frame on observing the deadline
 	// expiration making the client stream fail with a DeadlineExceeded status.
-	if _, err := s.readTo(make([]byte, 8)); err != io.EOF {
-		t.Fatalf("Read got unexpected error: %v, want %v", err, io.EOF)
+	_, err = s.readTo(make([]byte, 8))
+	if st, ok := status.FromError(err); !ok || st.Code() != codes.DeadlineExceeded {
+		t.Fatalf("Read got unexpected error: %v, want status with code %v", err, codes.DeadlineExceeded)
 	}
 	if got, want := s.Status().Code(), codes.DeadlineExceeded; got != want {
 		t.Fatalf("Read got status %v with code %v, want %v", s.Status(), got, want)
diff --git a/stats/stats_test.go b/stats/stats_test.go
index ec5ffa04..c83e12f8 100644
--- a/stats/stats_test.go
+++ b/stats/stats_test.go
@@ -30,6 +30,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/grpctest"
@@ -268,13 +269,12 @@ func (te *test) startServer(ts testgrpc.TestServiceServer) {
 	te.srvAddr = lis.Addr().String()
 }
 
-func (te *test) clientConn() *grpc.ClientConn {
+func (te *test) clientConn(ctx context.Context) *grpc.ClientConn {
 	if te.cc != nil {
 		return te.cc
 	}
 	opts := []grpc.DialOption{
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithBlock(),
 		grpc.WithUserAgent("test/0.0.1"),
 	}
 	if te.compress == "gzip" {
@@ -288,10 +288,12 @@ func (te *test) clientConn() *grpc.ClientConn {
 	}
 
 	var err error
-	te.cc, err = grpc.Dial(te.srvAddr, opts...)
+	te.cc, err = grpc.NewClient(te.srvAddr, opts...)
 	if err != nil {
-		te.t.Fatalf("Dial(%q) = %v", te.srvAddr, err)
+		te.t.Fatalf("grpc.NewClient(%q) failed: %v", te.srvAddr, err)
 	}
+	te.cc.Connect()
+	testutils.AwaitState(ctx, te.t, te.cc, connectivity.Ready)
 	return te.cc
 }
 
@@ -317,15 +319,15 @@ func (te *test) doUnaryCall(c *rpcConfig) (*testpb.SimpleRequest, *testpb.Simple
 		req  *testpb.SimpleRequest
 		err  error
 	)
-	tc := testgrpc.NewTestServiceClient(te.clientConn())
+	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	tc := testgrpc.NewTestServiceClient(te.clientConn(tCtx))
 	if c.success {
 		req = &testpb.SimpleRequest{Payload: idToPayload(errorID + 1)}
 	} else {
 		req = &testpb.SimpleRequest{Payload: idToPayload(errorID)}
 	}
 
-	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	resp, err = tc.UnaryCall(metadata.NewOutgoingContext(tCtx, testMetadata), req, grpc.WaitForReady(!c.failfast))
 	return req, resp, err
 }
@@ -336,9 +338,9 @@ func (te *test) doFullDuplexCallRoundtrip(c *rpcConfig) ([]proto.Message, []prot
 		resps []proto.Message
 		err   error
 	)
-	tc := testgrpc.NewTestServiceClient(te.clientConn())
 	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
+	tc := testgrpc.NewTestServiceClient(te.clientConn(tCtx))
 	stream, err := tc.FullDuplexCall(metadata.NewOutgoingContext(tCtx, testMetadata), grpc.WaitForReady(!c.failfast))
 	if err != nil {
 		return reqs, resps, err
@@ -377,9 +379,9 @@ func (te *test) doClientStreamCall(c *rpcConfig) ([]proto.Message, *testpb.Strea
 		resp *testpb.StreamingInputCallResponse
 		err  error
 	)
-	tc := testgrpc.NewTestServiceClient(te.clientConn())
 	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
+	tc := testgrpc.NewTestServiceClient(te.clientConn(tCtx))
 	stream, err := tc.StreamingInputCall(metadata.NewOutgoingContext(tCtx, testMetadata), grpc.WaitForReady(!c.failfast))
 	if err != nil {
 		return reqs, resp, err
@@ -407,16 +409,15 @@ func (te *test) doServerStreamCall(c *rpcConfig) (*testpb.StreamingOutputCallReq
 		resps []proto.Message
 		err   error
 	)
-
-	tc := testgrpc.NewTestServiceClient(te.clientConn())
+	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	tc := testgrpc.NewTestServiceClient(te.clientConn(tCtx))
 
 	var startID int32
 	if !c.success {
 		startID = errorID
 	}
 	req = &testpb.StreamingOutputCallRequest{Payload: idToPayload(startID)}
-	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	stream, err := tc.StreamingOutputCall(metadata.NewOutgoingContext(tCtx, testMetadata), req, grpc.WaitForReady(!c.failfast))
 	if err != nil {
 		return req, resps, err
diff --git a/status/status_test.go b/status/status_test.go
index 03085d97..b0f1f4e7 100644
--- a/status/status_test.go
+++ b/status/status_test.go
@@ -354,6 +354,9 @@ func (s) TestStatus_ErrorDetails(t *testing.T) {
 			t.Fatalf("(%v).WithDetails(%+v) failed: %v", str(s), tc.details, err)
 		}
 		details := s.Details()
+		if len(details) != len(tc.details) {
+			t.Fatalf("len(s.Details()) = %d, want = %d.", len(details), len(tc.details))
+		}
 		for i := range details {
 			if !proto.Equal(details[i].(protoreflect.ProtoMessage), tc.details[i].(protoreflect.ProtoMessage)) {
 				t.Fatalf("(%v).Details()[%d] = %+v, want %+v", str(s), i, details[i], tc.details[i])
@@ -420,7 +423,7 @@ func (s) TestStatus_ErrorDetails_Fail(t *testing.T) {
 	for _, tc := range tests {
 		details := tc.s.Details()
 		if len(details) != len(tc.want) {
-			t.Fatalf("len(s.Details()) = %v, want = %v.", len(details), len(tc.want))
+			t.Fatalf("len(s.Details()) = %d, want = %d.", len(details), len(tc.want))
 		}
 		for i, d := range details {
 			// s.Details can either contain an error or a proto message.  We
diff --git a/test/channelz_test.go b/test/channelz_test.go
index 0a9b88bd..93847126 100644
--- a/test/channelz_test.go
+++ b/test/channelz_test.go
@@ -304,11 +304,11 @@ func (s) TestCZTopChannelRegistrationAndDeletion(t *testing.T) {
 	}
 }
 
-func (s) TestCZTopChannelRegistrationAndDeletionWhenDialFail(t *testing.T) {
-	// Make dial fails (due to no transport security specified)
+func (s) TestCZTopChannelRegistrationAndDeletionWhenNewClientFail(t *testing.T) {
+	// Make newclient fails (due to no transport security specified)
 	_, err := grpc.NewClient("fake.addr")
 	if err == nil {
-		t.Fatal("expecting dial to fail")
+		t.Fatal("expecting newclient to fail")
 	}
 	if tcs, end := channelz.GetTopChannels(0, 0); tcs != nil || !end {
 		t.Fatalf("GetTopChannels(0, 0) = %v, %v, want <nil>, true", tcs, end)
diff --git a/test/end2end_test.go b/test/end2end_test.go
index a4258771..fd5e29ea 100644
--- a/test/end2end_test.go
+++ b/test/end2end_test.go
@@ -854,10 +854,11 @@ func (te *test) clientConn(opts ...grpc.DialOption) *grpc.ClientConn {
 	var scheme string
 	opts, scheme = te.configDial(opts...)
 	var err error
-	te.cc, err = grpc.Dial(scheme+te.srvAddr, opts...)
+	te.cc, err = grpc.NewClient(scheme+te.srvAddr, opts...)
 	if err != nil {
-		te.t.Fatalf("Dial(%q) = %v", scheme+te.srvAddr, err)
+		te.t.Fatalf("grpc.NewClient(%q) failed: %v", scheme+te.srvAddr, err)
 	}
+	te.cc.Connect()
 	return te.cc
 }
 
diff --git a/test/transport_test.go b/test/transport_test.go
index 2f61a6c0..10e9ab57 100644
--- a/test/transport_test.go
+++ b/test/transport_test.go
@@ -19,16 +19,20 @@ package test
 
 import (
 	"context"
+	"encoding/binary"
 	"io"
 	"net"
 	"sync"
 	"testing"
 
+	"golang.org/x/net/http2"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal/grpcsync"
 	"google.golang.org/grpc/internal/stubserver"
+	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/internal/transport"
 	"google.golang.org/grpc/status"
 
@@ -153,3 +157,75 @@ func (s) TestClientTransportRestartsAfterStreamIDExhausted(t *testing.T) {
 		t.Fatal("Timeout expired when waiting for first client transport to close")
 	}
 }
+
+// Tests that an RST_STREAM frame that causes an io.ErrUnexpectedEOF while
+// reading a gRPC message is correctly converted to a gRPC status with code
+// CANCELLED. The test sends a data frame with a partial gRPC message, followed
+// by an RST_STREAM frame with HTTP/2 code CANCELLED. The test asserts the
+// client receives the correct status.
+func (s) TestRSTDuringMessageRead(t *testing.T) {
+	lis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer lis.Close()
+
+	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(%s) = %v", lis.Addr().String(), err)
+	}
+	defer cc.Close()
+
+	go func() {
+		conn, err := lis.Accept()
+		if err != nil {
+			t.Errorf("lis.Accept() = %v", err)
+			return
+		}
+		defer conn.Close()
+		framer := http2.NewFramer(conn, conn)
+
+		if _, err := io.ReadFull(conn, make([]byte, len(clientPreface))); err != nil {
+			t.Errorf("Error while reading client preface: %v", err)
+			return
+		}
+		if err := framer.WriteSettings(); err != nil {
+			t.Errorf("Error while writing settings: %v", err)
+			return
+		}
+		if err := framer.WriteSettingsAck(); err != nil {
+			t.Errorf("Error while writing settings: %v", err)
+			return
+		}
+		for ctx.Err() == nil {
+			frame, err := framer.ReadFrame()
+			if err != nil {
+				return
+			}
+			switch frame := frame.(type) {
+			case *http2.HeadersFrame:
+				// When the client creates a stream, write a partial gRPC
+				// message followed by an RST_STREAM.
+				const messageLen = 2048
+				buf := make([]byte, messageLen/2)
+				// Write the gRPC message length header.
+				binary.BigEndian.PutUint32(buf[1:5], uint32(messageLen))
+				if err := framer.WriteData(1, false, buf); err != nil {
+					return
+				}
+				framer.WriteRSTStream(1, http2.ErrCodeCancel)
+			default:
+				t.Logf("Server received frame: %v", frame)
+			}
+		}
+	}()
+
+	// The server will send a partial gRPC message before cancelling the stream.
+	// The client should get a gRPC status with code CANCELLED.
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.Canceled {
+		t.Fatalf("client.EmptyCall() returned %v; want status with code %v", err, codes.Canceled)
+	}
+}
diff --git a/xds/internal/clients/grpctransport/examples_test.go b/xds/internal/clients/grpctransport/examples_test.go
index 40b0cb27..57191d1f 100644
--- a/xds/internal/clients/grpctransport/examples_test.go
+++ b/xds/internal/clients/grpctransport/examples_test.go
@@ -30,10 +30,10 @@ import (
 // its extensions.
 //
 // This example is creating clients.ServerIdentifier to connect to server at
-// localhost:5678 using the credentials named "local". Note that "local" must
-// exist as an entry in the provided credentials to grpctransport.Builder.
+// localhost:5678 using the config named "local". Note that "local" must
+// exist as an entry in the provided configs to grpctransport.Builder.
 func ExampleServerIdentifierExtension() {
 	// Note the Extensions field is set by value and not by pointer.
-	fmt.Printf("%+v", clients.ServerIdentifier{ServerURI: "localhost:5678", Extensions: grpctransport.ServerIdentifierExtension{Credentials: "local"}})
-	// Output: {ServerURI:localhost:5678 Extensions:{Credentials:local}}
+	fmt.Printf("%+v", clients.ServerIdentifier{ServerURI: "localhost:5678", Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "local"}})
+	// Output: {ServerURI:localhost:5678 Extensions:{ConfigName:local}}
 }
diff --git a/xds/internal/clients/grpctransport/grpc_transport_ext_test.go b/xds/internal/clients/grpctransport/grpc_transport_ext_test.go
index b34a2057..75ac5146 100644
--- a/xds/internal/clients/grpctransport/grpc_transport_ext_test.go
+++ b/xds/internal/clients/grpctransport/grpc_transport_ext_test.go
@@ -72,14 +72,14 @@ func (s) TestBuild_Single(t *testing.T) {
 
 	serverID := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "local"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "local"},
 	}
-	credentials := map[string]credentials.Bundle{
-		"local": &testCredentials{transportCredentials: local.NewCredentials()},
+	configs := map[string]grpctransport.Config{
+		"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
 	}
 
 	// Calling Build() first time should create new gRPC transport.
-	builder := grpctransport.NewBuilder(credentials)
+	builder := grpctransport.NewBuilder(configs)
 	tr, err := builder.Build(serverID)
 	if err != nil {
 		t.Fatalf("Failed to build transport: %v", err)
@@ -173,19 +173,19 @@ func (s) TestBuild_Multiple(t *testing.T) {
 
 	serverID1 := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "local"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "local"},
 	}
 	serverID2 := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
-	credentials := map[string]credentials.Bundle{
-		"local":    &testCredentials{transportCredentials: local.NewCredentials()},
-		"insecure": insecure.NewBundle(),
+	configs := map[string]grpctransport.Config{
+		"local":    {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
+		"insecure": {Credentials: insecure.NewBundle()},
 	}
 
 	// Create two gRPC transports.
-	builder := grpctransport.NewBuilder(credentials)
+	builder := grpctransport.NewBuilder(configs)
 
 	tr1, err := builder.Build(serverID1)
 	if err != nil {
diff --git a/xds/internal/clients/grpctransport/grpc_transport_test.go b/xds/internal/clients/grpctransport/grpc_transport_test.go
index ab2cd7c1..23e75257 100644
--- a/xds/internal/clients/grpctransport/grpc_transport_test.go
+++ b/xds/internal/clients/grpctransport/grpc_transport_test.go
@@ -137,15 +137,15 @@ func (tc *testCredentials) PerRPCCredentials() credentials.PerRPCCredentials {
 // Transport in both cases when provided clients.ServerIdentifer is same
 // one of the existing transport or a new one.
 func (s) TestBuild_Success(t *testing.T) {
-	credentials := map[string]credentials.Bundle{
-		"local":    &testCredentials{transportCredentials: local.NewCredentials()},
-		"insecure": insecure.NewBundle(),
+	configs := map[string]Config{
+		"local":    {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
+		"insecure": {Credentials: insecure.NewBundle()},
 	}
-	b := NewBuilder(credentials)
+	b := NewBuilder(configs)
 
 	serverID1 := clients.ServerIdentifier{
 		ServerURI:  "server-address",
-		Extensions: ServerIdentifierExtension{Credentials: "local"},
+		Extensions: ServerIdentifierExtension{ConfigName: "local"},
 	}
 	tr1, err := b.Build(serverID1)
 	if err != nil {
@@ -155,7 +155,7 @@ func (s) TestBuild_Success(t *testing.T) {
 
 	serverID2 := clients.ServerIdentifier{
 		ServerURI:  "server-address",
-		Extensions: ServerIdentifierExtension{Credentials: "local"},
+		Extensions: ServerIdentifierExtension{ConfigName: "local"},
 	}
 	tr2, err := b.Build(serverID2)
 	if err != nil {
@@ -165,7 +165,7 @@ func (s) TestBuild_Success(t *testing.T) {
 
 	serverID3 := clients.ServerIdentifier{
 		ServerURI:  "server-address",
-		Extensions: ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 	tr3, err := b.Build(serverID3)
 	if err != nil {
@@ -191,7 +191,7 @@ func (s) TestBuild_Failure(t *testing.T) {
 			name: "ServerURI is empty",
 			serverID: clients.ServerIdentifier{
 				ServerURI:  "",
-				Extensions: ServerIdentifierExtension{Credentials: "local"},
+				Extensions: ServerIdentifierExtension{ConfigName: "local"},
 			},
 		},
 		{
@@ -206,26 +206,41 @@ func (s) TestBuild_Failure(t *testing.T) {
 			},
 		},
 		{
-			name: "ServerIdentifierExtension Credentials is nil",
+			name: "ServerIdentifierExtension without ConfigName",
 			serverID: clients.ServerIdentifier{
 				ServerURI:  "server-address",
 				Extensions: ServerIdentifierExtension{},
 			},
 		},
+		{
+			name: "ServerIdentifierExtension ConfigName is not present",
+			serverID: clients.ServerIdentifier{
+				ServerURI:  "server-address",
+				Extensions: ServerIdentifierExtension{ConfigName: "unknown"},
+			},
+		},
+		{
+			name: "ServerIdentifierExtension ConfigName maps to nil credentials",
+			serverID: clients.ServerIdentifier{
+				ServerURI:  "server-address",
+				Extensions: ServerIdentifierExtension{ConfigName: "nil-credentials"},
+			},
+		},
 		{
 			name: "ServerIdentifierExtension is added as pointer",
 			serverID: clients.ServerIdentifier{
 				ServerURI:  "server-address",
-				Extensions: &ServerIdentifierExtension{Credentials: "local"},
+				Extensions: &ServerIdentifierExtension{ConfigName: "local"},
 			},
 		},
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			credentials := map[string]credentials.Bundle{
-				"local": &testCredentials{transportCredentials: local.NewCredentials()},
+			configs := map[string]Config{
+				"local":           {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
+				"nil-credentials": {Credentials: nil},
 			}
-			b := NewBuilder(credentials)
+			b := NewBuilder(configs)
 			tr, err := b.Build(test.serverID)
 			if err == nil {
 				t.Fatalf("Build() succeeded, want error")
@@ -238,18 +253,19 @@ func (s) TestBuild_Failure(t *testing.T) {
 }
 
 // TestNewStream_Success verifies that NewStream() successfully creates a new
-// client stream for the server when provided a valid server URI.
+// client stream for the server when provided a valid server URI and a config
+// with valid credentials.
 func (s) TestNewStream_Success(t *testing.T) {
 	ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})
 
 	serverCfg := clients.ServerIdentifier{
 		ServerURI:  ts.address,
-		Extensions: ServerIdentifierExtension{Credentials: "local"},
+		Extensions: ServerIdentifierExtension{ConfigName: "local"},
 	}
-	credentials := map[string]credentials.Bundle{
-		"local": &testCredentials{transportCredentials: local.NewCredentials()},
+	configs := map[string]Config{
+		"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
 	}
-	builder := NewBuilder(credentials)
+	builder := NewBuilder(configs)
 	transport, err := builder.Build(serverCfg)
 	if err != nil {
 		t.Fatalf("Failed to build transport: %v", err)
@@ -263,17 +279,65 @@ func (s) TestNewStream_Success(t *testing.T) {
 	}
 }
 
+// TestNewStream_Success_WithCustomGRPCNewClient verifies that NewStream()
+// successfully creates a new client stream for the server when provided a
+// valid server URI and a config with valid credentials and a custom gRPC
+// NewClient function.
+func (s) TestNewStream_Success_WithCustomGRPCNewClient(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})
+
+	// Create a custom dialer function that will be used by the gRPC client.
+	customDialerCalled := make(chan struct{}, 1)
+	customGRPCNewClient := func(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+		customDialerCalled <- struct{}{}
+		return grpc.NewClient(target, opts...)
+	}
+
+	configs := map[string]Config{
+		"custom-dialer-config": {
+			Credentials:   &testCredentials{transportCredentials: local.NewCredentials()},
+			GRPCNewClient: customGRPCNewClient,
+		},
+	}
+	builder := NewBuilder(configs)
+
+	serverID := clients.ServerIdentifier{
+		ServerURI:  ts.address,
+		Extensions: ServerIdentifierExtension{ConfigName: "custom-dialer-config"},
+	}
+
+	transport, err := builder.Build(serverID)
+	if err != nil {
+		t.Fatalf("builder.Build(%+v) failed: %v", serverID, err)
+	}
+	defer transport.Close()
+
+	select {
+	case <-customDialerCalled:
+	case <-ctx.Done():
+		t.Fatalf("Timeout waiting for custom dialer to be called: %v", ctx.Err())
+	}
+
+	// Verify that the transport works by creating a stream.
+	if _, err = transport.NewStream(ctx, "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources"); err != nil {
+		t.Fatalf("transport.NewStream() failed with custom dialer: %v", err)
+	}
+}
+
 // TestNewStream_Error verifies that NewStream() returns an error
 // when attempting to create a stream with an invalid server URI.
 func (s) TestNewStream_Error(t *testing.T) {
 	serverCfg := clients.ServerIdentifier{
 		ServerURI:  "invalid-server-uri",
-		Extensions: ServerIdentifierExtension{Credentials: "local"},
+		Extensions: ServerIdentifierExtension{ConfigName: "local"},
 	}
-	credentials := map[string]credentials.Bundle{
-		"local": &testCredentials{transportCredentials: local.NewCredentials()},
+	configs := map[string]Config{
+		"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
 	}
-	builder := NewBuilder(credentials)
+	builder := NewBuilder(configs)
 	transport, err := builder.Build(serverCfg)
 	if err != nil {
 		t.Fatalf("Failed to build transport: %v", err)
@@ -304,12 +368,12 @@ func (s) TestStream_SendAndRecv(t *testing.T) {
 	// Build a grpc-based transport to the above server.
 	serverCfg := clients.ServerIdentifier{
 		ServerURI:  ts.address,
-		Extensions: ServerIdentifierExtension{Credentials: "local"},
+		Extensions: ServerIdentifierExtension{ConfigName: "local"},
 	}
-	credentials := map[string]credentials.Bundle{
-		"local": &testCredentials{transportCredentials: local.NewCredentials()},
+	configs := map[string]Config{
+		"local": {Credentials: &testCredentials{transportCredentials: local.NewCredentials()}},
 	}
-	builder := NewBuilder(credentials)
+	builder := NewBuilder(configs)
 	transport, err := builder.Build(serverCfg)
 	if err != nil {
 		t.Fatalf("Failed to build transport: %v", err)
diff --git a/xds/internal/clients/lrsclient/loadreport_test.go b/xds/internal/clients/lrsclient/loadreport_test.go
index 1c5072ac..bcdec9c6 100644
--- a/xds/internal/clients/lrsclient/loadreport_test.go
+++ b/xds/internal/clients/lrsclient/loadreport_test.go
@@ -28,7 +28,6 @@ import (
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/google/uuid"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/status"
@@ -132,17 +131,17 @@ func (s) TestReportLoad_ConnectionCreation(t *testing.T) {
 	// reporting is per-server and not per-authority.
 	nodeID := uuid.New().String()
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	config := lrsclient.Config{
 		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 	}
 	client, err := lrsclient.New(config)
 	if err != nil {
 		t.Fatalf("lrsclient.New() failed: %v", err)
 	}
 
-	serverIdentifier1 := clients.ServerIdentifier{ServerURI: mgmtServer1.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	serverIdentifier1 := clients.ServerIdentifier{ServerURI: mgmtServer1.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
 	loadStore1, err := client.ReportLoad(serverIdentifier1)
 	if err != nil {
 		t.Fatalf("client.ReportLoad() failed: %v", err)
@@ -162,7 +161,7 @@ func (s) TestReportLoad_ConnectionCreation(t *testing.T) {
 
 	// Call the load reporting API to report load to the first management
 	// server, and ensure that a connection to the server is created.
-	serverIdentifier2 := clients.ServerIdentifier{ServerURI: mgmtServer2.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	serverIdentifier2 := clients.ServerIdentifier{ServerURI: mgmtServer2.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
 	loadStore2, err := client.ReportLoad(serverIdentifier2)
 	if err != nil {
 		t.Fatalf("client.ReportLoad() failed: %v", err)
@@ -276,10 +275,10 @@ func (s) TestReportLoad_StreamCreation(t *testing.T) {
 	// Create an LRS client with configuration pointing to the above server.
 	nodeID := uuid.New().String()
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	config := lrsclient.Config{
 		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 	}
 	client, err := lrsclient.New(config)
 	if err != nil {
@@ -287,7 +286,7 @@ func (s) TestReportLoad_StreamCreation(t *testing.T) {
 	}
 
 	// Call the load reporting API, and ensure that an LRS stream is created.
-	serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
 	loadStore1, err := client.ReportLoad(serverIdentifier)
 	if err != nil {
 		t.Fatalf("client.ReportLoad() failed: %v", err)
@@ -471,10 +470,10 @@ func (s) TestReportLoad_StopWithContext(t *testing.T) {
 	// Create an LRS client with configuration pointing to the above server.
 	nodeID := uuid.New().String()
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	config := lrsclient.Config{
 		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 	}
 	client, err := lrsclient.New(config)
 	if err != nil {
@@ -482,7 +481,7 @@ func (s) TestReportLoad_StopWithContext(t *testing.T) {
 	}
 
 	// Call the load reporting API, and ensure that an LRS stream is created.
-	serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"}}
+	serverIdentifier := clients.ServerIdentifier{ServerURI: mgmtServer.Address, Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}
 	loadStore, err := client.ReportLoad(serverIdentifier)
 	if err != nil {
 		t.Fatalf("client.ReportLoad() failed: %v", err)
diff --git a/xds/internal/clients/xdsclient/channel_test.go b/xds/internal/clients/xdsclient/channel_test.go
index d7417f31..1a9f37f8 100644
--- a/xds/internal/clients/xdsclient/channel_test.go
+++ b/xds/internal/clients/xdsclient/channel_test.go
@@ -29,7 +29,6 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/google/uuid"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -55,10 +54,10 @@ func xdsChannelForTest(t *testing.T, serverURI, nodeID string, watchExpiryTimeou
 	// Create a grpc transport to the above management server.
 	si := clients.ServerIdentifier{
 		ServerURI:  serverURI,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
-	tr, err := (grpctransport.NewBuilder(credentials)).Build(si)
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+	tr, err := (grpctransport.NewBuilder(configs)).Build(si)
 	if err != nil {
 		t.Fatalf("Failed to create a transport for server config %v: %v", si, err)
 	}
diff --git a/xds/internal/clients/xdsclient/test/authority_test.go b/xds/internal/clients/xdsclient/test/authority_test.go
index 27fd32ef..35d86921 100644
--- a/xds/internal/clients/xdsclient/test/authority_test.go
+++ b/xds/internal/clients/xdsclient/test/authority_test.go
@@ -24,7 +24,6 @@ import (
 	"testing"
 
 	"github.com/google/uuid"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -81,7 +80,7 @@ func setupForAuthorityTests(ctx context.Context, t *testing.T) (*testutils.Liste
 	nodeID := uuid.New().String()
 
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
-	ext := grpctransport.ServerIdentifierExtension{Credentials: "insecure"}
+	ext := grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}
 	siDefault := clients.ServerIdentifier{
 		ServerURI:  defaultAuthorityServer.Address,
 		Extensions: ext,
@@ -91,11 +90,11 @@ func setupForAuthorityTests(ctx context.Context, t *testing.T) (*testutils.Liste
 		Extensions: ext,
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: siDefault}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
@@ -242,20 +241,20 @@ func (s) TestAuthority_Fallback(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	psi := clients.ServerIdentifier{
 		ServerURI:  primaryMgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 	ssi := clients.ServerIdentifier{
 		ServerURI:  secondaryMgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
 	// Create config with the above primary and fallback management servers,
 	// and an xDS client with that configuration.
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: psi}, {ServerIdentifier: ssi}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp resource names used in this test do not specify an
 		// authority. These will end up looking up an entry with the
diff --git a/xds/internal/clients/xdsclient/test/dump_test.go b/xds/internal/clients/xdsclient/test/dump_test.go
index 9bf34e95..49f09453 100644
--- a/xds/internal/clients/xdsclient/test/dump_test.go
+++ b/xds/internal/clients/xdsclient/test/dump_test.go
@@ -28,7 +28,6 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -124,14 +123,14 @@ func (s) TestDumpResources_ManyToOne(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp resource names used in this test do not specify an
 		// authority. These will end up looking up an entry with the
@@ -308,18 +307,18 @@ func (s) TestDumpResources_ManyToMany(t *testing.T) {
 	resourceTypes[xdsresource.V3ListenerURL] = listenerType
 	si1 := clients.ServerIdentifier{
 		ServerURI:  mgmtServer1.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 	si2 := clients.ServerIdentifier{
 		ServerURI:  mgmtServer2.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si1}},
 		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
diff --git a/xds/internal/clients/xdsclient/test/helpers_test.go b/xds/internal/clients/xdsclient/test/helpers_test.go
index d296fee6..16453c80 100644
--- a/xds/internal/clients/xdsclient/test/helpers_test.go
+++ b/xds/internal/clients/xdsclient/test/helpers_test.go
@@ -20,6 +20,7 @@ package xdsclient_test
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"strconv"
@@ -29,12 +30,14 @@ import (
 
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/xds/internal/clients/internal/pretty"
+	"google.golang.org/grpc/xds/internal/clients/internal/testutils"
 	"google.golang.org/grpc/xds/internal/clients/xdsclient"
 	"google.golang.org/grpc/xds/internal/clients/xdsclient/internal/xdsresource"
 	"google.golang.org/protobuf/proto"
 
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
 	v3httppb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3"
+	"github.com/google/go-cmp/cmp"
 )
 
 type s struct {
@@ -262,3 +265,36 @@ func buildResourceName(typeName, auth, id string, ctxParams map[string]string) s
 		ContextParams: ctxParams,
 	}).String()
 }
+
+// testMetricsReporter is a MetricsReporter to be used in tests. It sends
+// recording events on channels and provides helpers to check if certain events
+// have taken place.
+type testMetricsReporter struct {
+	metricsCh *testutils.Channel
+}
+
+// newTestMetricsReporter returns a new testMetricsReporter.
+func newTestMetricsReporter() *testMetricsReporter {
+	return &testMetricsReporter{
+		metricsCh: testutils.NewChannelWithSize(1),
+	}
+}
+
+// waitForMetric waits for a metric to be recorded and verifies that the
+// recorded metrics data matches the expected metricsDataWant. Returns
+// an error if failed to wait or received wrong data.
+func (r *testMetricsReporter) waitForMetric(ctx context.Context, metricsDataWant any) error {
+	got, err := r.metricsCh.Receive(ctx)
+	if err != nil {
+		return fmt.Errorf("timeout waiting for int64Count")
+	}
+	if diff := cmp.Diff(got, metricsDataWant); diff != "" {
+		return fmt.Errorf("received unexpected metrics value (-got, +want): %v", diff)
+	}
+	return nil
+}
+
+// ReportMetric sends the metrics data to the metricsCh channel.
+func (r *testMetricsReporter) ReportMetric(m any) {
+	r.metricsCh.Send(m)
+}
diff --git a/xds/internal/clients/xdsclient/test/lds_watchers_test.go b/xds/internal/clients/xdsclient/test/lds_watchers_test.go
index 6bf32e73..5fe60424 100644
--- a/xds/internal/clients/xdsclient/test/lds_watchers_test.go
+++ b/xds/internal/clients/xdsclient/test/lds_watchers_test.go
@@ -28,7 +28,6 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/google/uuid"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -252,14 +251,14 @@ func (s) TestLDSWatch(t *testing.T) {
 			resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 			si := clients.ServerIdentifier{
 				ServerURI:  mgmtServer.Address,
-				Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+				Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 			}
 
-			credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+			configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 			xdsClientConfig := xdsclient.Config{
 				Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 				Node:             clients.Node{ID: nodeID},
-				TransportBuilder: grpctransport.NewBuilder(credentials),
+				TransportBuilder: grpctransport.NewBuilder(configs),
 				ResourceTypes:    resourceTypes,
 				// Xdstp resource names used in this test do not specify an
 				// authority. These will end up looking up an entry with the
@@ -398,14 +397,14 @@ func (s) TestLDSWatch_TwoWatchesForSameResourceName(t *testing.T) {
 			resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 			si := clients.ServerIdentifier{
 				ServerURI:  mgmtServer.Address,
-				Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+				Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 			}
 
-			credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+			configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 			xdsClientConfig := xdsclient.Config{
 				Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 				Node:             clients.Node{ID: nodeID},
-				TransportBuilder: grpctransport.NewBuilder(credentials),
+				TransportBuilder: grpctransport.NewBuilder(configs),
 				ResourceTypes:    resourceTypes,
 				// Xdstp resource names used in this test do not specify an
 				// authority. These will end up looking up an entry with the
@@ -503,14 +502,14 @@ func (s) TestLDSWatch_ThreeWatchesForDifferentResourceNames(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
@@ -610,14 +609,14 @@ func (s) TestLDSWatch_ResourceCaching(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -684,7 +683,7 @@ func (s) TestLDSWatch_ResourceCaching(t *testing.T) {
 // does not receive an LDS response for the request that it sends. The test
 // verifies that the watch callback is invoked with an error once the
 // watchExpiryTimer fires.
-func (s) TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
+func TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
 	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
 
 	nodeID := uuid.New().String()
@@ -692,14 +691,14 @@ func (s) TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -743,14 +742,14 @@ func (s) TestLDSWatch_ValidResponseCancelsExpiryTimerBehavior(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -820,14 +819,14 @@ func (s) TestLDSWatch_ResourceRemoved(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
@@ -949,14 +948,14 @@ func (s) TestLDSWatch_NewWatcherForRemovedResource(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -1035,14 +1034,14 @@ func (s) TestLDSWatch_NACKError(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -1103,14 +1102,14 @@ func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 	}
 
@@ -1196,14 +1195,14 @@ func (s) TestLDSWatch_PartialValid(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
@@ -1286,14 +1285,14 @@ func (s) TestLDSWatch_PartialResponse(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
diff --git a/xds/internal/clients/xdsclient/test/metrics_test.go b/xds/internal/clients/xdsclient/test/metrics_test.go
new file mode 100644
index 00000000..00857361
--- /dev/null
+++ b/xds/internal/clients/xdsclient/test/metrics_test.go
@@ -0,0 +1,284 @@
+/*
+ *
+ * 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 xdsclient_test
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"github.com/google/uuid"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal/testutils"
+	"google.golang.org/grpc/xds/internal/clients"
+	"google.golang.org/grpc/xds/internal/clients/grpctransport"
+	"google.golang.org/grpc/xds/internal/clients/internal/testutils/e2e"
+	"google.golang.org/grpc/xds/internal/clients/xdsclient"
+	"google.golang.org/grpc/xds/internal/clients/xdsclient/internal/xdsresource"
+	"google.golang.org/grpc/xds/internal/clients/xdsclient/metrics"
+
+	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+)
+
+// TestResourceUpdateMetrics configures an xDS client, and a management server
+// to send valid and invalid LDS updates, and verifies that the expected metrics
+// for both good and bad updates are emitted.
+func (s) TestResourceUpdateMetrics(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout*1000)
+	defer cancel()
+
+	tmr := newTestMetricsReporter()
+	l, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("net.Listen() failed: %v", err)
+	}
+
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{Listener: l})
+	const listenerResourceName = "test-listener-resource"
+	const routeConfigurationName = "test-route-configuration-resource"
+	nodeID := uuid.New().String()
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(listenerResourceName, routeConfigurationName)},
+		SkipValidation: true,
+	}
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
+	}
+
+	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
+	si := clients.ServerIdentifier{
+		ServerURI:  mgmtServer.Address,
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
+	}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+	xdsClientConfig := xdsclient.Config{
+		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
+		Node:             clients.Node{ID: nodeID},
+		TransportBuilder: grpctransport.NewBuilder(configs),
+		ResourceTypes:    resourceTypes,
+		// Xdstp resource names used in this test do not specify an
+		// authority. These will end up looking up an entry with the
+		// empty key in the authorities map. Having an entry with an
+		// empty key and empty configuration, results in these
+		// resources also using the top-level configuration.
+		Authorities: map[string]xdsclient.Authority{
+			"": {XDSServers: []xdsclient.ServerConfig{}},
+		},
+		MetricsReporter: tmr,
+	}
+	// Create an xDS client with the above config.
+	client, err := xdsclient.New(xdsClientConfig)
+	if err != nil {
+		t.Fatalf("Failed to create xDS client: %v", err)
+	}
+	defer client.Close()
+
+	// Watch the valid listener configured on the management server. This should
+	// cause a resource update valid metric to emit eventually.
+	client.WatchResource(listenerType.TypeURL, listenerResourceName, noopListenerWatcher{})
+	if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil {
+		t.Fatal(err.Error())
+	}
+
+	// Update management server with a bad update. This should cause a resource
+	// update invalid metric to emit eventually.
+	resources = e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(listenerResourceName, routeConfigurationName)},
+		SkipValidation: true,
+	}
+	resources.Listeners[0].ApiListener = nil
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
+	}
+	if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateInvalid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil {
+		t.Fatal(err.Error())
+	}
+
+	// Resource update valid metric should have not emitted.
+	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer sCancel()
+	if err := tmr.waitForMetric(sCtx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err == nil {
+		t.Fatal("tmr.WaitForInt64Count(ctx, mdWant) succeeded when expected to timeout.")
+	}
+}
+
+// TestServerFailureMetrics_BeforeResponseRecv configures an xDS client, and a
+// management server. It then register a watcher and stops the management
+// server before sending a resource update, and verifies that the expected
+// metric for server failure is emitted.
+func (s) TestServerFailureMetrics_BeforeResponseRecv(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	tmr := newTestMetricsReporter()
+	l, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("net.Listen() failed: %v", err)
+	}
+
+	lis := testutils.NewRestartableListener(l)
+	streamOpened := make(chan struct{}, 1)
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		Listener: lis,
+		OnStreamOpen: func(context.Context, int64, string) error {
+			select {
+			case streamOpened <- struct{}{}:
+			default:
+			}
+			return nil
+		},
+	})
+
+	nodeID := uuid.New().String()
+
+	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
+	si := clients.ServerIdentifier{
+		ServerURI:  mgmtServer.Address,
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
+	}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+	xdsClientConfig := xdsclient.Config{
+		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
+		Node:             clients.Node{ID: nodeID},
+		TransportBuilder: grpctransport.NewBuilder(configs),
+		ResourceTypes:    resourceTypes,
+		// Xdstp resource names used in this test do not specify an
+		// authority. These will end up looking up an entry with the
+		// empty key in the authorities map. Having an entry with an
+		// empty key and empty configuration, results in these
+		// resources also using the top-level configuration.
+		Authorities: map[string]xdsclient.Authority{
+			"": {XDSServers: []xdsclient.ServerConfig{}},
+		},
+		MetricsReporter: tmr,
+	}
+	// Create an xDS client with the above config.
+	client, err := xdsclient.New(xdsClientConfig)
+	if err != nil {
+		t.Fatalf("Failed to create xDS client: %v", err)
+	}
+	defer client.Close()
+
+	const listenerResourceName = "test-listener-resource"
+
+	// Watch for the listener on the above management server.
+	client.WatchResource(listenerType.TypeURL, listenerResourceName, noopListenerWatcher{})
+	// Verify that an ADS stream is opened and an LDS request with the above
+	// resource name is sent.
+	select {
+	case <-streamOpened:
+	case <-ctx.Done():
+		t.Fatal("Timeout when waiting for ADS stream to open")
+	}
+
+	// Close the listener and ensure that the ADS stream breaks. This should
+	// cause a server failure metric to emit eventually.
+	lis.Stop()
+
+	// Restart to prevent the attempt to create a new ADS stream after back off.
+	lis.Restart()
+
+	if err := tmr.waitForMetric(ctx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err != nil {
+		t.Fatal(err.Error())
+	}
+}
+
+// TestServerFailureMetrics_AfterResponseRecv configures an xDS client, and a
+// management server to send a valid LDS updates, and verifies that the
+// resource update valid metric is emitted. It then closes the management server
+// listener to close the ADS stream and verifies that the server failure metric
+// is not emitted because the ADS stream was closed after having received a
+// response on the stream.
+func (s) TestServerFailureMetrics_AfterResponseRecv(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	tmr := newTestMetricsReporter()
+	l, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("net.Listen() failed: %v", err)
+	}
+
+	lis := testutils.NewRestartableListener(l)
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{Listener: lis})
+	const listenerResourceName = "test-listener-resource"
+	const routeConfigurationName = "test-route-configuration-resource"
+	nodeID := uuid.New().String()
+	resources := e2e.UpdateOptions{
+		NodeID:         nodeID,
+		Listeners:      []*v3listenerpb.Listener{e2e.DefaultClientListener(listenerResourceName, routeConfigurationName)},
+		SkipValidation: true,
+	}
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err)
+	}
+
+	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
+	si := clients.ServerIdentifier{
+		ServerURI:  mgmtServer.Address,
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
+	}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+	xdsClientConfig := xdsclient.Config{
+		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
+		Node:             clients.Node{ID: nodeID},
+		TransportBuilder: grpctransport.NewBuilder(configs),
+		ResourceTypes:    resourceTypes,
+		// Xdstp resource names used in this test do not specify an
+		// authority. These will end up looking up an entry with the
+		// empty key in the authorities map. Having an entry with an
+		// empty key and empty configuration, results in these
+		// resources also using the top-level configuration.
+		Authorities: map[string]xdsclient.Authority{
+			"": {XDSServers: []xdsclient.ServerConfig{}},
+		},
+		MetricsReporter: tmr,
+	}
+	// Create an xDS client with the above config.
+	client, err := xdsclient.New(xdsClientConfig)
+	if err != nil {
+		t.Fatalf("Failed to create xDS client: %v", err)
+	}
+	defer client.Close()
+
+	// Watch the valid listener configured on the management server. This should
+	// cause a resource updates valid count to emit eventually.
+	client.WatchResource(listenerType.TypeURL, listenerResourceName, noopListenerWatcher{})
+	if err := tmr.waitForMetric(ctx, &metrics.ResourceUpdateValid{ServerURI: mgmtServer.Address, ResourceType: "ListenerResource"}); err != nil {
+		t.Fatal(err.Error())
+	}
+
+	// Close the listener and ensure that the ADS stream breaks. This should
+	// cause a server failure metric to emit eventually.
+	lis.Stop()
+	if ctx.Err() != nil {
+		t.Fatalf("Timeout when waiting for ADS stream to close")
+	}
+	// Restart to prevent the attempt to create a new ADS stream after back off.
+	lis.Restart()
+
+	// Server failure should not have emitted.
+	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer sCancel()
+	if err := tmr.waitForMetric(sCtx, &metrics.ServerFailure{ServerURI: mgmtServer.Address}); err == nil {
+		t.Fatal("tmr.WaitForInt64Count(ctx, mdWant) succeeded when expected to timeout.")
+	}
+}
diff --git a/xds/internal/clients/xdsclient/test/misc_watchers_test.go b/xds/internal/clients/xdsclient/test/misc_watchers_test.go
index 39702272..b6d74a56 100644
--- a/xds/internal/clients/xdsclient/test/misc_watchers_test.go
+++ b/xds/internal/clients/xdsclient/test/misc_watchers_test.go
@@ -26,7 +26,6 @@ import (
 	"testing"
 
 	"github.com/google/uuid"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -109,14 +108,14 @@ func (s) TestWatchCallAnotherWatch(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp style resource names used in this test use a slash removed
 		// version of t.Name as their authority, and the empty config
@@ -220,14 +219,14 @@ func (s) TestNodeProtoSentOnlyInFirstRequest(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp resource names used in this test do not specify an
 		// authority. These will end up looking up an entry with the
@@ -385,14 +384,14 @@ func (s) TestWatchErrorsContainNodeID(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	xdsClientConfig := xdsclient.Config{
 		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
 		Node:             clients.Node{ID: nodeID},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
 		// Xdstp resource names used in this test do not specify an
 		// authority. These will end up looking up an entry with the
@@ -483,7 +482,7 @@ func (s) TestWatchErrorsContainNodeID_ChannelCreationFailure(t *testing.T) {
 	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
 	si := clients.ServerIdentifier{
 		ServerURI:  mgmtServer.Address,
-		Extensions: grpctransport.ServerIdentifierExtension{Credentials: "insecure"},
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
 	}
 
 	xdsClientConfig := xdsclient.Config{
diff --git a/xds/internal/clients/xdsclient/xdsclient_test.go b/xds/internal/clients/xdsclient/xdsclient_test.go
index 6c5c7b59..69b64006 100644
--- a/xds/internal/clients/xdsclient/xdsclient_test.go
+++ b/xds/internal/clients/xdsclient/xdsclient_test.go
@@ -22,7 +22,6 @@ import (
 	"strings"
 	"testing"
 
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/xds/internal/clients"
 	"google.golang.org/grpc/xds/internal/clients/grpctransport"
@@ -30,7 +29,7 @@ import (
 )
 
 func (s) TestXDSClient_New(t *testing.T) {
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 
 	tests := []struct {
 		name    string
@@ -62,7 +61,7 @@ func (s) TestXDSClient_New(t *testing.T) {
 			config: Config{
 				Node:             clients.Node{ID: "node-id"},
 				ResourceTypes:    map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
-				TransportBuilder: grpctransport.NewBuilder(credentials),
+				TransportBuilder: grpctransport.NewBuilder(configs),
 			},
 			wantErr: "no servers or authorities specified",
 		},
@@ -71,7 +70,7 @@ func (s) TestXDSClient_New(t *testing.T) {
 			config: Config{
 				Node:             clients.Node{ID: "node-id"},
 				ResourceTypes:    map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
-				TransportBuilder: grpctransport.NewBuilder(credentials),
+				TransportBuilder: grpctransport.NewBuilder(configs),
 				Servers:          []ServerConfig{{ServerIdentifier: clients.ServerIdentifier{ServerURI: "dummy-server"}}},
 			},
 			wantErr: "",
@@ -81,7 +80,7 @@ func (s) TestXDSClient_New(t *testing.T) {
 			config: Config{
 				Node:             clients.Node{ID: "node-id"},
 				ResourceTypes:    map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
-				TransportBuilder: grpctransport.NewBuilder(credentials),
+				TransportBuilder: grpctransport.NewBuilder(configs),
 				Authorities:      map[string]Authority{"authority-name": {XDSServers: []ServerConfig{{ServerIdentifier: clients.ServerIdentifier{ServerURI: "dummy-server"}}}}},
 			},
 			wantErr: "",
@@ -107,11 +106,11 @@ func (s) TestXDSClient_New(t *testing.T) {
 }
 
 func (s) TestXDSClient_Close(t *testing.T) {
-	credentials := map[string]credentials.Bundle{"insecure": insecure.NewBundle()}
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
 	config := Config{
 		Node:             clients.Node{ID: "node-id"},
 		ResourceTypes:    map[string]ResourceType{xdsresource.V3ListenerURL: listenerType},
-		TransportBuilder: grpctransport.NewBuilder(credentials),
+		TransportBuilder: grpctransport.NewBuilder(configs),
 		Servers:          []ServerConfig{{ServerIdentifier: clients.ServerIdentifier{ServerURI: "dummy-server"}}},
 	}
 	c, err := New(config)