Repository: grpc/grpc-go

Issue:
Flaky test: Test/RingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartup
```
--- FAIL: Test (23.36s)
    --- FAIL: Test/RingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartup (10.01s)
        tlogger.go:116: INFO server.go:685 [core] [Server #409]Server created  (t=+276.144µs)
        stubserver.go:288: Started test service backend at "127.0.0.1:44079"
        tlogger.go:116: INFO server.go:881 [core] [Server #409 ListenSocket #410]ListenSocket created  (t=+423.177µs)
        tlogger.go:116: INFO restartable_listener.go:84 [testutils] Stopping restartable listener "127.0.0.1:34327"  (t=+762.212µs)
        ringhash_balancer_test.go:438: Created new snapshot cache...
        tlogger.go:116: INFO server.go:685 [core] [Server #411]Server created  (t=+1.204154ms)
        ringhash_balancer_test.go:438: Registered Aggregated Discovery Service (ADS)...
        ringhash_balancer_test.go:438: xDS management server serving at: 127.0.0.1:35659...
        tlogger.go:116: INFO server.go:881 [core] [Server #411 ListenSocket #412]ListenSocket created  (t=+1.870166ms)
        server.go:[229](https://github.com/grpc/grpc-go/actions/runs/12684369727/job/35352925778?pr=7992#step:8:230): Created new resource snapshot...
        server.go:235: Updated snapshot cache with resource snapshot...
        tlogger.go:116: INFO clientconn.go:1674 [core] original dial target is: "xds:///test.server"  (t=+3.552713ms)
        tlogger.go:116: INFO clientconn.go:314 [core] [Channel #413]Channel created  (t=+3.622452ms)
        tlogger.go:116: INFO clientconn.go:193 [core] [Channel #413]parsed dial target is: resolver.Target{URL:url.URL{Scheme:"xds", Opaque:"", User:(*url.Userinfo)(nil), Host:"", Path:"/test.server", RawPath:"", OmitHost:false, ForceQuery:false, RawQuery:"", Fragment:"", RawFragment:""}}  (t=+3.73948ms)
        tlogger.go:116: INFO clientconn.go:194 [core] [Channel #413]Channel authority set to "test.server"  (t=+3.798049ms)
        tlogger.go:116: INFO xds_resolver.go:100 [xds] [xds-resolver 0xc000785320] Creating resolver for target: xds:///test.server  (t=+4.049948ms)
        tlogger.go:116: INFO client_refcounted.go:76 [xds] [xds-client 0xc000389a40] Created client with name "xds:///test.server" and bootstrap configuration:
             {
              "xds_servers": [
               {
                "server_uri": "passthrough:///127.0.0.1:35659",
                "channel_creds": [
                 {
                  "type": "insecure"
                 }
                ]
               }
              ],
              "certificate_providers": {
               "client-side-certificate-provider-instance": {
                "plugin_name": "file_watcher",
                "config": {
                 "certificate_file": "/tmp/testClientSideXDS35[230](https://github.com/grpc/grpc-go/actions/runs/12684369727/job/35352925778?pr=7992#step:8:231)94675/cert.pem",
                 "private_key_file": "/tmp/testClientSideXDS3523094675/key.pem",
                 "ca_certificate_file": "/tmp/testClientSideXDS3523094675/ca.pem",
                 "refresh_interval": "600s"
                }
               },
               "server-side-certificate-provider-instance": {
                "plugin_name": "file_watcher",
                "config": {
                 "certificate_file": "/tmp/testServerSideXDS420696777/cert.pem",
                 "private_key_file": "/tmp/testServerSideXDS420696777/key.pem",
                 "ca_certificate_file": "/tmp/testServerSideXDS420696777/ca.pem",
                 "refresh_interval": "600s"
                }
               }
              },
              "server_listener_resource_name_template": "grpc/server?xds.resource.listening_address=%s",
              "client_default_listener_resource_name_template": "%s",
              "node": {
               "id": "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed",
               "locality": {}
              }
             }  (t=+5.461273ms)
        tlogger.go:116: INFO client_refcounted.go:81 [xds] xDS node ID: 9deb4ef4-9391-4307-b8ba-e560cb9cc3ed  (t=+5.576247ms)
        tlogger.go:116: INFO clientconn.go:1674 [core] original dial target is: "passthrough:///127.0.0.1:35659"  (t=+5.729802ms)
        tlogger.go:116: INFO clientconn.go:314 [core] [Channel #414]Channel created  (t=+5.829618ms)
        tlogger.go:116: INFO clientconn.go:193 [core] [Channel #414]parsed dial target is: resolver.Target{URL:url.URL{Scheme:"passthrough", Opaque:"", User:(*url.Userinfo)(nil), Host:"", Path:"/127.0.0.1:35659", RawPath:"", OmitHost:false, ForceQuery:false, RawQuery:"", Fragment:"", RawFragment:""}}  (t=+5.969428ms)
        tlogger.go:116: INFO clientconn.go:194 [core] [Channel #414]Channel authority set to "127.0.0.1:35659"  (t=+6.044247ms)
        tlogger.go:116: INFO resolver_wrapper.go:200 [core] [Channel #414]Resolver state updated: {
              "Addresses": [
                {
                  "Addr": "127.0.0.1:35659",
                  "ServerName": "",
                  "Attributes": null,
                  "BalancerAttributes": null,
                  "Metadata": null
                }
              ],
              "Endpoints": [
                {
                  "Addresses": [
                    {
                      "Addr": "127.0.0.1:35659",
                      "ServerName": "",
                      "Attributes": null,
                      "BalancerAttributes": null,
                      "Metadata": null
                    }
                  ],
                  "Attributes": null
                }
              ],
              "ServiceConfig": null,
              "Attributes": null
            } (resolver returned new addresses)  (t=+6.435836ms)
        tlogger.go:116: INFO balancer_wrapper.go:117 [core] [Channel #414]Channel switches to new LB policy "pick_first"  (t=+6.641057ms)
        tlogger.go:116: INFO clientconn.go:544 [core] [Channel #414]Channel Connectivity change to CONNECTING  (t=+6.738749ms)
        tlogger.go:116: INFO clientconn.go:858 [core] [Channel #414 SubChannel #415]Subchannel created  (t=+6.848183ms)
        tlogger.go:116: INFO clientconn.go:1219 [core] [Channel #414 SubChannel #415]Subchannel Connectivity change to CONNECTING  (t=+6.952096ms)
        tlogger.go:116: INFO clientconn.go:1339 [core] [Channel #414 SubChannel #415]Subchannel picks a new address "127.0.0.1:35659" to connect  (t=+7.016496ms)
        tlogger.go:116: INFO clientconn.go:314 [core] [Channel #414]Channel exiting idle mode  (t=+7.827194ms)
        tlogger.go:116: INFO clientconn.go:314 [core] [Channel #413]Channel exiting idle mode  (t=+8.216047ms)
        tlogger.go:116: INFO clientconn.go:1219 [core] [Channel #414 SubChannel #415]Subchannel Connectivity change to READY  (t=+10.783082ms)
        tlogger.go:116: INFO clientconn.go:544 [core] [Channel #414]Channel Connectivity change to READY  (t=+11.036874ms)
        logging.go:30: nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed" requested type.googleapis.com/envoy.config.listener.v3.Listener[test.server] and known map[]. Diff [test.server]
        logging.go:30: respond type.googleapis.com/envoy.config.listener.v3.Listener[test.server] version "" with version "1"
        logging.go:30: nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed" requested type.googleapis.com/envoy.config.listener.v3.Listener[test.server] and known map[test.server:{}]. Diff []
        logging.go:30: open watch 1 for type.googleapis.com/envoy.config.listener.v3.Listener[test.server] from nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed", version "1"
        logging.go:30: nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed" requested type.googleapis.com/envoy.config.route.v3.RouteConfiguration[new_route] and known map[]. Diff [new_route]
        logging.go:30: respond type.googleapis.com/envoy.config.route.v3.RouteConfiguration[new_route] version "" with version "1"
        tlogger.go:116: INFO xds_resolver.go:284 [xds] [xds-resolver 0xc000785320] For Listener resource "test.server" and RouteConfiguration resource "new_route", generated service config: {
              "loadBalancingConfig": [
                {
                  "xds_cluster_manager_experimental": {
                    "children": {
                      "cluster:aggregate_cluster": {
                        "childPolicy": [
                          {
                            "cds_experimental": {
                              "cluster": "aggregate_cluster"
                            }
                          }
                        ]
                      }
                    }
                  }
                }
              ]
            }  (t=+14.317477ms)
        tlogger.go:116: INFO resolver_wrapper.go:200 [core] [Channel #413]Resolver state updated: {
              "Addresses": null,
              "Endpoints": [],
              "ServiceConfig": {
                "Config": {
                  "Config": null,
                  "Methods": {}
                },
                "Err": null
              },
              "Attributes": {
                "\u003c%!p(resolver.csKeyType=grpc.internal.resolver.configSelector)\u003e": "\u003c0xc0006f60a0\u003e",
                "\u003c%!p(xdsclient.clientKeyType=grpc.xds.internal.client.Client)\u003e": "\u003c0xc000739c40\u003e"
              }
            } (service config updated)  (t=+14.696121ms)
        tlogger.go:116: INFO balancer_wrapper.go:117 [core] [Channel #413]Channel switches to new LB policy "xds_cluster_manager_experimental"  (t=+14.779026ms)
        tlogger.go:116: INFO clustermanager.go:59 [xds] [xds-cluster-manager-lb 0xc000160cc0] Created  (t=+14.828808ms)
        tlogger.go:116: INFO balancergroup.go:299 [xds] [xds-cluster-manager-lb 0xc000160cc0] Adding child policy of type "cds_experimental" for child "cluster:aggregate_cluster"  (t=+14.879082ms)
        tlogger.go:116: INFO balancergroup.go:102 [xds] [xds-cluster-manager-lb 0xc000160cc0] Creating child policy of type "cds_experimental" for child "cluster:aggregate_cluster"  (t=+14.918615ms)
        tlogger.go:116: INFO cdsbalancer.go:109 [xds] [cds-lb 0xc000247520] Created  (t=+14.966885ms)
        logging.go:30: nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed" requested type.googleapis.com/envoy.config.route.v3.RouteConfiguration[new_route] and known map[new_route:{}]. Diff []
        logging.go:30: open watch 2 for type.googleapis.com/envoy.config.route.v3.RouteConfiguration[new_route] from nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed", version "1"
        tlogger.go:116: INFO cdsbalancer.go:121 [xds] [cds-lb 0xc000247520] xDS credentials in use: false  (t=+15.364675ms)
        tlogger.go:116: INFO cdsbalancer.go:290 [xds] [cds-lb 0xc000247520] Received balancer config update: {
              "LoadBalancingConfig": null,
              "Cluster": "aggregate_cluster"
            }  (t=+15.460233ms)
        tlogger.go:116: INFO balancerstateaggregator.go:191 [xds] [xds-cluster-manager-lb 0xc000160cc0] Child pickers: map[cluster:aggregate_cluster:picker:0xc00077a670,state:CONNECTING,stateToAggregate:CONNECTING]  (t=+15.634617ms)
        tlogger.go:116: INFO clientconn.go:544 [core] [Channel #413]Channel Connectivity change to CONNECTING  (t=+15.713023ms)
        logging.go:30: nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed" requested type.googleapis.com/envoy.config.cluster.v3.Cluster[aggregate_cluster] and known map[]. Diff [aggregate_cluster]
        logging.go:30: respond type.googleapis.com/envoy.config.cluster.v3.Cluster[aggregate_cluster] version "" with version "1"
        logging.go:30: nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed" requested type.googleapis.com/envoy.config.cluster.v3.Cluster[aggregate_cluster] and known map[aggregate_cluster:{}]. Diff []
        logging.go:30: open watch 3 for type.googleapis.com/envoy.config.cluster.v3.Cluster[aggregate_cluster] from nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed", version "1"
        tlogger.go:116: INFO cdsbalancer.go:414 [xds] [cds-lb 0xc000247520] Received Cluster resource: {
              "ClusterType": 2,
              "ClusterName": "aggregate_cluster",
              "EDSServiceName": "",
              "LRSServerConfig": null,
              "SecurityCfg": null,
              "MaxRequests": null,
              "DNSHostName": "",
              "PrioritizedClusterNames": [
                "eds_cluster",
                "logical_dns_cluster"
              ],
              "LBPolicy": [
                {
                  "ring_hash_experimental": {
                    "minRingSize": 1024,
                    "maxRingSize": 8388608
                  }
                }
              ],
              "OutlierDetection": null,
              "Raw": {
                "type_url": "type.googleapis.com/envoy.config.cluster.v3.Cluster",
                "value": "ChFhZ2dyZWdhdGVfY2x1c3RlcjACsgKKAQoYZW52b3kuY2x1c3RlcnMuYWdncmVnYXRlEm4KSHR5cGUuZ29vZ2xlYXBpcy5jb20vZW52b3kuZXh0ZW5zaW9ucy5jbHVzdGVycy5hZ2dyZWdhdGUudjMuQ2x1c3RlckNvbmZpZxIiCgtlZHNfY2x1c3RlcgoTbG9naWNhbF9kbnNfY2x1c3Rlcg=="
              },
              "TelemetryLabels": {
                "csm.service_name": "unknown",
                "csm.service_namespace_name": "unknown"
              }
            }  (t=+16.746595ms)
        logging.go:30: nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed" requested type.googleapis.com/envoy.config.cluster.v3.Cluster[aggregate_cluster eds_cluster] and known map[aggregate_cluster:{}]. Diff [eds_cluster]
        logging.go:30: respond type.googleapis.com/envoy.config.cluster.v3.Cluster[aggregate_cluster eds_cluster] version "1" with version "1"
        tlogger.go:116: INFO cdsbalancer.go:414 [xds] [cds-lb 0xc000247520] Received Cluster resource: {
              "ClusterType": 0,
              "ClusterName": "eds_cluster",
              "EDSServiceName": "eds_cluster",
              "LRSServerConfig": null,
              "SecurityCfg": null,
              "MaxRequests": null,
              "DNSHostName": "",
              "PrioritizedClusterNames": null,
              "LBPolicy": [
                {
                  "xds_wrr_locality_experimental": {
                    "childPolicy": [
                      {
                        "round_robin": {}
                      }
                    ]
                  }
                }
              ],
              "OutlierDetection": null,
              "Raw": {
                "type_url": "type.googleapis.com/envoy.config.cluster.v3.Cluster",
                "value": "CgtlZHNfY2x1c3RlchoRCgIaABILZWRzX2NsdXN0ZXIQAw=="
              },
              "TelemetryLabels": {
                "csm.service_name": "unknown",
                "csm.service_namespace_name": "unknown"
              }
            }  (t=+18.182486ms)
        logging.go:30: nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed" requested type.googleapis.com/envoy.config.cluster.v3.Cluster[aggregate_cluster eds_cluster logical_dns_cluster] and known map[aggregate_cluster:{} eds_cluster:{}]. Diff [logical_dns_cluster]
        logging.go:30: respond type.googleapis.com/envoy.config.cluster.v3.Cluster[aggregate_cluster eds_cluster logical_dns_cluster] version "1" with version "1"
        tlogger.go:116: INFO cdsbalancer.go:414 [xds] [cds-lb 0xc000247520] Received Cluster resource: {
              "ClusterType": 1,
              "ClusterName": "logical_dns_cluster",
              "EDSServiceName": "",
              "LRSServerConfig": null,
              "SecurityCfg": null,
              "MaxRequests": null,
              "DNSHostName": "server.example.com:443",
              "PrioritizedClusterNames": null,
              "LBPolicy": [
                {
                  "xds_wrr_locality_experimental": {
                    "childPolicy": [
                      {
                        "round_robin": {}
                      }
                    ]
                  }
                }
              ],
              "OutlierDetection": null,
              "Raw": {
                "type_url": "type.googleapis.com/envoy.config.cluster.v3.Cluster",
                "value": "ChNsb2dpY2FsX2Ruc19jbHVzdGVyigIhEh8SHQobChkKFxISc2VydmVyLmV4YW1wbGUuY29tGLsDEAI="
              },
              "TelemetryLabels": {
                "csm.service_name": "unknown",
                "csm.service_namespace_name": "unknown"
              }
            }  (t=+19.145708ms)
        logging.go:30: nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed" requested type.googleapis.com/envoy.config.cluster.v3.Cluster[aggregate_cluster eds_cluster logical_dns_cluster] and known map[aggregate_cluster:{} eds_cluster:{} logical_dns_cluster:{}]. Diff []
        logging.go:30: open watch 4 for type.googleapis.com/envoy.config.cluster.v3.Cluster[aggregate_cluster eds_cluster logical_dns_cluster] from nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed", version "1"
        tlogger.go:116: INFO clusterresolver.go:85 [xds] [xds-cluster-resolver-lb 0xc0008cea80] Created  (t=+19.61966ms)
        tlogger.go:116: INFO cdsbalancer.go:455 [xds] [cds-lb 0xc000247520] Created child policy 0xc0008cea80 of type cluster_resolver_experimental  (t=+19.708395ms)
        ringhash_balancer_test.go:460: rpc EmptyCall() failed: rpc error: code = DeadlineExceeded desc = received context error while waiting for new LB policy update: context deadline exceeded
        tlogger.go:116: INFO clientconn.go:544 [core] [Channel #413]Channel Connectivity change to SHUTDOWN  (t=+10.008614258s)
        tlogger.go:116: INFO resolver_wrapper.go:100 [core] [Channel #413]Closing the name resolver  (t=+10.008656096s)
        tlogger.go:116: INFO balancer_wrapper.go:155 [core] [Channel #413]ccBalancerWrapper: closing  (t=+10.008718722s)
        tlogger.go:116: INFO watch_service.go:56 [xds] [xds-resolver 0xc000785320] Canceling watch on Listener resource "test.server"  (t=+10.008880952s)
        tlogger.go:116: INFO watch_service.go:91 [xds] [xds-resolver 0xc000785320] Canceling watch on RouteConfiguration resource "new_route"  (t=+10.00908305s)
        tlogger.go:116: INFO clusterresolver.go:355 [xds] [xds-cluster-resolver-lb 0xc0008cea80] Shutdown  (t=+10.00918091s)
        tlogger.go:116: WARNING ads_stream.go:514 [xds] [xds-client 0xc000389a40] [xds-channel 0xc00064c120] [ads-stream 0xc0006f0800] ADS stream closed: rpc error: code = Canceled desc = context canceled  (t=+10.009[232](https://github.com/grpc/grpc-go/actions/runs/12684369727/job/35352925778?pr=7992#step:8:233)788s)
        tlogger.go:116: INFO cdsbalancer.go:380 [xds] [cds-lb 0xc000247520] Shutdown  (t=+10.009283252s)
        tlogger.go:116: INFO ads_stream.go:178 [xds] [xds-client 0xc000389a40] [xds-channel 0xc00064c120] [ads-stream 0xc0006f0800] Stopping ADS stream  (t=+10.009327324s)
        logging.go:30: nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed" requested type.googleapis.com/envoy.config.cluster.v3.Cluster[eds_cluster logical_dns_cluster] and known map[aggregate_cluster:{} eds_cluster:{} logical_dns_cluster:{}]. Diff []
        tlogger.go:116: INFO clientconn.go:544 [core] [Channel #414]Channel Connectivity change to SHUTDOWN  (t=+10.009384731s)
        logging.go:30: open watch 5 for type.googleapis.com/envoy.config.cluster.v3.Cluster[eds_cluster logical_dns_cluster] from nodeID "9deb4ef4-9391-4307-b8ba-e560cb9cc3ed", version "1"
        tlogger.go:116: INFO resolver_wrapper.go:100 [core] [Channel #414]Closing the name resolver  (t=+10.009489346s)
        tlogger.go:116: INFO clustermanager.go:197 [xds] [xds-cluster-manager-lb 0xc000160cc0] Shutdown  (t=+10.009563402s)
        tlogger.go:116: INFO balancer_wrapper.go:155 [core] [Channel #414]ccBalancerWrapper: closing  (t=+10.009610319s)
        tlogger.go:116: INFO clientconn.go:1219 [core] [Channel #414 SubChannel #415]Subchannel Connectivity change to SHUTDOWN  (t=+10.009743041s)
        tlogger.go:116: INFO clientconn.go:1547 [core] [Channel #414 SubChannel #415]Subchannel deleted  (t=+10.009789337s)
        tlogger.go:116: INFO clientconn.go:314 [core] [Channel #414]Channel deleted  (t=+10.010027204s)
        tlogger.go:116: INFO channel.go:145 [xds] [xds-client 0xc000389a40] [xds-channel 0xc00064c120] Shutdown  (t=+10.010089721s)
        tlogger.go:116: INFO clientimpl.go:185 [xds] [xds-client 0xc000389a40] Shutdown  (t=+10.010159192s)
        tlogger.go:116: INFO xds_resolver.go:253 [xds] [xds-resolver 0xc000785320] Shutdown  (t=+10.01019071s)
        tlogger.go:116: INFO clientconn.go:314 [core] [Channel #413]Channel deleted  (t=+10.010[238](https://github.com/grpc/grpc-go/actions/runs/12684369727/job/35352925778?pr=7992#step:8:239)539s)
        tlogger.go:116: INFO server.go:817 [core] [Server #411 ListenSocket #412]ListenSocket deleted  (t=+10.010355049s)
        tlogger.go:116: INFO server.go:817 [core] [Server #409 ListenSocket #410]ListenSocket deleted  (t=+10.010458542s)
FAIL
```

Hints from issue comments:
@arjan-bal: In the success case, the next few log lines after `Created child policy 0x14000530820 of type cluster_resolver_experimental` are similar to the following:
> logging.go:30: nodeID "267a4abd-bf2b-4af8-9119-b88eeb7e1d98" requested type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment[cluster] and known map[]. Diff [cluster]
    logging.go:30: respond type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment[cluster] version "" with version "1"
    tlogger.go:116: INFO balancer.go:76 [xds] [priority-lb 0xc000378230] Created  (t=+9.91661ms)

It seems like that clusterresolver didn't start a watch for EDS resources.

Able to repro the failure on forge in 7600 runs.

Solution patch:
diff --git a/balancer/weightedroundrobin/weightedroundrobin.go b/balancer/weightedroundrobin/weightedroundrobin.go
index 258cdd5d..f50c2ef2 100644
--- a/balancer/weightedroundrobin/weightedroundrobin.go
+++ b/balancer/weightedroundrobin/weightedroundrobin.go
@@ -34,7 +34,8 @@ import (
 )
 
 // attributeKey is the type used as the key to store AddrInfo in the
-// BalancerAttributes field of resolver.Address.
+// BalancerAttributes field of resolver.Address or Attributes field of
+// resolver.Endpoint.
 type attributeKey struct{}
 
 // AddrInfo will be stored in the BalancerAttributes field of Address in order
@@ -71,6 +72,14 @@ func GetAddrInfo(addr resolver.Address) AddrInfo {
 	return ai
 }
 
+// AddrInfoFromEndpoint returns the AddrInfo stored in the Attributes field of
+// endpoint.
+func AddrInfoFromEndpoint(endpoint resolver.Endpoint) AddrInfo {
+	v := endpoint.Attributes.Value(attributeKey{})
+	ai, _ := v.(AddrInfo)
+	return ai
+}
+
 func (a AddrInfo) String() string {
 	return fmt.Sprintf("Weight: %d", a.Weight)
 }
diff --git a/xds/internal/balancer/ringhash/logging.go b/xds/internal/balancer/ringhash/logging.go
index 3e0f0adf..64a1d467 100644
--- a/xds/internal/balancer/ringhash/logging.go
+++ b/xds/internal/balancer/ringhash/logging.go
@@ -32,7 +32,3 @@ var logger = grpclog.Component("xds")
 func prefixLogger(p *ringhashBalancer) *internalgrpclog.PrefixLogger {
 	return internalgrpclog.NewPrefixLogger(logger, fmt.Sprintf(prefix, p))
 }
-
-func subConnPrefixLogger(p *ringhashBalancer, sc *subConn) *internalgrpclog.PrefixLogger {
-	return internalgrpclog.NewPrefixLogger(logger, fmt.Sprintf(prefix, p)+fmt.Sprintf("[subConn %p] ", sc))
-}
diff --git a/xds/internal/balancer/ringhash/picker.go b/xds/internal/balancer/ringhash/picker.go
index 5ce72cad..fc6bf675 100644
--- a/xds/internal/balancer/ringhash/picker.go
+++ b/xds/internal/balancer/ringhash/picker.go
@@ -22,140 +22,44 @@ import (
 	"fmt"
 
 	"google.golang.org/grpc/balancer"
-	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/internal/grpclog"
-	"google.golang.org/grpc/status"
 )
 
 type picker struct {
-	ring          *ring
-	logger        *grpclog.PrefixLogger
-	subConnStates map[*subConn]connectivity.State
-}
-
-func newPicker(ring *ring, logger *grpclog.PrefixLogger) *picker {
-	states := make(map[*subConn]connectivity.State)
-	for _, e := range ring.items {
-		states[e.sc] = e.sc.effectiveState()
-	}
-	return &picker{ring: ring, logger: logger, subConnStates: states}
-}
-
-// handleRICSResult is the return type of handleRICS. It's needed to wrap the
-// returned error from Pick() in a struct. With this, if the return values are
-// `balancer.PickResult, error, bool`, linter complains because error is not the
-// last return value.
-type handleRICSResult struct {
-	pr  balancer.PickResult
-	err error
-}
-
-// handleRICS generates pick result if the entry is in Ready, Idle, Connecting
-// or Shutdown. TransientFailure will be handled specifically after this
-// function returns.
-//
-// The first return value indicates if the state is in Ready, Idle, Connecting
-// or Shutdown. If it's true, the PickResult and error should be returned from
-// Pick() as is.
-func (p *picker) handleRICS(e *ringEntry) (handleRICSResult, bool) {
-	switch state := p.subConnStates[e.sc]; state {
-	case connectivity.Ready:
-		return handleRICSResult{pr: balancer.PickResult{SubConn: e.sc.sc}}, true
-	case connectivity.Idle:
-		// Trigger Connect() and queue the pick.
-		e.sc.queueConnect()
-		return handleRICSResult{err: balancer.ErrNoSubConnAvailable}, true
-	case connectivity.Connecting:
-		return handleRICSResult{err: balancer.ErrNoSubConnAvailable}, true
-	case connectivity.TransientFailure:
-		// Return ok==false, so TransientFailure will be handled afterwards.
-		return handleRICSResult{}, false
-	case connectivity.Shutdown:
-		// Shutdown can happen in a race where the old picker is called. A new
-		// picker should already be sent.
-		return handleRICSResult{err: balancer.ErrNoSubConnAvailable}, true
-	default:
-		// Should never reach this. All the connectivity states are already
-		// handled in the cases.
-		p.logger.Errorf("SubConn has undefined connectivity state: %v", state)
-		return handleRICSResult{err: status.Errorf(codes.Unavailable, "SubConn has undefined connectivity state: %v", state)}, true
-	}
+	ring   *ring
+	logger *grpclog.PrefixLogger
+	// endpointStates is a cache of endpoint connectivity states and pickers.
+	// The ringhash balancer stores endpoint states in a `resolver.EndpointMap`,
+	// with access guarded by `ringhashBalancer.mu`. The `endpointStates` cache
+	// in the picker helps avoid locking the ringhash balancer's mutex when
+	// reading the latest state at RPC time.
+	endpointStates map[string]balancer.State // endpointState.firstAddr -> balancer.State
 }
 
 func (p *picker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
 	e := p.ring.pick(getRequestHash(info.Ctx))
-	if hr, ok := p.handleRICS(e); ok {
-		return hr.pr, hr.err
-	}
-	// ok was false, the entry is in transient failure.
-	return p.handleTransientFailure(e)
-}
-
-func (p *picker) handleTransientFailure(e *ringEntry) (balancer.PickResult, error) {
-	// Queue a connect on the first picked SubConn.
-	e.sc.queueConnect()
-
-	// Find next entry in the ring, skipping duplicate SubConns.
-	e2 := nextSkippingDuplicates(p.ring, e)
-	if e2 == nil {
-		// There's no next entry available, fail the pick.
-		return balancer.PickResult{}, fmt.Errorf("the only SubConn is in Transient Failure")
-	}
-
-	// For the second SubConn, also check Ready/Idle/Connecting as if it's the
-	// first entry.
-	if hr, ok := p.handleRICS(e2); ok {
-		return hr.pr, hr.err
-	}
-
-	// The second SubConn is also in TransientFailure. Queue a connect on it.
-	e2.sc.queueConnect()
-
-	// If it gets here, this is after the second SubConn, and the second SubConn
-	// was in TransientFailure.
-	//
-	// Loop over all other SubConns:
-	// - If all SubConns so far are all TransientFailure, trigger Connect() on
-	// the TransientFailure SubConns, and keep going.
-	// - If there's one SubConn that's not in TransientFailure, keep checking
-	// the remaining SubConns (in case there's a Ready, which will be returned),
-	// but don't not trigger Connect() on the other SubConns.
-	var firstNonFailedFound bool
-	for ee := nextSkippingDuplicates(p.ring, e2); ee != e; ee = nextSkippingDuplicates(p.ring, ee) {
-		scState := p.subConnStates[ee.sc]
-		if scState == connectivity.Ready {
-			return balancer.PickResult{SubConn: ee.sc.sc}, nil
-		}
-		if firstNonFailedFound {
-			continue
-		}
-		if scState == connectivity.TransientFailure {
-			// This will queue a connect.
-			ee.sc.queueConnect()
-			continue
-		}
-		// This is a SubConn in a non-failure state. We continue to check the
-		// other SubConns, but remember that there was a non-failed SubConn
-		// seen. After this, Pick() will never trigger any SubConn to Connect().
-		firstNonFailedFound = true
-		if scState == connectivity.Idle {
-			// This is the first non-failed SubConn, and it is in a real Idle
-			// state. Trigger it to Connect().
-			ee.sc.queueConnect()
+	ringSize := len(p.ring.items)
+	// Per gRFC A61, because of sticky-TF with PickFirst's auto reconnect on TF,
+	// we ignore all TF subchannels and find the first ring entry in READY,
+	// CONNECTING or IDLE.  If that entry is in IDLE, we need to initiate a
+	// connection. The idlePicker returned by the LazyLB or the new Pickfirst
+	// should do this automatically.
+	for i := 0; i < ringSize; i++ {
+		index := (e.idx + i) % ringSize
+		balState := p.balancerState(p.ring.items[index])
+		switch balState.ConnectivityState {
+		case connectivity.Ready, connectivity.Connecting, connectivity.Idle:
+			return balState.Picker.Pick(info)
+		case connectivity.TransientFailure:
+		default:
+			panic(fmt.Sprintf("Found child balancer in unknown state: %v", balState.ConnectivityState))
 		}
 	}
-	return balancer.PickResult{}, fmt.Errorf("no connection is Ready")
+	// All children are in transient failure. Return the first failure.
+	return p.balancerState(e).Picker.Pick(info)
 }
 
-// nextSkippingDuplicates finds the next entry in the ring, with a different
-// subconn from the given entry.
-func nextSkippingDuplicates(ring *ring, entry *ringEntry) *ringEntry {
-	for next := ring.next(entry); next != entry; next = ring.next(next) {
-		if next.sc != entry.sc {
-			return next
-		}
-	}
-	// There's no qualifying next entry.
-	return nil
+func (p *picker) balancerState(e *ringEntry) balancer.State {
+	return p.endpointStates[e.firstAddr]
 }
diff --git a/xds/internal/balancer/ringhash/ring.go b/xds/internal/balancer/ringhash/ring.go
index 45dbb2d2..c2e556bb 100644
--- a/xds/internal/balancer/ringhash/ring.go
+++ b/xds/internal/balancer/ringhash/ring.go
@@ -32,24 +32,26 @@ type ring struct {
 	items []*ringEntry
 }
 
-type subConnWithWeight struct {
-	sc     *subConn
-	weight float64
+type endpointInfo struct {
+	firstAddr      string
+	scaledWeight   float64
+	originalWeight uint32
 }
 
 type ringEntry struct {
-	idx  int
-	hash uint64
-	sc   *subConn
+	idx       int
+	hash      uint64
+	firstAddr string
+	weight    uint32
 }
 
-// newRing creates a ring from the subConns stored in the AddressMap. The ring
+// newRing creates a ring from the endpoints stored in the EndpointMap. The ring
 // size is limited by the passed in max/min.
 //
-// ring entries will be created for each subConn, and subConn with high weight
-// (specified by the address) may have multiple entries.
+// ring entries will be created for each endpoint, and endpoints with high
+// weight (specified by the address) may have multiple entries.
 //
-// For example, for subConns with weights {a:3, b:3, c:4}, a generated ring of
+// For example, for endpoints with weights {a:3, b:3, c:4}, a generated ring of
 // size 10 could be:
 // - {idx:0 hash:3689675255460411075  b}
 // - {idx:1 hash:4262906501694543955  c}
@@ -65,16 +67,16 @@ type ringEntry struct {
 // To pick from a ring, a binary search will be done for the given target hash,
 // and first item with hash >= given hash will be returned.
 //
-// Must be called with a non-empty subConns map.
-func newRing(subConns *resolver.AddressMap, minRingSize, maxRingSize uint64, logger *grpclog.PrefixLogger) *ring {
+// Must be called with a non-empty endpoints map.
+func newRing(endpoints *resolver.EndpointMap, minRingSize, maxRingSize uint64, logger *grpclog.PrefixLogger) *ring {
 	if logger.V(2) {
-		logger.Infof("newRing: number of subConns is %d, minRingSize is %d, maxRingSize is %d", subConns.Len(), minRingSize, maxRingSize)
+		logger.Infof("newRing: number of endpoints is %d, minRingSize is %d, maxRingSize is %d", endpoints.Len(), minRingSize, maxRingSize)
 	}
 
 	// https://github.com/envoyproxy/envoy/blob/765c970f06a4c962961a0e03a467e165b276d50f/source/common/upstream/ring_hash_lb.cc#L114
-	normalizedWeights, minWeight := normalizeWeights(subConns)
+	normalizedWeights, minWeight := normalizeWeights(endpoints)
 	if logger.V(2) {
-		logger.Infof("newRing: normalized subConn weights is %v", normalizedWeights)
+		logger.Infof("newRing: normalized endpoint weights is %v", normalizedWeights)
 	}
 
 	// Normalized weights for {3,3,4} is {0.3,0.3,0.4}.
@@ -99,16 +101,16 @@ func newRing(subConns *resolver.AddressMap, minRingSize, maxRingSize uint64, log
 	// A hash is generated for each item, and later the results will be sorted
 	// based on the hash.
 	var currentHashes, targetHashes float64
-	for _, scw := range normalizedWeights {
-		targetHashes += scale * scw.weight
+	for _, epInfo := range normalizedWeights {
+		targetHashes += scale * epInfo.scaledWeight
 		// This index ensures that ring entries corresponding to the same
-		// address hash to different values. And since this index is
-		// per-address, these entries hash to the same value across address
+		// endpoint hash to different values. And since this index is
+		// per-endpoint, these entries hash to the same value across address
 		// updates.
 		idx := 0
 		for currentHashes < targetHashes {
-			h := xxhash.Sum64String(scw.sc.addr + "_" + strconv.Itoa(idx))
-			items = append(items, &ringEntry{hash: h, sc: scw.sc})
+			h := xxhash.Sum64String(epInfo.firstAddr + "_" + strconv.Itoa(idx))
+			items = append(items, &ringEntry{hash: h, firstAddr: epInfo.firstAddr, weight: epInfo.originalWeight})
 			idx++
 			currentHashes++
 		}
@@ -122,45 +124,51 @@ func newRing(subConns *resolver.AddressMap, minRingSize, maxRingSize uint64, log
 	return &ring{items: items}
 }
 
-// normalizeWeights calculates the normalized weights for each subConn in the
-// given subConns map. It returns a slice of subConnWithWeight structs, where
-// each struct contains a subConn and its corresponding weight. The function
-// also returns the minimum weight among all subConns.
+// normalizeWeights calculates the normalized weights for each endpoint in the
+// given endpoints map. It returns a slice of endpointWithState structs, where
+// each struct contains the picker for an endpoint and its corresponding weight.
+// The function also returns the minimum weight among all endpoints.
 //
-// The normalized weight of each subConn is calculated by dividing its weight
-// attribute by the sum of all subConn weights. If the weight attribute is not
-// found on the address, a default weight of 1 is used.
+// The normalized weight of each endpoint is calculated by dividing its weight
+// attribute by the sum of all endpoint weights. If the weight attribute is not
+// found on the endpoint, a default weight of 1 is used.
 //
-// The addresses are sorted in ascending order to ensure consistent results.
+// The endpoints are sorted in ascending order to ensure consistent results.
 //
-// Must be called with a non-empty subConns map.
-func normalizeWeights(subConns *resolver.AddressMap) ([]subConnWithWeight, float64) {
+// Must be called with a non-empty endpoints map.
+func normalizeWeights(endpoints *resolver.EndpointMap) ([]endpointInfo, float64) {
 	var weightSum uint32
-	// Since attributes are explicitly ignored in the AddressMap key, we need to
-	// iterate over the values to get the weights.
-	scVals := subConns.Values()
-	for _, a := range scVals {
-		weightSum += a.(*subConn).weight
+	// Since attributes are explicitly ignored in the EndpointMap key, we need
+	// to iterate over the values to get the weights.
+	endpointVals := endpoints.Values()
+	for _, a := range endpointVals {
+		weightSum += a.(*endpointState).weight
 	}
-	ret := make([]subConnWithWeight, 0, subConns.Len())
+	ret := make([]endpointInfo, 0, endpoints.Len())
 	min := 1.0
-	for _, a := range scVals {
-		scInfo := a.(*subConn)
-		// (*subConn).weight is set to 1 if the weight attribute is not found on
-		// the address. And since this function is guaranteed to be called with
-		// a non-empty subConns map, weightSum is guaranteed to be non-zero. So,
-		// we need not worry about divide by zero error here.
-		nw := float64(scInfo.weight) / float64(weightSum)
-		ret = append(ret, subConnWithWeight{sc: scInfo, weight: nw})
+	for _, a := range endpointVals {
+		epState := a.(*endpointState)
+		// (*endpointState).weight is set to 1 if the weight attribute is not
+		// found on the endpoint. And since this function is guaranteed to be
+		// called with a non-empty endpoints map, weightSum is guaranteed to be
+		// non-zero. So, we need not worry about divide by zero error here.
+		nw := float64(epState.weight) / float64(weightSum)
+		ret = append(ret, endpointInfo{
+			firstAddr:      epState.firstAddr,
+			scaledWeight:   nw,
+			originalWeight: epState.weight,
+		})
 		min = math.Min(min, nw)
 	}
-	// Sort the addresses to return consistent results.
+	// Sort the endpoints to return consistent results.
 	//
 	// Note: this might not be necessary, but this makes sure the ring is
-	// consistent as long as the addresses are the same, for example, in cases
-	// where an address is added and then removed, the RPCs will still pick the
-	// same old SubConn.
-	sort.Slice(ret, func(i, j int) bool { return ret[i].sc.addr < ret[j].sc.addr })
+	// consistent as long as the endpoints are the same, for example, in cases
+	// where an endpoint is added and then removed, the RPCs will still pick the
+	// same old endpoint.
+	sort.Slice(ret, func(i, j int) bool {
+		return ret[i].firstAddr < ret[j].firstAddr
+	})
 	return ret, min
 }
 
diff --git a/xds/internal/balancer/ringhash/ringhash.go b/xds/internal/balancer/ringhash/ringhash.go
index 8c44f19c..a5597c70 100644
--- a/xds/internal/balancer/ringhash/ringhash.go
+++ b/xds/internal/balancer/ringhash/ringhash.go
@@ -27,6 +27,9 @@ import (
 
 	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/balancer/base"
+	"google.golang.org/grpc/balancer/endpointsharding"
+	"google.golang.org/grpc/balancer/lazy"
+	"google.golang.org/grpc/balancer/pickfirst/pickfirstleaf"
 	"google.golang.org/grpc/balancer/weightedroundrobin"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/internal/grpclog"
@@ -38,20 +41,23 @@ import (
 // Name is the name of the ring_hash balancer.
 const Name = "ring_hash_experimental"
 
+func lazyPickFirstBuilder(cc balancer.ClientConn, opts balancer.BuildOptions) balancer.Balancer {
+	return lazy.NewBalancer(cc, opts, balancer.Get(pickfirstleaf.Name).Build)
+}
+
 func init() {
 	balancer.Register(bb{})
 }
 
 type bb struct{}
 
-func (bb) Build(cc balancer.ClientConn, _ balancer.BuildOptions) balancer.Balancer {
+func (bb) Build(cc balancer.ClientConn, opts balancer.BuildOptions) balancer.Balancer {
 	b := &ringhashBalancer{
-		cc:              cc,
-		subConns:        resolver.NewAddressMap(),
-		scStates:        make(map[balancer.SubConn]*subConn),
-		csEvltr:         &connectivityStateEvaluator{},
-		orderedSubConns: make([]*subConn, 0),
+		ClientConn:     cc,
+		endpointStates: resolver.NewEndpointMap(),
 	}
+	esOpts := endpointsharding.Options{DisableAutoReconnect: true}
+	b.child = endpointsharding.NewBalancer(b, opts, lazyPickFirstBuilder, esOpts)
 	b.logger = prefixLogger(b)
 	b.logger.Infof("Created")
 	return b
@@ -65,450 +71,272 @@ func (bb) ParseConfig(c json.RawMessage) (serviceconfig.LoadBalancingConfig, err
 	return parseConfig(c)
 }
 
-type subConn struct {
-	addr   string
-	weight uint32
-	sc     balancer.SubConn
-	logger *grpclog.PrefixLogger
-
-	mu sync.RWMutex
-	// This is the actual state of this SubConn (as updated by the ClientConn).
-	// The effective state can be different, see comment of attemptedToConnect.
-	state connectivity.State
-	// failing is whether this SubConn is in a failing state. A subConn is
-	// considered to be in a failing state if it was previously in
-	// TransientFailure.
-	//
-	// This affects the effective connectivity state of this SubConn, e.g.
-	// - if the actual state is Idle or Connecting, but this SubConn is failing,
-	// the effective state is TransientFailure.
-	//
-	// This is used in pick(). E.g. if a subConn is Idle, but has failing as
-	// true, pick() will
-	// - consider this SubConn as TransientFailure, and check the state of the
-	// next SubConn.
-	// - trigger Connect() (note that normally a SubConn in real
-	// TransientFailure cannot Connect())
-	//
-	// A subConn starts in non-failing (failing is false). A transition to
-	// TransientFailure sets failing to true (and it stays true). A transition
-	// to Ready sets failing to false.
-	failing bool
-	// connectQueued is true if a Connect() was queued for this SubConn while
-	// it's not in Idle (most likely was in TransientFailure). A Connect() will
-	// be triggered on this SubConn when it turns Idle.
-	//
-	// When connectivity state is updated to Idle for this SubConn, if
-	// connectQueued is true, Connect() will be called on the SubConn.
-	connectQueued bool
-	// attemptingToConnect indicates if this subconn is attempting to connect.
-	// It's set when queueConnect is called. It's unset when the state is
-	// changed to Ready/Shutdown, or Idle (and if connectQueued is false).
-	attemptingToConnect bool
-}
-
-// setState updates the state of this SubConn.
-//
-// It also handles the queued Connect(). If the new state is Idle, and a
-// Connect() was queued, this SubConn will be triggered to Connect().
-func (sc *subConn) setState(s connectivity.State) {
-	sc.mu.Lock()
-	defer sc.mu.Unlock()
-	switch s {
-	case connectivity.Idle:
-		// Trigger Connect() if new state is Idle, and there is a queued connect.
-		if sc.connectQueued {
-			sc.connectQueued = false
-			sc.logger.Infof("Executing a queued connect for subConn moving to state: %v", sc.state)
-			sc.sc.Connect()
-		} else {
-			sc.attemptingToConnect = false
-		}
-	case connectivity.Connecting:
-		// Clear connectQueued if the SubConn isn't failing. This state
-		// transition is unlikely to happen, but handle this just in case.
-		sc.connectQueued = false
-	case connectivity.Ready:
-		// Clear connectQueued if the SubConn isn't failing. This state
-		// transition is unlikely to happen, but handle this just in case.
-		sc.connectQueued = false
-		sc.attemptingToConnect = false
-		// Set to a non-failing state.
-		sc.failing = false
-	case connectivity.TransientFailure:
-		// Set to a failing state.
-		sc.failing = true
-	case connectivity.Shutdown:
-		sc.attemptingToConnect = false
-	}
-	sc.state = s
-}
-
-// effectiveState returns the effective state of this SubConn. It can be
-// different from the actual state, e.g. Idle while the subConn is failing is
-// considered TransientFailure. Read comment of field failing for other cases.
-func (sc *subConn) effectiveState() connectivity.State {
-	sc.mu.RLock()
-	defer sc.mu.RUnlock()
-	if sc.failing && (sc.state == connectivity.Idle || sc.state == connectivity.Connecting) {
-		return connectivity.TransientFailure
-	}
-	return sc.state
-}
-
-// queueConnect sets a boolean so that when the SubConn state changes to Idle,
-// it's Connect() will be triggered. If the SubConn state is already Idle, it
-// will just call Connect().
-func (sc *subConn) queueConnect() {
-	sc.mu.Lock()
-	defer sc.mu.Unlock()
-	sc.attemptingToConnect = true
-	if sc.state == connectivity.Idle {
-		sc.logger.Infof("Executing a queued connect for subConn in state: %v", sc.state)
-		sc.sc.Connect()
-		return
-	}
-	// Queue this connect, and when this SubConn switches back to Idle (happens
-	// after backoff in TransientFailure), it will Connect().
-	sc.logger.Infof("Queueing a connect for subConn in state: %v", sc.state)
-	sc.connectQueued = true
-}
-
-func (sc *subConn) isAttemptingToConnect() bool {
-	sc.mu.Lock()
-	defer sc.mu.Unlock()
-	return sc.attemptingToConnect
-}
-
 type ringhashBalancer struct {
-	cc     balancer.ClientConn
-	logger *grpclog.PrefixLogger
+	// The following fields are initialized at build time and read-only after
+	// that and therefore do not need to be guarded by a mutex.
 
-	config   *LBConfig
-	subConns *resolver.AddressMap // Map from resolver.Address to `*subConn`.
-	scStates map[balancer.SubConn]*subConn
-
-	// ring is always in sync with subConns. When subConns change, a new ring is
-	// generated. Note that address weights updates (they are keys in the
-	// subConns map) also regenerates the ring.
-	ring    *ring
-	picker  balancer.Picker
-	csEvltr *connectivityStateEvaluator
-	state   connectivity.State
-
-	resolverErr error // the last error reported by the resolver; cleared on successful resolution
-	connErr     error // the last connection error; cleared upon leaving TransientFailure
-
-	// orderedSubConns contains the list of subconns in the order that addresses
-	// appear from the resolver. Together with lastInternallyTriggeredSCIndex,
-	// this allows triggering connection attempts to all SubConns independently
-	// of the order they appear on the ring. Always in sync with ring and
-	// subConns. The index is reset when addresses change.
-	orderedSubConns                []*subConn
-	lastInternallyTriggeredSCIndex int
+	// ClientConn is embedded to intercept UpdateState calls from the child
+	// endpointsharding balancer.
+	balancer.ClientConn
+	logger *grpclog.PrefixLogger
+	child  balancer.Balancer
+
+	mu                   sync.Mutex
+	config               *LBConfig
+	inhibitChildUpdates  bool
+	shouldRegenerateRing bool
+	endpointStates       *resolver.EndpointMap // Map from endpoint -> *endpointState
+
+	// ring is always in sync with endpoints. When endpoints change, a new ring
+	// is generated. Note that address weights updates also regenerates the
+	// ring.
+	ring *ring
 }
 
-// updateAddresses creates new SubConns and removes SubConns, based on the
-// address update.
-//
-// The return value is whether the new address list is different from the
-// previous. True if
-// - an address was added
-// - an address was removed
-// - an address's weight was updated
-//
-// Note that this function doesn't trigger SubConn connecting, so all the new
-// SubConn states are Idle.
-func (b *ringhashBalancer) updateAddresses(addrs []resolver.Address) bool {
-	var addrsUpdated bool
-	// addrsSet is the set converted from addrs, used for quick lookup.
-	addrsSet := resolver.NewAddressMap()
-
-	b.orderedSubConns = b.orderedSubConns[:0] // reuse the underlying array.
-
-	for _, addr := range addrs {
-		addrsSet.Set(addr, true)
-		newWeight := getWeightAttribute(addr)
-		if val, ok := b.subConns.Get(addr); !ok {
-			var sc balancer.SubConn
-			opts := balancer.NewSubConnOptions{
-				HealthCheckEnabled: true,
-				StateListener:      func(state balancer.SubConnState) { b.updateSubConnState(sc, state) },
-			}
-			sc, err := b.cc.NewSubConn([]resolver.Address{addr}, opts)
-			if err != nil {
-				b.logger.Warningf("Failed to create new SubConn: %v", err)
-				continue
+// UpdateState intercepts child balancer state updates. It updates the
+// per-endpoint state stored in the ring, and also the aggregated state based on
+// the child picker. It also reconciles the endpoint list. It sets
+// `b.shouldRegenerateRing` to true if the new endpoint list is different from
+// the previous, i.e. any of the following is true:
+// - an endpoint was added
+// - an endpoint was removed
+// - an endpoint's weight was updated
+// - the first addresses of the endpoint has changed
+func (b *ringhashBalancer) UpdateState(state balancer.State) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	childStates := endpointsharding.ChildStatesFromPicker(state.Picker)
+	// endpointsSet is the set converted from endpoints, used for quick lookup.
+	endpointsSet := resolver.NewEndpointMap()
+
+	for _, childState := range childStates {
+		endpoint := childState.Endpoint
+		endpointsSet.Set(endpoint, true)
+		newWeight := getWeightAttribute(endpoint)
+		if val, ok := b.endpointStates.Get(endpoint); !ok {
+			es := &endpointState{
+				balancer:  childState.Balancer,
+				weight:    newWeight,
+				firstAddr: endpoint.Addresses[0].Addr,
+				state:     childState.State,
 			}
-			scs := &subConn{addr: addr.Addr, weight: newWeight, sc: sc}
-			scs.logger = subConnPrefixLogger(b, scs)
-			scs.setState(connectivity.Idle)
-			b.state = b.csEvltr.recordTransition(connectivity.Shutdown, connectivity.Idle)
-			b.subConns.Set(addr, scs)
-			b.scStates[sc] = scs
-			b.orderedSubConns = append(b.orderedSubConns, scs)
-			addrsUpdated = true
+			b.endpointStates.Set(endpoint, es)
+			b.shouldRegenerateRing = true
 		} else {
-			// We have seen this address before and created a subConn for it. If the
-			// weight associated with the address has changed, update the subConns map
-			// with the new weight. This will be used when a new ring is created.
-			//
-			// There is no need to call UpdateAddresses on the subConn at this point
-			// since *only* the weight attribute has changed, and that does not affect
-			// subConn uniqueness.
-			scInfo := val.(*subConn)
-			b.orderedSubConns = append(b.orderedSubConns, scInfo)
-			if oldWeight := scInfo.weight; oldWeight != newWeight {
-				scInfo.weight = newWeight
-				b.subConns.Set(addr, scInfo)
-				// Return true to force recreation of the ring.
-				addrsUpdated = true
+			// We have seen this endpoint before and created a `endpointState`
+			// object for it. If the weight or the first address of the endpoint
+			// has changed, update the endpoint state map with the new weight.
+			// This will be used when a new ring is created.
+			es := val.(*endpointState)
+			if oldWeight := es.weight; oldWeight != newWeight {
+				b.shouldRegenerateRing = true
+				es.weight = newWeight
+			}
+			if es.firstAddr != endpoint.Addresses[0].Addr {
+				// If the order of the addresses for a given endpoint change,
+				// that will change the position of the endpoint in the ring.
+				// -A61
+				b.shouldRegenerateRing = true
+				es.firstAddr = endpoint.Addresses[0].Addr
 			}
+			es.state = childState.State
 		}
 	}
-	for _, addr := range b.subConns.Keys() {
-		// addr was removed by resolver.
-		if _, ok := addrsSet.Get(addr); !ok {
-			v, _ := b.subConns.Get(addr)
-			scInfo := v.(*subConn)
-			scInfo.sc.Shutdown()
-			b.subConns.Delete(addr)
-			addrsUpdated = true
-			// Keep the state of this sc in b.scStates until sc's state becomes Shutdown.
-			// The entry will be deleted in updateSubConnState.
+
+	for _, endpoint := range b.endpointStates.Keys() {
+		if _, ok := endpointsSet.Get(endpoint); ok {
+			continue
 		}
+		// endpoint was removed by resolver.
+		b.endpointStates.Delete(endpoint)
+		b.shouldRegenerateRing = true
 	}
-	if addrsUpdated {
-		b.lastInternallyTriggeredSCIndex = 0
-	}
-	return addrsUpdated
+
+	b.updatePickerLocked()
 }
 
-func (b *ringhashBalancer) UpdateClientConnState(s balancer.ClientConnState) error {
+func (b *ringhashBalancer) UpdateClientConnState(ccs balancer.ClientConnState) error {
 	if b.logger.V(2) {
-		b.logger.Infof("Received update from resolver, balancer config: %+v", pretty.ToJSON(s.BalancerConfig))
+		b.logger.Infof("Received update from resolver, balancer config: %+v", pretty.ToJSON(ccs.BalancerConfig))
 	}
 
-	newConfig, ok := s.BalancerConfig.(*LBConfig)
+	newConfig, ok := ccs.BalancerConfig.(*LBConfig)
 	if !ok {
-		return fmt.Errorf("unexpected balancer config with type: %T", s.BalancerConfig)
+		return fmt.Errorf("unexpected balancer config with type: %T", ccs.BalancerConfig)
 	}
 
-	// If addresses were updated, whether it resulted in SubConn
-	// creation/deletion, or just weight update, we need to regenerate the ring
-	// and send a new picker.
-	regenerateRing := b.updateAddresses(s.ResolverState.Addresses)
+	b.mu.Lock()
+	b.inhibitChildUpdates = true
+	b.mu.Unlock()
+
+	defer func() {
+		b.mu.Lock()
+		b.inhibitChildUpdates = false
+		b.updatePickerLocked()
+		b.mu.Unlock()
+	}()
+
+	if err := b.child.UpdateClientConnState(balancer.ClientConnState{
+		// Make pickfirst children use health listeners for outlier detection
+		// and health checking to work.
+		ResolverState: pickfirstleaf.EnableHealthListener(ccs.ResolverState),
+	}); err != nil {
+		return err
+	}
 
-	// If the ring configuration has changed, we need to regenerate the ring and
-	// send a new picker.
+	b.mu.Lock()
+	// Ring updates can happen due to the following:
+	// 1. Addition or deletion of endpoints: The synchronous picker update from
+	//    the child endpointsharding balancer would contain the list of updated
+	//    endpoints.  Updates triggered by the child after handling the
+	//    `UpdateClientConnState` call will not change the endpoint list.
+	// 2. Change in the `LoadBalancerConfig`: Ring config such as max/min ring
+	//    size.
+	// To avoid extra ring updates, a boolean is used to track the need for a
+	// ring update and the update is done only once at the end.
+	//
+	// If the ring configuration has changed, we need to regenerate the ring
+	// while sending a new picker.
 	if b.config == nil || b.config.MinRingSize != newConfig.MinRingSize || b.config.MaxRingSize != newConfig.MaxRingSize {
-		regenerateRing = true
+		b.shouldRegenerateRing = true
 	}
 	b.config = newConfig
-
-	// If resolver state contains no addresses, return an error so ClientConn
-	// will trigger re-resolve. Also records this as an resolver error, so when
-	// the overall state turns transient failure, the error message will have
-	// the zero address information.
-	if len(s.ResolverState.Addresses) == 0 {
-		b.ResolverError(errors.New("produced zero addresses"))
-		return balancer.ErrBadResolverState
-	}
-
-	if regenerateRing {
-		// Ring creation is guaranteed to not fail because we call newRing()
-		// with a non-empty subConns map.
-		b.ring = newRing(b.subConns, b.config.MinRingSize, b.config.MaxRingSize, b.logger)
-		b.regeneratePicker()
-		b.cc.UpdateState(balancer.State{ConnectivityState: b.state, Picker: b.picker})
-	}
-
-	// Successful resolution; clear resolver error and return nil.
-	b.resolverErr = nil
+	b.mu.Unlock()
 	return nil
 }
 
 func (b *ringhashBalancer) ResolverError(err error) {
-	b.resolverErr = err
-	if b.subConns.Len() == 0 {
-		b.state = connectivity.TransientFailure
-	}
-
-	if b.state != connectivity.TransientFailure {
-		// The picker will not change since the balancer does not currently
-		// report an error.
-		return
-	}
-	b.regeneratePicker()
-	b.cc.UpdateState(balancer.State{
-		ConnectivityState: b.state,
-		Picker:            b.picker,
-	})
+	b.child.ResolverError(err)
 }
 
 func (b *ringhashBalancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
 	b.logger.Errorf("UpdateSubConnState(%v, %+v) called unexpectedly", sc, state)
 }
 
-// updateSubConnState updates the per-SubConn state stored in the ring, and also
-// the aggregated state.
-//
-//	It triggers an update to cc when:
-//	- the new state is TransientFailure, to update the error message
-//	  - it's possible that this is a noop, but sending an extra update is easier
-//	    than comparing errors
-//
-//	- the aggregated state is changed
-//	  - the same picker will be sent again, but this update may trigger a re-pick
-//	    for some RPCs.
-func (b *ringhashBalancer) updateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
-	s := state.ConnectivityState
-	if logger.V(2) {
-		b.logger.Infof("Handle SubConn state change: %p, %v", sc, s)
-	}
-	scs, ok := b.scStates[sc]
-	if !ok {
-		b.logger.Infof("Received state change for an unknown SubConn: %p, %v", sc, s)
-		return
-	}
-	oldSCState := scs.effectiveState()
-	scs.setState(s)
-	newSCState := scs.effectiveState()
-	b.logger.Infof("SubConn's effective old state was: %v, new state is %v", oldSCState, newSCState)
-
-	b.state = b.csEvltr.recordTransition(oldSCState, newSCState)
-
-	switch s {
-	case connectivity.TransientFailure:
-		// Save error to be reported via picker.
-		b.connErr = state.ConnectionError
-	case connectivity.Shutdown:
-		// When an address was removed by resolver, b called Shutdown but kept
-		// the sc's state in scStates. Remove state for this sc here.
-		delete(b.scStates, sc)
-	}
-
-	if oldSCState != newSCState {
-		// Because the picker caches the state of the subconns, we always
-		// regenerate and update the picker when the effective SubConn state
-		// changes.
-		b.regeneratePicker()
-		b.logger.Infof("Pushing new state %v and picker %p", b.state, b.picker)
-		b.cc.UpdateState(balancer.State{ConnectivityState: b.state, Picker: b.picker})
-	}
-
-	switch b.state {
-	case connectivity.Connecting, connectivity.TransientFailure:
+func (b *ringhashBalancer) updatePickerLocked() {
+	state := b.aggregatedStateLocked()
+	// Start connecting to new endpoints if necessary.
+	if state == connectivity.Connecting || state == connectivity.TransientFailure {
 		// When overall state is TransientFailure, we need to make sure at least
-		// one SubConn is attempting to connect, otherwise this balancer may
+		// one endpoint is attempting to connect, otherwise this balancer may
 		// never get picks if the parent is priority.
 		//
 		// Because we report Connecting as the overall state when only one
-		// SubConn is in TransientFailure, we do the same check for Connecting
+		// endpoint is in TransientFailure, we do the same check for Connecting
 		// here.
 		//
-		// Note that this check also covers deleting SubConns due to address
-		// change. E.g. if the SubConn attempting to connect is deleted, and the
-		// overall state is TF. Since there must be at least one SubConn
-		// attempting to connect, we need to trigger one. But since the deleted
-		// SubConn will eventually send a shutdown update, this code will run
-		// and trigger the next SubConn to connect.
-		for _, v := range b.subConns.Values() {
-			sc := v.(*subConn)
-			if sc.isAttemptingToConnect() {
-				return
+		// Note that this check also covers deleting endpoints. E.g. if the
+		// endpoint attempting to connect is deleted, and the overall state is
+		// TF. Since there must be at least one endpoint attempting to connect,
+		// we need to trigger one.
+		//
+		// TODO: https://github.com/grpc/grpc-go/issues/8085 - Restrict the
+		// condition under which an endpoint is connected. The pseudocode
+		// mentioned in A61 doesn't handle the following edge cases where the
+		// aggregated state is TF, but no endpoint actually enters TF:
+		// 1. There are four endpoints in the following states: TF, TF, READY,
+		//    and IDLE. If the READY endpoint fails, it transitions to IDLE,
+		//    resulting in the new states: TF, TF, IDLE, IDLE.
+		// 2. There are four endpoints in the following states: TF, TF,
+		//    CONNECTING, and IDLE. If the CONNECTING endpoint is removed, the
+		//    new states become: TF, TF, IDLE.
+		var idleBalancer balancer.ExitIdler
+		for _, val := range b.endpointStates.Values() {
+			es := val.(*endpointState)
+			connState := es.state.ConnectivityState
+			if connState == connectivity.Connecting {
+				idleBalancer = nil
+				break
+			}
+			if idleBalancer == nil && connState == connectivity.Idle {
+				idleBalancer = es.balancer
 			}
 		}
-
-		// Trigger a SubConn (the next in the order addresses appear in the
-		// resolver) to connect if nobody is attempting to connect.
-		b.lastInternallyTriggeredSCIndex = (b.lastInternallyTriggeredSCIndex + 1) % len(b.orderedSubConns)
-		b.orderedSubConns[b.lastInternallyTriggeredSCIndex].queueConnect()
+		if idleBalancer != nil {
+			idleBalancer.ExitIdle()
+		}
 	}
-}
 
-// mergeErrors builds an error from the last connection error and the last
-// resolver error.  Must only be called if b.state is TransientFailure.
-func (b *ringhashBalancer) mergeErrors() error {
-	// connErr must always be non-nil unless there are no SubConns, in which
-	// case resolverErr must be non-nil.
-	if b.connErr == nil {
-		return fmt.Errorf("last resolver error: %v", b.resolverErr)
-	}
-	if b.resolverErr == nil {
-		return fmt.Errorf("last connection error: %v", b.connErr)
+	if b.inhibitChildUpdates {
+		return
 	}
-	return fmt.Errorf("last connection error: %v; last resolver error: %v", b.connErr, b.resolverErr)
-}
 
-func (b *ringhashBalancer) regeneratePicker() {
-	if b.state == connectivity.TransientFailure {
-		b.picker = base.NewErrPicker(b.mergeErrors())
-		return
+	// Update the channel.
+	if b.endpointStates.Len() > 0 && b.shouldRegenerateRing {
+		// with a non-empty list of endpoints.
+		b.ring = newRing(b.endpointStates, b.config.MinRingSize, b.config.MaxRingSize, b.logger)
 	}
-	b.picker = newPicker(b.ring, b.logger)
+	b.shouldRegenerateRing = false
+	var newPicker balancer.Picker
+	if b.endpointStates.Len() == 0 {
+		newPicker = base.NewErrPicker(errors.New("produced zero addresses"))
+	} else {
+		newPicker = b.newPickerLocked()
+	}
+	b.logger.Infof("Pushing new state %v and picker %p", state, newPicker)
+	b.ClientConn.UpdateState(balancer.State{
+		ConnectivityState: state,
+		Picker:            newPicker,
+	})
 }
 
 func (b *ringhashBalancer) Close() {
 	b.logger.Infof("Shutdown")
+	b.child.Close()
 }
 
 func (b *ringhashBalancer) ExitIdle() {
 	// ExitIdle implementation is a no-op because connections are either
-	// triggers from picks or from subConn state changes.
+	// triggers from picks or from child balancer state changes.
 }
 
-// connectivityStateEvaluator takes the connectivity states of multiple SubConns
-// and returns one aggregated connectivity state.
-//
-// It's not thread safe.
-type connectivityStateEvaluator struct {
-	sum  uint64
-	nums [5]uint64
+// newPickerLocked generates a picker. The picker copies the endpoint states
+// over to avoid locking the mutex at RPC time. The picker should be
+// re-generated every time an endpoint state is updated.
+func (b *ringhashBalancer) newPickerLocked() *picker {
+	states := make(map[string]balancer.State)
+	for _, val := range b.endpointStates.Values() {
+		epState := val.(*endpointState)
+		states[epState.firstAddr] = epState.state
+	}
+	return &picker{ring: b.ring, logger: b.logger, endpointStates: states}
 }
 
-// recordTransition records state change happening in subConn and based on that
-// it evaluates what aggregated state should be.
-//
-// - If there is at least one subchannel in READY state, report READY.
-// - If there are 2 or more subchannels in TRANSIENT_FAILURE state, report TRANSIENT_FAILURE.
-// - If there is at least one subchannel in CONNECTING state, report CONNECTING.
-// - If there is one subchannel in TRANSIENT_FAILURE and there is more than one subchannel, report state CONNECTING.
-// - If there is at least one subchannel in Idle state, report Idle.
-// - Otherwise, report TRANSIENT_FAILURE.
+// aggregatedStateLocked returns the aggregated child balancers state
+// based on the following rules.
+//   - If there is at least one endpoint in READY state, report READY.
+//   - If there are 2 or more endpoints in TRANSIENT_FAILURE state, report
+//     TRANSIENT_FAILURE.
+//   - If there is at least one endpoint in CONNECTING state, report CONNECTING.
+//   - If there is one endpoint in TRANSIENT_FAILURE and there is more than one
+//     endpoint, report state CONNECTING.
+//   - If there is at least one endpoint in Idle state, report Idle.
+//   - Otherwise, report TRANSIENT_FAILURE.
 //
 // Note that if there are 1 connecting, 2 transient failure, the overall state
 // is transient failure. This is because the second transient failure is a
-// fallback of the first failing SubConn, and we want to report transient
+// fallback of the first failing endpoint, and we want to report transient
 // failure to failover to the lower priority.
-func (cse *connectivityStateEvaluator) recordTransition(oldState, newState connectivity.State) connectivity.State {
-	// Update counters.
-	for idx, state := range []connectivity.State{oldState, newState} {
-		updateVal := 2*uint64(idx) - 1 // -1 for oldState and +1 for new.
-		cse.nums[state] += updateVal
-	}
-	if oldState == connectivity.Shutdown {
-		// There's technically no transition from Shutdown. But we record a
-		// Shutdown->Idle transition when a new SubConn is created.
-		cse.sum++
-	}
-	if newState == connectivity.Shutdown {
-		cse.sum--
+func (b *ringhashBalancer) aggregatedStateLocked() connectivity.State {
+	var nums [5]int
+	for _, val := range b.endpointStates.Values() {
+		es := val.(*endpointState)
+		nums[es.state.ConnectivityState]++
 	}
 
-	if cse.nums[connectivity.Ready] > 0 {
+	if nums[connectivity.Ready] > 0 {
 		return connectivity.Ready
 	}
-	if cse.nums[connectivity.TransientFailure] > 1 {
+	if nums[connectivity.TransientFailure] > 1 {
 		return connectivity.TransientFailure
 	}
-	if cse.nums[connectivity.Connecting] > 0 {
+	if nums[connectivity.Connecting] > 0 {
 		return connectivity.Connecting
 	}
-	if cse.nums[connectivity.TransientFailure] > 0 && cse.sum > 1 {
+	if nums[connectivity.TransientFailure] == 1 && b.endpointStates.Len() > 1 {
 		return connectivity.Connecting
 	}
-	if cse.nums[connectivity.Idle] > 0 {
+	if nums[connectivity.Idle] > 0 {
 		return connectivity.Idle
 	}
 	return connectivity.TransientFailure
@@ -521,10 +349,24 @@ func (cse *connectivityStateEvaluator) recordTransition(oldState, newState conne
 // When used in the xDS context, the weight attribute is guaranteed to be
 // non-zero. But, when used in a non-xDS context, the weight attribute could be
 // unset. A Default of 1 is used in the latter case.
-func getWeightAttribute(addr resolver.Address) uint32 {
-	w := weightedroundrobin.GetAddrInfo(addr).Weight
+func getWeightAttribute(e resolver.Endpoint) uint32 {
+	w := weightedroundrobin.AddrInfoFromEndpoint(e).Weight
 	if w == 0 {
 		return 1
 	}
 	return w
 }
+
+type endpointState struct {
+	// firstAddr is the first address in the endpoint. Per gRFC A61, each entry
+	// in the ring is an endpoint, positioned based on the hash of the
+	// endpoint's first address.
+	firstAddr string
+	weight    uint32
+	balancer  balancer.ExitIdler
+
+	// state is updated by the balancer while receiving resolver updates from
+	// the channel and picker updates from its children. Access to it is guarded
+	// by ringhashBalancer.mu.
+	state balancer.State
+}



Test patch:
diff --git a/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go b/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
index 79ef02f0..a98c0c68 100644
--- a/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
+++ b/xds/internal/balancer/ringhash/e2e/ringhash_balancer_test.go
@@ -72,7 +72,8 @@ func Test(t *testing.T) {
 }
 
 const (
-	defaultTestTimeout = 10 * time.Second
+	defaultTestTimeout      = 10 * time.Second
+	defaultTestShortTimeout = 10 * time.Millisecond
 
 	errorTolerance = .05 // For tests that rely on statistical significance.
 
@@ -161,16 +162,25 @@ func (s) TestRingHash_ReconnectToMoveOutOfTransientFailure(t *testing.T) {
 	}
 }
 
-// startTestServiceBackends starts num stub servers. It returns their addresses.
-// Servers are closed when the test is stopped.
-func startTestServiceBackends(t *testing.T, num int) []string {
+// startTestServiceBackends starts num stub servers. It returns the list of
+// stubservers. Servers are closed when the test is stopped.
+func startTestServiceBackends(t *testing.T, num int) []*stubserver.StubServer {
 	t.Helper()
 
-	addrs := make([]string, 0, num)
+	servers := make([]*stubserver.StubServer, 0, num)
 	for i := 0; i < num; i++ {
 		server := stubserver.StartTestService(t, nil)
 		t.Cleanup(server.Stop)
-		addrs = append(addrs, server.Address)
+		servers = append(servers, server)
+	}
+	return servers
+}
+
+// backendAddrs returns a list of address strings for the given stubservers.
+func backendAddrs(servers []*stubserver.StubServer) []string {
+	addrs := make([]string, 0, len(servers))
+	for _, s := range servers {
+		addrs = append(addrs, s.Address)
 	}
 	return addrs
 }
@@ -179,10 +189,25 @@ func startTestServiceBackends(t *testing.T, num int) []string {
 // addresses.
 func backendOptions(t *testing.T, serverAddrs []string) []e2e.BackendOptions {
 	t.Helper()
+	backendAddrs := [][]string{}
+	for _, addr := range serverAddrs {
+		backendAddrs = append(backendAddrs, []string{addr})
+	}
+	return backendOptionsForEndpointsWithMultipleAddrs(t, backendAddrs)
+}
+
+// backendOptions returns a slice of e2e.BackendOptions for the given server
+// addresses. Each endpoint can have multiple addresses.
+func backendOptionsForEndpointsWithMultipleAddrs(t *testing.T, backendAddrs [][]string) []e2e.BackendOptions {
+	t.Helper()
 
 	var backendOpts []e2e.BackendOptions
-	for _, addr := range serverAddrs {
-		backendOpts = append(backendOpts, e2e.BackendOptions{Ports: []uint32{testutils.ParsePort(t, addr)}})
+	for _, backend := range backendAddrs {
+		ports := []uint32{}
+		for _, addr := range backend {
+			ports = append(ports, testutils.ParsePort(t, addr))
+		}
+		backendOpts = append(backendOpts, e2e.BackendOptions{Ports: ports})
 	}
 	return backendOpts
 }
@@ -291,7 +316,7 @@ func xdsUpdateOpts(nodeID string, endpoints *v3endpointpb.ClusterLoadAssignment,
 // the first cluster is in transient failure, all RPCs are sent to the second
 // cluster using the ring hash policy.
 func (s) TestRingHash_AggregateClusterFallBackFromRingHashAtStartup(t *testing.T) {
-	addrs := startTestServiceBackends(t, 2)
+	addrs := backendAddrs(startTestServiceBackends(t, 2))
 
 	const primaryClusterName = "new_cluster_1"
 	const primaryServiceName = "new_eds_service_1"
@@ -396,7 +421,7 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartup(
 	const logicalDNSClusterName = "logical_dns_cluster"
 	const clusterName = "aggregate_cluster"
 
-	backends := startTestServiceBackends(t, 1)
+	backends := backendAddrs(startTestServiceBackends(t, 1))
 
 	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
 		ClusterName: edsClusterName,
@@ -473,7 +498,7 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartupN
 	const logicalDNSClusterName = "logical_dns_cluster"
 	const clusterName = "aggregate_cluster"
 
-	backends := startTestServiceBackends(t, 1)
+	backends := backendAddrs(startTestServiceBackends(t, 1))
 
 	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
 		ClusterName: edsClusterName,
@@ -601,10 +626,21 @@ func (s) TestRingHash_AggregateClusterFallBackFromRingHashToLogicalDnsAtStartupN
 // with the given addresses.
 func endpointResource(t *testing.T, clusterName string, addrs []string) *v3endpointpb.ClusterLoadAssignment {
 	t.Helper()
+	backendAddrs := [][]string{}
+	for _, addr := range addrs {
+		backendAddrs = append(backendAddrs, []string{addr})
+	}
+	return endpointResourceForBackendsWithMultipleAddrs(t, clusterName, backendAddrs)
+}
+
+// endpointResourceForBackendsWithMultipleAddrs creates a ClusterLoadAssignment
+// containing a single locality with the given addresses.
+func endpointResourceForBackendsWithMultipleAddrs(t *testing.T, clusterName string, addrs [][]string) *v3endpointpb.ClusterLoadAssignment {
+	t.Helper()
 
 	// We must set the host name socket address in EDS, as the ring hash policy
 	// uses it to construct the ring.
-	host, _, err := net.SplitHostPort(addrs[0])
+	host, _, err := net.SplitHostPort(addrs[0][0])
 	if err != nil {
 		t.Fatalf("Failed to split host and port from stubserver: %v", err)
 	}
@@ -613,7 +649,7 @@ func endpointResource(t *testing.T, clusterName string, addrs []string) *v3endpo
 		ClusterName: clusterName,
 		Host:        host,
 		Localities: []e2e.LocalityOptions{{
-			Backends: backendOptions(t, addrs),
+			Backends: backendOptionsForEndpointsWithMultipleAddrs(t, addrs),
 			Weight:   1,
 		}},
 	})
@@ -622,7 +658,7 @@ func endpointResource(t *testing.T, clusterName string, addrs []string) *v3endpo
 // Tests that ring hash policy that hashes using channel id ensures all RPCs to
 // go 1 particular backend.
 func (s) TestRingHash_ChannelIdHashing(t *testing.T) {
-	backends := startTestServiceBackends(t, 4)
+	backends := backendAddrs(startTestServiceBackends(t, 4))
 
 	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
 
@@ -682,7 +718,7 @@ func headerHashRoute(routeName, virtualHostName, clusterName, header string) *v3
 // Tests that ring hash policy that hashes using a header value can send RPCs
 // to specific backends based on their hash.
 func (s) TestRingHash_HeaderHashing(t *testing.T) {
-	backends := startTestServiceBackends(t, 4)
+	backends := backendAddrs(startTestServiceBackends(t, 4))
 
 	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
 
@@ -726,7 +762,7 @@ func (s) TestRingHash_HeaderHashing(t *testing.T) {
 // Tests that ring hash policy that hashes using a header value and regex
 // rewrite to aggregate RPCs to 1 backend.
 func (s) TestRingHash_HeaderHashingWithRegexRewrite(t *testing.T) {
-	backends := startTestServiceBackends(t, 4)
+	backends := backendAddrs(startTestServiceBackends(t, 4))
 
 	clusterName := "cluster"
 	endpoints := endpointResource(t, clusterName, backends)
@@ -826,7 +862,7 @@ func setRingHashLBPolicyWithHighMinRingSize(t *testing.T, cluster *v3clusterpb.C
 
 // Tests that ring hash policy that hashes using a random value.
 func (s) TestRingHash_NoHashPolicy(t *testing.T) {
-	backends := startTestServiceBackends(t, 2)
+	backends := backendAddrs(startTestServiceBackends(t, 2))
 	numRPCs := computeIdealNumberOfRPCs(t, .5, errorTolerance)
 
 	const clusterName = "cluster"
@@ -867,7 +903,7 @@ func (s) TestRingHash_NoHashPolicy(t *testing.T) {
 
 // Tests that we observe endpoint weights.
 func (s) TestRingHash_EndpointWeights(t *testing.T) {
-	backends := startTestServiceBackends(t, 3)
+	backends := backendAddrs(startTestServiceBackends(t, 3))
 
 	const clusterName = "cluster"
 	backendOpts := []e2e.BackendOptions{
@@ -933,7 +969,7 @@ func (s) TestRingHash_EndpointWeights(t *testing.T) {
 // Tests that ring hash policy evaluation will continue past the terminal hash
 // policy if no results are produced yet.
 func (s) TestRingHash_ContinuesPastTerminalPolicyThatDoesNotProduceResult(t *testing.T) {
-	backends := startTestServiceBackends(t, 2)
+	backends := backendAddrs(startTestServiceBackends(t, 2))
 
 	const clusterName = "cluster"
 	endpoints := endpointResource(t, clusterName, backends)
@@ -1002,7 +1038,7 @@ func (s) TestRingHash_ContinuesPastTerminalPolicyThatDoesNotProduceResult(t *tes
 // Tests that a random hash is used when header hashing policy specified a
 // header field that the RPC did not have.
 func (s) TestRingHash_HashOnHeaderThatIsNotPresent(t *testing.T) {
-	backends := startTestServiceBackends(t, 2)
+	backends := backendAddrs(startTestServiceBackends(t, 2))
 	wantFractionPerBackend := .5
 	numRPCs := computeIdealNumberOfRPCs(t, wantFractionPerBackend, errorTolerance)
 
@@ -1053,7 +1089,7 @@ func (s) TestRingHash_HashOnHeaderThatIsNotPresent(t *testing.T) {
 // Tests that a random hash is used when only unsupported hash policies are
 // configured.
 func (s) TestRingHash_UnsupportedHashPolicyDefaultToRandomHashing(t *testing.T) {
-	backends := startTestServiceBackends(t, 2)
+	backends := backendAddrs(startTestServiceBackends(t, 2))
 	wantFractionPerBackend := .5
 	numRPCs := computeIdealNumberOfRPCs(t, wantFractionPerBackend, errorTolerance)
 
@@ -1120,7 +1156,7 @@ func (s) TestRingHash_UnsupportedHashPolicyDefaultToRandomHashing(t *testing.T)
 // Tests that unsupported hash policy types are all ignored before a supported
 // hash policy.
 func (s) TestRingHash_UnsupportedHashPolicyUntilChannelIdHashing(t *testing.T) {
-	backends := startTestServiceBackends(t, 2)
+	backends := backendAddrs(startTestServiceBackends(t, 2))
 
 	const clusterName = "cluster"
 	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
@@ -1196,7 +1232,7 @@ func (s) TestRingHash_UnsupportedHashPolicyUntilChannelIdHashing(t *testing.T) {
 // Tests that ring hash policy that hashes using a random value can spread RPCs
 // across all the backends according to locality weight.
 func (s) TestRingHash_RandomHashingDistributionAccordingToLocalityAndEndpointWeight(t *testing.T) {
-	backends := startTestServiceBackends(t, 2)
+	backends := backendAddrs(startTestServiceBackends(t, 2))
 
 	const clusterName = "cluster"
 	const locality1Weight = uint32(1)
@@ -1267,7 +1303,7 @@ func (s) TestRingHash_RandomHashingDistributionAccordingToLocalityAndEndpointWei
 // to go 1 particular backend; and that subsequent hashing policies are ignored
 // due to the setting of terminal.
 func (s) TestRingHash_FixedHashingTerminalPolicy(t *testing.T) {
-	backends := startTestServiceBackends(t, 2)
+	backends := backendAddrs(startTestServiceBackends(t, 2))
 	const clusterName = "cluster"
 	endpoints := endpointResource(t, clusterName, backends)
 	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
@@ -1344,7 +1380,7 @@ func (s) TestRingHash_FixedHashingTerminalPolicy(t *testing.T) {
 // before moving to ready via the public API).
 // TODO: we should be able to catch all state transitions by using the internal.SubscribeToConnectivityStateChanges API.
 func (s) TestRingHash_IdleToReady(t *testing.T) {
-	backends := startTestServiceBackends(t, 1)
+	backends := backendAddrs(startTestServiceBackends(t, 1))
 
 	const clusterName = "cluster"
 	endpoints := endpointResource(t, clusterName, backends)
@@ -1452,7 +1488,7 @@ func (s) TestRingHash_ContinuesConnectingWithoutPicks(t *testing.T) {
 // Tests that when the first pick is down leading to a transient failure, we
 // will move on to the next ring hash entry.
 func (s) TestRingHash_TransientFailureCheckNextOne(t *testing.T) {
-	backends := startTestServiceBackends(t, 1)
+	backends := backendAddrs(startTestServiceBackends(t, 1))
 	unReachableBackends := makeUnreachableBackends(t, 1)
 
 	const clusterName = "cluster"
@@ -1535,7 +1571,7 @@ func (s) TestRingHash_ReattemptWhenGoingFromTransientFailureToIdle(t *testing.T)
 	testutils.AwaitState(ctx, t, conn, connectivity.TransientFailure)
 
 	t.Log("Updating EDS with a new backend endpoint.")
-	backends := startTestServiceBackends(t, 1)
+	backends := backendAddrs(startTestServiceBackends(t, 1))
 	endpoints = e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
 		ClusterName: clusterName,
 		Localities: []e2e.LocalityOptions{{
@@ -1730,7 +1766,7 @@ func (s) TestRingHash_ReattemptWhenAllEndpointsUnreachable(t *testing.T) {
 	t.Log("Restarting the backend server")
 	restartableListener.Restart()
 
-	// Wait for channel to become connected without any pending RPC.
+	// Wait for channel to become READY without any pending RPC.
 	testutils.AwaitState(ctx, t, conn, connectivity.Ready)
 }
 
@@ -1751,7 +1787,7 @@ func (s) TestRingHash_SwitchToLowerPriorityAndThenBack(t *testing.T) {
 	})
 	defer restartableServer.Stop()
 
-	otherBackend := startTestServiceBackends(t, 1)[0]
+	otherBackend := backendAddrs(startTestServiceBackends(t, 1))[0]
 
 	// We must set the host name socket address in EDS, as the ring hash policy
 	// uses it to construct the ring.
@@ -1844,15 +1880,15 @@ func (s) TestRingHash_SwitchToLowerPriorityAndThenBack(t *testing.T) {
 	}
 }
 
-// Tests that when we trigger internal connection attempts without picks, we do
-// so for only one subchannel at a time.
-func (s) TestRingHash_ContinuesConnectingWithoutPicksOneSubchannelAtATime(t *testing.T) {
-	backends := startTestServiceBackends(t, 1)
-	unReachableBackends := makeUnreachableBackends(t, 3)
+// Tests that when we trigger internal connection attempts without picks, we
+// keep retrying all the SubConns that have reported TF previously.
+func (s) TestRingHash_ContinuesConnectingWithoutPicksToMultipleSubConnsConcurrently(t *testing.T) {
+	const backendsCount = 4
+	backends := backendAddrs(startTestServiceBackends(t, backendsCount))
 
 	const clusterName = "cluster"
 
-	endpoints := endpointResource(t, clusterName, append(unReachableBackends, backends...))
+	endpoints := endpointResource(t, clusterName, backends)
 	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
 		ClusterName: clusterName,
 		ServiceName: clusterName,
@@ -1881,17 +1917,20 @@ func (s) TestRingHash_ContinuesConnectingWithoutPicksOneSubchannelAtATime(t *tes
 		t.Fatalf("Failed to create client: %s", err)
 	}
 	defer conn.Close()
-	client := testgrpc.NewTestServiceClient(conn)
 
-	holdNonExistent0 := dialer.Hold(unReachableBackends[0])
-	holdNonExistent1 := dialer.Hold(unReachableBackends[1])
-	holdNonExistent2 := dialer.Hold(unReachableBackends[2])
-	holdGood := dialer.Hold(backends[0])
+	// Create holds for each backend address to delay a successful connection
+	// until the end of the test.
+	holds := make([]*testutils.Hold, backendsCount)
+	for i := 0; i < len(backends); i++ {
+		holds[i] = dialer.Hold(backends[i])
+	}
+
+	client := testgrpc.NewTestServiceClient(conn)
 
 	rpcCtx, rpcCancel := context.WithCancel(ctx)
 	errCh := make(chan error, 1)
 	go func() {
-		rpcCtx = metadata.NewOutgoingContext(rpcCtx, metadata.Pairs("address_hash", unReachableBackends[0]+"_0"))
+		rpcCtx = metadata.NewOutgoingContext(rpcCtx, metadata.Pairs("address_hash", backends[0]+"_0"))
 		_, err := client.EmptyCall(rpcCtx, &testpb.Empty{})
 		if status.Code(err) == codes.Canceled {
 			errCh <- nil
@@ -1902,7 +1941,7 @@ func (s) TestRingHash_ContinuesConnectingWithoutPicksOneSubchannelAtATime(t *tes
 
 	// Wait for the RPC to trigger a connection attempt to the first address,
 	// then cancel the RPC.  No other connection attempts should be started yet.
-	if !holdNonExistent0.Wait(ctx) {
+	if !holds[0].Wait(ctx) {
 		t.Fatalf("Timeout waiting for connection attempt to backend 0")
 	}
 	rpcCancel()
@@ -1910,86 +1949,594 @@ func (s) TestRingHash_ContinuesConnectingWithoutPicksOneSubchannelAtATime(t *tes
 		t.Fatalf("Expected RPC to fail be canceled, got %v", err)
 	}
 
-	// Since the connection attempt to the first address is still blocked, no
-	// other connection attempts should be started yet.
-	if holdNonExistent1.IsStarted() {
-		t.Errorf("Got connection attempt to backend 1, expected no connection attempt.")
+	// In every iteration of the following loop, we count the number of backends
+	// that are dialed. After counting, we fail all the connection attempts.
+	// This should cause the number of dialed backends to increase by 1 in every
+	// iteration of the loop as ringhash tries to exit TRANSIENT_FAILURE.
+	activeAddrs := map[string]bool{}
+	for wantBackendCount := 1; wantBackendCount <= backendsCount; wantBackendCount++ {
+		newAddrIdx := -1
+		for ; ctx.Err() == nil; <-time.After(time.Millisecond) {
+			for i, hold := range holds {
+				if !hold.IsStarted() {
+					continue
+				}
+				if _, ok := activeAddrs[backends[i]]; ok {
+					continue
+				}
+				activeAddrs[backends[i]] = true
+				newAddrIdx = i
+			}
+			if len(activeAddrs) > wantBackendCount {
+				t.Fatalf("More backends dialed than expected: got %d, want %d", len(activeAddrs), wantBackendCount)
+			}
+			if len(activeAddrs) == wantBackendCount {
+				break
+			}
+		}
+
+		// Wait for a short time and verify no more backends are contacted.
+		<-time.After(defaultTestShortTimeout)
+		for i, hold := range holds {
+			if !hold.IsStarted() {
+				continue
+			}
+			activeAddrs[backends[i]] = true
+		}
+		if len(activeAddrs) != wantBackendCount {
+			t.Fatalf("Unexpected number of backends dialed: got %d, want %d", len(activeAddrs), wantBackendCount)
+		}
+
+		// Create a new hold for the address dialed in this iteration and fail
+		// the existing hold.
+		hold := holds[newAddrIdx]
+		holds[newAddrIdx] = dialer.Hold(backends[newAddrIdx])
+		hold.Fail(errors.New("Test error"))
+	}
+
+	// Allow the request to a backend to succeed.
+	if !holds[1].Wait(ctx) {
+		t.Fatalf("Context timed out waiting %q to be dialed again.", backends[1])
+	}
+	holds[1].Resume()
+
+	// Wait for channel to become READY without any pending RPC.
+	testutils.AwaitState(ctx, t, conn, connectivity.Ready)
+}
+
+// Tests that first address of an endpoint is used to generate the ring. The
+// test sends a request to a random endpoint. The test then reverses the
+// addresses of every endpoint and verifies that an RPC with header pointing to
+// the second address of the endpoint is sent to the initial address. The test
+// then swaps the second and third address of the endpoint and verifies that an
+// RPC with the header used earlier still reaches the same backend.
+func (s) TestRingHash_ReorderAddressessWithinEndpoint(t *testing.T) {
+	origDualstackEndpointsEnabled := envconfig.XDSDualstackEndpointsEnabled
+	defer func() {
+		envconfig.XDSDualstackEndpointsEnabled = origDualstackEndpointsEnabled
+	}()
+	envconfig.XDSDualstackEndpointsEnabled = true
+	backends := backendAddrs(startTestServiceBackends(t, 6))
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+
+	const clusterName = "cluster"
+	addrGroups := [][]string{
+		{backends[0], backends[1], backends[2]},
+		{backends[3], backends[4], backends[5]},
+	}
+	endpoints := endpointResourceForBackendsWithMultipleAddrs(t, clusterName, addrGroups)
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+	route := headerHashRoute("new_route", virtualHostName, clusterName, "address_hash")
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	rpcCtx := metadata.NewOutgoingContext(ctx, metadata.Pairs(
+		"address_hash", fmt.Sprintf("%d", rand.Int()),
+	))
+	var remote peer.Peer
+	if _, err := client.EmptyCall(rpcCtx, &testpb.Empty{}, grpc.Peer(&remote)); err != nil {
+		t.Fatalf("rpc EmptyCall() failed: %v", err)
+	}
+
+	initialFirstAddr := ""
+	newFirstAddr := ""
+	switch remote.Addr.String() {
+	case addrGroups[0][0]:
+		initialFirstAddr = addrGroups[0][0]
+		newFirstAddr = addrGroups[0][2]
+	case addrGroups[1][0]:
+		initialFirstAddr = addrGroups[1][0]
+		newFirstAddr = addrGroups[1][2]
+	default:
+		t.Fatalf("Request went to unexpected address: %q", remote.Addr)
+	}
+
+	t.Log("Reversing addresses within each endpoint.")
+	addrGroups1 := [][]string{
+		{addrGroups[0][2], addrGroups[0][1], addrGroups[0][0]},
+		{addrGroups[1][2], addrGroups[1][1], addrGroups[1][0]},
+	}
+	endpoints = endpointResourceForBackendsWithMultipleAddrs(t, clusterName, addrGroups1)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	// The first address of an endpoint is used to create the ring. This means
+	// that requests should continue to go to the first address, but the hash
+	// should be computed based on the last address in the original list.
+	for ; ctx.Err() == nil; <-time.After(time.Millisecond) {
+		rpcCtx := metadata.NewOutgoingContext(ctx, metadata.Pairs(
+			"address_hash", newFirstAddr+"_0",
+		))
+		if _, err := client.EmptyCall(rpcCtx, &testpb.Empty{}, grpc.Peer(&remote)); err != nil {
+			t.Fatalf("rpc EmptyCall() failed: %v", err)
+		}
+		if remote.Addr.String() == initialFirstAddr {
+			break
+		}
+	}
+
+	if ctx.Err() != nil {
+		t.Fatalf("Context timed out waiting for request to be sent to %q, last request went to %q", initialFirstAddr, remote.Addr)
+	}
+
+	t.Log("Swapping the second and third addresses within each endpoint.")
+	// This should not effect the ring, since only the first address is used
+	// by the ring.
+	addrGroups2 := [][]string{
+		{addrGroups1[0][0], addrGroups[0][2], addrGroups[0][1]},
+		{addrGroups1[1][0], addrGroups[1][2], addrGroups[1][1]},
+	}
+	endpoints = endpointResourceForBackendsWithMultipleAddrs(t, clusterName, addrGroups2)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	// Verify that requests with the hash of the last address in chosenAddrGroup
+	// continue reaching the first address in chosenAddrGroup.
+	shortCtx, cancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer cancel()
+	for ; shortCtx.Err() == nil; <-time.After(time.Millisecond) {
+		rpcCtx := metadata.NewOutgoingContext(ctx, metadata.Pairs(
+			"address_hash", newFirstAddr+"_0",
+		))
+		if _, err := client.EmptyCall(rpcCtx, &testpb.Empty{}, grpc.Peer(&remote)); err != nil {
+			t.Fatalf("rpc EmptyCall() failed: %v", err)
+		}
+		if remote.Addr.String() == initialFirstAddr {
+			continue
+		}
+		t.Fatalf("Request went to unexpected backend %q, want backend %q", remote.Addr, initialFirstAddr)
+	}
+}
+
+// Tests that requests are sent to the next address within the same endpoint
+// after the first address becomes unreachable.
+func (s) TestRingHash_FallBackWithinEndpoint(t *testing.T) {
+	origDualstackEndpointsEnabled := envconfig.XDSDualstackEndpointsEnabled
+	defer func() {
+		envconfig.XDSDualstackEndpointsEnabled = origDualstackEndpointsEnabled
+	}()
+	envconfig.XDSDualstackEndpointsEnabled = true
+	backends := startTestServiceBackends(t, 4)
+	backendAddrs := backendAddrs(backends)
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+
+	const clusterName = "cluster"
+	endpoints := endpointResourceForBackendsWithMultipleAddrs(t, clusterName, [][]string{{backendAddrs[0], backendAddrs[1]}, {backendAddrs[2], backendAddrs[3]}})
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+	route := channelIDHashRoute("new_route", virtualHostName, clusterName)
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	conn, err := grpc.NewClient("xds:///test.server", grpc.WithResolvers(xdsResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
+	}
+	defer conn.Close()
+	client := testgrpc.NewTestServiceClient(conn)
+
+	const numRPCs = 5
+	received := checkRPCSendOK(ctx, t, client, numRPCs)
+	if len(received) != 1 {
+		t.Errorf("Got RPCs routed to %v backends, want %v", len(received), 1)
+	}
+	var got int
+	var initialAddr string
+	for initialAddr, got = range received {
+	}
+	if got != numRPCs {
+		t.Errorf("Got %v RPCs routed to a backend, want %v", got, numRPCs)
+	}
+
+	// Due to the channel ID hashing policy, the request could go to the first
+	// address of either endpoint.
+	var backendIdx int
+	switch initialAddr {
+	case backendAddrs[0]:
+		backendIdx = 0
+	case backendAddrs[2]:
+		backendIdx = 2
+	default:
+		t.Fatalf("Request sent to unexpected backend: %q", initialAddr)
+	}
+	otherEndpointAddr := backendAddrs[backendIdx+1]
+
+	// Shut down the previously used backend.
+	backends[backendIdx].Stop()
+	testutils.AwaitState(ctx, t, conn, connectivity.Idle)
+
+	// Verify that the requests go to the remaining address in the same
+	// endpoint.
+	received = checkRPCSendOK(ctx, t, client, numRPCs)
+	if len(received) != 1 {
+		t.Errorf("Got RPCs routed to %v backends, want %v", len(received), 1)
+	}
+	var newAddr string
+	for newAddr, got = range received {
+	}
+	if got != numRPCs {
+		t.Errorf("Got %v RPCs routed to a backend, want %v", got, numRPCs)
+	}
+
+	if newAddr != otherEndpointAddr {
+		t.Errorf("Requests went to unexpected address, got=%q, want=%q", newAddr, otherEndpointAddr)
+	}
+}
+
+// Tests that ringhash is able to recover automatically in situations when a
+// READY endpoint enters IDLE making the aggregated state TRANSIENT_FAILURE. The
+// test creates 4 endpoints in the following connectivity states: [TF, TF,
+// READY, IDLE]. The test fails the READY backend and verifies that the last
+// IDLE endopint is dialed and the channel enters READY.
+func (s) TestRingHash_RecoverWhenEndpointEntersIdle(t *testing.T) {
+	const backendsCount = 4
+	backends := startTestServiceBackends(t, backendsCount)
+	backendAddrs := backendAddrs(backends)
+
+	const clusterName = "cluster"
+
+	endpoints := endpointResource(t, clusterName, backendAddrs)
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+	route := headerHashRoute("new_route", virtualHostName, clusterName, "address_hash")
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
 	}
-	if holdNonExistent2.IsStarted() {
-		t.Errorf("Got connection attempt to backend 2, expected no connection attempt.")
+
+	dialer := testutils.NewBlockingDialer()
+	dialOpts := []grpc.DialOption{
+		grpc.WithResolvers(xdsResolver),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithContextDialer(dialer.DialContext),
+		grpc.WithConnectParams(fastConnectParams),
 	}
-	if holdGood.IsStarted() {
-		t.Errorf("Got connection attempt to good backend, expected no connection attempt.")
+	conn, err := grpc.NewClient("xds:///test.server", dialOpts...)
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
 	}
+	defer conn.Close()
 
-	// Allow the connection attempt to the first address to resume and wait for
-	// the attempt for the second address. No other connection attempts should
-	// be started yet.
-	holdNonExistent0Again := dialer.Hold(unReachableBackends[0])
-	holdNonExistent0.Resume()
-	if !holdNonExistent1.Wait(ctx) {
-		t.Fatalf("Timeout waiting for connection attempt to backend 1")
+	// Create holds for each backend address to delay a successful connection
+	// until the end of the test.
+	holds := make([]*testutils.Hold, backendsCount)
+	for i := 0; i < len(backendAddrs); i++ {
+		holds[i] = dialer.Hold(backendAddrs[i])
 	}
-	if holdNonExistent0Again.IsStarted() {
-		t.Errorf("Got connection attempt to backend 0 again, expected no connection attempt.")
+
+	client := testgrpc.NewTestServiceClient(conn)
+
+	rpcCtx, rpcCancel := context.WithCancel(ctx)
+	errCh := make(chan error, 1)
+	go func() {
+		rpcCtx = metadata.NewOutgoingContext(rpcCtx, metadata.Pairs("address_hash", backendAddrs[0]+"_0"))
+		_, err := client.EmptyCall(rpcCtx, &testpb.Empty{})
+		if status.Code(err) == codes.Canceled {
+			errCh <- nil
+			return
+		}
+		errCh <- err
+	}()
+
+	// Wait for the RPC to trigger a connection attempt to the first address,
+	// then cancel the RPC.  No other connection attempts should be started yet.
+	if !holds[0].Wait(ctx) {
+		t.Fatalf("Timeout waiting for connection attempt to backend 0")
 	}
-	if holdNonExistent2.IsStarted() {
-		t.Errorf("Got connection attempt to backend 2, expected no connection attempt.")
+	rpcCancel()
+	if err := <-errCh; err != nil {
+		t.Fatalf("Expected RPC to fail be canceled, got %v", err)
 	}
-	if holdGood.IsStarted() {
-		t.Errorf("Got connection attempt to good backend, expected no connection attempt.")
+
+	// The number of dialed backends increases by 1 in every iteration of the
+	// loop as ringhash tries to exit TRANSIENT_FAILURE. Run the loop twice to
+	// get two endpoints in TRANSIENT_FAILURE.
+	activeAddrs := map[string]bool{}
+	for wantFailingBackendCount := 1; wantFailingBackendCount <= 2; wantFailingBackendCount++ {
+		newAddrIdx := -1
+		for ; ctx.Err() == nil && len(activeAddrs) < wantFailingBackendCount; <-time.After(time.Millisecond) {
+			for i, hold := range holds {
+				if !hold.IsStarted() {
+					continue
+				}
+				if _, ok := activeAddrs[backendAddrs[i]]; ok {
+					continue
+				}
+				activeAddrs[backendAddrs[i]] = true
+				newAddrIdx = i
+			}
+		}
+
+		if ctx.Err() != nil {
+			t.Fatal("Context timed out waiting for new backneds to be dialed.")
+		}
+		if len(activeAddrs) > wantFailingBackendCount {
+			t.Fatalf("More backends dialed than expected: got %d, want %d", len(activeAddrs), wantFailingBackendCount)
+		}
+
+		// Create a new hold for the address dialed in this iteration and fail
+		// the existing hold.
+		hold := holds[newAddrIdx]
+		holds[newAddrIdx] = dialer.Hold(backendAddrs[newAddrIdx])
+		hold.Fail(errors.New("Test error"))
 	}
 
-	// Allow the connection attempt to the second address to resume and wait for
-	// the attempt for the third address.  No other connection attempts should
-	// be started yet.
-	holdNonExistent1Again := dialer.Hold(unReachableBackends[1])
-	holdNonExistent1.Resume()
-	if !holdNonExistent2.Wait(ctx) {
-		t.Fatalf("Timeout waiting for connection attempt to backend 2")
+	// Current state of endpoints: [TF, TF, READY, IDLE].
+	// Two endpoints failing should cause the channel to enter
+	// TRANSIENT_FAILURE.
+	testutils.AwaitState(ctx, t, conn, connectivity.TransientFailure)
+
+	// Allow the request to the backend dialed next to succeed.
+	readyBackendIdx := -1
+	for ; ctx.Err() == nil && readyBackendIdx == -1; <-time.After(time.Millisecond) {
+		for i, addr := range backendAddrs {
+			if _, ok := activeAddrs[addr]; ok || !holds[i].IsStarted() {
+				continue
+			}
+			readyBackendIdx = i
+			activeAddrs[addr] = true
+			holds[i].Resume()
+			break
+		}
 	}
-	if holdNonExistent0Again.IsStarted() {
-		t.Errorf("Got connection attempt to backend 0 again, expected no connection attempt.")
+
+	if ctx.Err() != nil {
+		t.Fatal("Context timed out waiting for the next backend to be contacted.")
 	}
-	if holdNonExistent1Again.IsStarted() {
-		t.Errorf("Got connection attempt to backend 1 again, expected no connection attempt.")
+
+	// Wait for channel to become READY without any pending RPC.
+	testutils.AwaitState(ctx, t, conn, connectivity.Ready)
+
+	// Current state of endpoints: [TF, TF, READY, IDLE].
+	// Stopping the READY backend should cause the channel to re-enter
+	// TRANSIENT_FAILURE.
+	backends[readyBackendIdx].Stop()
+	testutils.AwaitState(ctx, t, conn, connectivity.TransientFailure)
+
+	// To recover from TRANSIENT_FAILURE, ringhash should automatically try to
+	// connect to the final endpoint.
+	readyBackendIdx = -1
+	for ; ctx.Err() == nil && readyBackendIdx == -1; <-time.After(time.Millisecond) {
+		for i, addr := range backendAddrs {
+			if _, ok := activeAddrs[addr]; ok || !holds[i].IsStarted() {
+				continue
+			}
+			readyBackendIdx = i
+			activeAddrs[addr] = true
+			holds[i].Resume()
+			break
+		}
 	}
-	if holdGood.IsStarted() {
-		t.Errorf("Got connection attempt to good backend, expected no connection attempt.")
+
+	if ctx.Err() != nil {
+		t.Fatal("Context timed out waiting for next backend to be contacted.")
 	}
 
-	// Allow the connection attempt to the third address to resume and wait
-	// for the attempt for the final address. No other connection attempts
-	// should be started yet.
-	holdNonExistent2Again := dialer.Hold(unReachableBackends[2])
-	holdNonExistent2.Resume()
-	if !holdGood.Wait(ctx) {
-		t.Fatalf("Timeout waiting for connection attempt to good backend")
+	// Wait for channel to become READY without any pending RPC.
+	testutils.AwaitState(ctx, t, conn, connectivity.Ready)
+}
+
+// Tests that ringhash is able to recover automatically in situations when a
+// READY endpoint is removed by the resolver making the aggregated state
+// TRANSIENT_FAILURE. The test creates 4 endpoints in the following
+// connectivity states: [TF, TF, READY, IDLE]. The test removes the
+// READY endpoint and verifies that the last IDLE endopint is dialed and the
+// channel enters READY.
+func (s) TestRingHash_RecoverWhenResolverRemovesEndpoint(t *testing.T) {
+	const backendsCount = 4
+	backends := startTestServiceBackends(t, backendsCount)
+	backendAddrs := backendAddrs(backends)
+
+	const clusterName = "cluster"
+
+	endpoints := endpointResource(t, clusterName, backendAddrs)
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		ServiceName: clusterName,
+		Policy:      e2e.LoadBalancingPolicyRingHash,
+	})
+	route := headerHashRoute("new_route", virtualHostName, clusterName, "address_hash")
+	listener := e2e.DefaultClientListener(virtualHostName, route.Name)
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
 	}
-	if holdNonExistent0Again.IsStarted() {
-		t.Errorf("Got connection attempt to backend 0 again, expected no connection attempt.")
+
+	dialer := testutils.NewBlockingDialer()
+	dialOpts := []grpc.DialOption{
+		grpc.WithResolvers(xdsResolver),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithContextDialer(dialer.DialContext),
+		grpc.WithConnectParams(fastConnectParams),
 	}
-	if holdNonExistent1Again.IsStarted() {
-		t.Errorf("Got connection attempt to backend 1 again, expected no connection attempt.")
+	conn, err := grpc.NewClient("xds:///test.server", dialOpts...)
+	if err != nil {
+		t.Fatalf("Failed to create client: %s", err)
 	}
-	if holdNonExistent2Again.IsStarted() {
-		t.Errorf("Got connection attempt to backend 2 again, expected no connection attempt.")
+	defer conn.Close()
+
+	// Create holds for each backend address to delay a successful connection
+	// until the end of the test.
+	holds := make([]*testutils.Hold, backendsCount)
+	for i := 0; i < len(backendAddrs); i++ {
+		holds[i] = dialer.Hold(backendAddrs[i])
 	}
 
-	// Allow the final attempt to resume.
-	holdGood.Resume()
+	client := testgrpc.NewTestServiceClient(conn)
 
-	// Wait for channel to become connected without any pending RPC.
+	rpcCtx, rpcCancel := context.WithCancel(ctx)
+	errCh := make(chan error, 1)
+	go func() {
+		rpcCtx = metadata.NewOutgoingContext(rpcCtx, metadata.Pairs("address_hash", backendAddrs[0]+"_0"))
+		_, err := client.EmptyCall(rpcCtx, &testpb.Empty{})
+		if status.Code(err) == codes.Canceled {
+			errCh <- nil
+			return
+		}
+		errCh <- err
+	}()
+
+	// Wait for the RPC to trigger a connection attempt to the first address,
+	// then cancel the RPC.  No other connection attempts should be started yet.
+	if !holds[0].Wait(ctx) {
+		t.Fatalf("Timeout waiting for connection attempt to backend 0")
+	}
+	rpcCancel()
+	if err := <-errCh; err != nil {
+		t.Fatalf("Expected RPC to fail be canceled, got %v", err)
+	}
+
+	// The number of dialed backends increases by 1 in every iteration of the
+	// loop as ringhash tries to exit TRANSIENT_FAILURE. Run the loop twice to
+	// get two endpoints in TRANSIENT_FAILURE.
+	activeAddrs := map[string]bool{}
+	for wantFailingBackendCount := 1; wantFailingBackendCount <= 2; wantFailingBackendCount++ {
+		newAddrIdx := -1
+		for ; ctx.Err() == nil && len(activeAddrs) < wantFailingBackendCount; <-time.After(time.Millisecond) {
+			for i, hold := range holds {
+				if !hold.IsStarted() {
+					continue
+				}
+				if _, ok := activeAddrs[backendAddrs[i]]; ok {
+					continue
+				}
+				activeAddrs[backendAddrs[i]] = true
+				newAddrIdx = i
+			}
+		}
+
+		if ctx.Err() != nil {
+			t.Fatal("Context timed out waiting for new backneds to be dialed.")
+		}
+		if len(activeAddrs) > wantFailingBackendCount {
+			t.Fatalf("More backends dialed than expected: got %d, want %d", len(activeAddrs), wantFailingBackendCount)
+		}
+
+		// Create a new hold for the address dialed in this iteration and fail
+		// the existing hold.
+		hold := holds[newAddrIdx]
+		holds[newAddrIdx] = dialer.Hold(backendAddrs[newAddrIdx])
+		hold.Fail(errors.New("Test error"))
+	}
+
+	// Current state of endpoints: [TF, TF, READY, IDLE].
+	// Two endpoints failing should cause the channel to enter
+	// TRANSIENT_FAILURE.
+	testutils.AwaitState(ctx, t, conn, connectivity.TransientFailure)
+
+	// Allow the request to the backend dialed next to succeed.
+	readyBackendIdx := -1
+	for ; ctx.Err() == nil && readyBackendIdx == -1; <-time.After(time.Millisecond) {
+		for i, addr := range backendAddrs {
+			if _, ok := activeAddrs[addr]; ok || !holds[i].IsStarted() {
+				continue
+			}
+			readyBackendIdx = i
+			activeAddrs[addr] = true
+			holds[i].Resume()
+			break
+		}
+	}
+
+	if ctx.Err() != nil {
+		t.Fatal("Context timed out waiting for the next backend to be contacted.")
+	}
+
+	// Wait for channel to become READY without any pending RPC.
 	testutils.AwaitState(ctx, t, conn, connectivity.Ready)
 
-	// No other connection attempts should have been started
-	if holdNonExistent0Again.IsStarted() {
-		t.Errorf("Got connection attempt to backend 0 again, expected no connection attempt.")
+	// Current state of endpoints: [TF, TF, READY, IDLE].
+	// Removing the READY backend should cause the channel to re-enter
+	// TRANSIENT_FAILURE.
+	updatedAddrs := append([]string{}, backendAddrs[:readyBackendIdx]...)
+	updatedAddrs = append(updatedAddrs, backendAddrs[readyBackendIdx+1:]...)
+	updatedEndpoints := endpointResource(t, clusterName, updatedAddrs)
+	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, updatedEndpoints, cluster, route, listener)); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
 	}
-	if holdNonExistent1Again.IsStarted() {
-		t.Errorf("Got connection attempt to backend 1 again, expected no connection attempt.")
+	testutils.AwaitState(ctx, t, conn, connectivity.TransientFailure)
+
+	// To recover from TRANSIENT_FAILURE, ringhash should automatically try to
+	// connect to the final endpoint.
+	readyBackendIdx = -1
+	for ; ctx.Err() == nil && readyBackendIdx == -1; <-time.After(time.Millisecond) {
+		for i, addr := range backendAddrs {
+			if _, ok := activeAddrs[addr]; ok || !holds[i].IsStarted() {
+				continue
+			}
+			readyBackendIdx = i
+			activeAddrs[addr] = true
+			holds[i].Resume()
+			break
+		}
 	}
-	if holdNonExistent2Again.IsStarted() {
-		t.Errorf("Got connection attempt to backend 2 again, expected no connection attempt.")
+
+	if ctx.Err() != nil {
+		t.Fatal("Context timed out waiting for next backend to be contacted.")
 	}
+
+	// Wait for channel to become READY without any pending RPC.
+	testutils.AwaitState(ctx, t, conn, connectivity.Ready)
 }
diff --git a/xds/internal/balancer/ringhash/picker_test.go b/xds/internal/balancer/ringhash/picker_test.go
index 2fd4d947..ff5b6058 100644
--- a/xds/internal/balancer/ringhash/picker_test.go
+++ b/xds/internal/balancer/ringhash/picker_test.go
@@ -24,12 +24,11 @@ import (
 	"testing"
 	"time"
 
-	"github.com/google/go-cmp/cmp"
 	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/connectivity"
-	"google.golang.org/grpc/grpclog"
-	igrpclog "google.golang.org/grpc/internal/grpclog"
 	"google.golang.org/grpc/internal/testutils"
+
+	internalgrpclog "google.golang.org/grpc/internal/grpclog"
 )
 
 var testSubConns []*testutils.TestSubConn
@@ -40,76 +39,108 @@ func init() {
 	}
 }
 
-func newTestRing(cStats []connectivity.State) *ring {
+// fakeChildPicker is used to mock pickers from child pickfirst balancers.
+type fakeChildPicker struct {
+	connectivityState connectivity.State
+	subConn           *testutils.TestSubConn
+	tfError           error
+}
+
+func (p *fakeChildPicker) Pick(balancer.PickInfo) (balancer.PickResult, error) {
+	switch p.connectivityState {
+	case connectivity.Idle:
+		p.subConn.Connect()
+		return balancer.PickResult{}, balancer.ErrNoSubConnAvailable
+	case connectivity.Connecting:
+		return balancer.PickResult{}, balancer.ErrNoSubConnAvailable
+	case connectivity.Ready:
+		return balancer.PickResult{SubConn: p.subConn}, nil
+	default:
+		return balancer.PickResult{}, p.tfError
+	}
+}
+
+func testRingAndEndpointStates(states []connectivity.State) (*ring, map[string]balancer.State) {
 	var items []*ringEntry
-	for i, st := range cStats {
+	epStates := map[string]balancer.State{}
+	for i, st := range states {
 		testSC := testSubConns[i]
 		items = append(items, &ringEntry{
-			idx:  i,
-			hash: uint64((i + 1) * 10),
-			sc: &subConn{
-				addr:  testSC.String(),
-				sc:    testSC,
-				state: st,
-			},
+			idx:       i,
+			hash:      uint64((i + 1) * 10),
+			firstAddr: testSC.String(),
 		})
+		epState := balancer.State{
+			ConnectivityState: st,
+			Picker: &fakeChildPicker{
+				connectivityState: st,
+				tfError:           fmt.Errorf("%d", i),
+				subConn:           testSC,
+			},
+		}
+		epStates[testSC.String()] = epState
 	}
-	return &ring{items: items}
+	return &ring{items: items}, epStates
 }
 
 func (s) TestPickerPickFirstTwo(t *testing.T) {
 	tests := []struct {
-		name            string
-		ring            *ring
-		hash            uint64
-		wantSC          balancer.SubConn
-		wantErr         error
-		wantSCToConnect balancer.SubConn
+		name               string
+		connectivityStates []connectivity.State
+		hash               uint64
+		wantSC             balancer.SubConn
+		wantErr            error
+		wantSCToConnect    balancer.SubConn
 	}{
 		{
-			name:   "picked is Ready",
-			ring:   newTestRing([]connectivity.State{connectivity.Ready, connectivity.Idle}),
-			hash:   5,
-			wantSC: testSubConns[0],
+			name:               "picked is Ready",
+			connectivityStates: []connectivity.State{connectivity.Ready, connectivity.Idle},
+			hash:               5,
+			wantSC:             testSubConns[0],
 		},
 		{
-			name:    "picked is connecting, queue",
-			ring:    newTestRing([]connectivity.State{connectivity.Connecting, connectivity.Idle}),
-			hash:    5,
-			wantErr: balancer.ErrNoSubConnAvailable,
+			name:               "picked is connecting, queue",
+			connectivityStates: []connectivity.State{connectivity.Connecting, connectivity.Idle},
+			hash:               5,
+			wantErr:            balancer.ErrNoSubConnAvailable,
 		},
 		{
-			name:            "picked is Idle, connect and queue",
-			ring:            newTestRing([]connectivity.State{connectivity.Idle, connectivity.Idle}),
-			hash:            5,
-			wantErr:         balancer.ErrNoSubConnAvailable,
-			wantSCToConnect: testSubConns[0],
+			name:               "picked is Idle, connect and queue",
+			connectivityStates: []connectivity.State{connectivity.Idle, connectivity.Idle},
+			hash:               5,
+			wantErr:            balancer.ErrNoSubConnAvailable,
+			wantSCToConnect:    testSubConns[0],
 		},
 		{
-			name:   "picked is TransientFailure, next is ready, return",
-			ring:   newTestRing([]connectivity.State{connectivity.TransientFailure, connectivity.Ready}),
-			hash:   5,
-			wantSC: testSubConns[1],
+			name:               "picked is TransientFailure, next is ready, return",
+			connectivityStates: []connectivity.State{connectivity.TransientFailure, connectivity.Ready},
+			hash:               5,
+			wantSC:             testSubConns[1],
 		},
 		{
-			name:    "picked is TransientFailure, next is connecting, queue",
-			ring:    newTestRing([]connectivity.State{connectivity.TransientFailure, connectivity.Connecting}),
-			hash:    5,
-			wantErr: balancer.ErrNoSubConnAvailable,
+			name:               "picked is TransientFailure, next is connecting, queue",
+			connectivityStates: []connectivity.State{connectivity.TransientFailure, connectivity.Connecting},
+			hash:               5,
+			wantErr:            balancer.ErrNoSubConnAvailable,
 		},
 		{
-			name:            "picked is TransientFailure, next is Idle, connect and queue",
-			ring:            newTestRing([]connectivity.State{connectivity.TransientFailure, connectivity.Idle}),
-			hash:            5,
-			wantErr:         balancer.ErrNoSubConnAvailable,
-			wantSCToConnect: testSubConns[1],
+			name:               "picked is TransientFailure, next is Idle, connect and queue",
+			connectivityStates: []connectivity.State{connectivity.TransientFailure, connectivity.Idle},
+			hash:               5,
+			wantErr:            balancer.ErrNoSubConnAvailable,
+			wantSCToConnect:    testSubConns[1],
 		},
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			p := newPicker(tt.ring, igrpclog.NewPrefixLogger(grpclog.Component("xds"), "rh_test"))
+			ring, epStates := testRingAndEndpointStates(tt.connectivityStates)
+			p := &picker{
+				ring:           ring,
+				logger:         internalgrpclog.NewPrefixLogger(logger, "test-ringhash-picker"),
+				endpointStates: epStates,
+			}
 			got, err := p.Pick(balancer.PickInfo{
 				Ctx: SetRequestHash(ctx, tt.hash),
 			})
@@ -130,175 +161,3 @@ func (s) TestPickerPickFirstTwo(t *testing.T) {
 		})
 	}
 }
-
-// TestPickerPickTriggerTFConnect covers that if the picked SubConn is
-// TransientFailures, all SubConns until a non-TransientFailure are queued for
-// Connect().
-func (s) TestPickerPickTriggerTFConnect(t *testing.T) {
-	ring := newTestRing([]connectivity.State{
-		connectivity.TransientFailure, connectivity.TransientFailure, connectivity.TransientFailure, connectivity.TransientFailure,
-		connectivity.Idle, connectivity.TransientFailure, connectivity.TransientFailure, connectivity.TransientFailure,
-	})
-	p := newPicker(ring, igrpclog.NewPrefixLogger(grpclog.Component("xds"), "rh_test"))
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	_, err := p.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, 5)})
-	if err == nil {
-		t.Fatalf("Pick() error = %v, want non-nil", err)
-	}
-	// The first 4 SubConns, all in TransientFailure, should be queued to
-	// connect.
-	for i := 0; i < 4; i++ {
-		it := ring.items[i]
-		if !it.sc.connectQueued {
-			t.Errorf("the %d-th SubConn is not queued for connect", i)
-		}
-	}
-	// The other SubConns, after the first Idle, should not be queued to
-	// connect.
-	for i := 5; i < len(ring.items); i++ {
-		it := ring.items[i]
-		if it.sc.connectQueued {
-			t.Errorf("the %d-th SubConn is unexpected queued for connect", i)
-		}
-	}
-}
-
-// TestPickerPickTriggerTFReturnReady covers that if the picked SubConn is
-// TransientFailure, SubConn 2 and 3 are TransientFailure, 4 is Ready. SubConn 2
-// and 3 will Connect(), and 4 will be returned.
-func (s) TestPickerPickTriggerTFReturnReady(t *testing.T) {
-	ring := newTestRing([]connectivity.State{
-		connectivity.TransientFailure, connectivity.TransientFailure, connectivity.TransientFailure, connectivity.Ready,
-	})
-	p := newPicker(ring, igrpclog.NewPrefixLogger(grpclog.Component("xds"), "rh_test"))
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	pr, err := p.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, 5)})
-	if err != nil {
-		t.Fatalf("Pick() error = %v, want nil", err)
-	}
-	if wantSC := testSubConns[3]; pr.SubConn != wantSC {
-		t.Fatalf("Pick() = %v, want %v", pr.SubConn, wantSC)
-	}
-	// The first 3 SubConns, all in TransientFailure, should be queued to
-	// connect.
-	for i := 0; i < 3; i++ {
-		it := ring.items[i]
-		if !it.sc.connectQueued {
-			t.Errorf("the %d-th SubConn is not queued for connect", i)
-		}
-	}
-}
-
-// TestPickerPickTriggerTFWithIdle covers that if the picked SubConn is
-// TransientFailure, SubConn 2 is TransientFailure, 3 is Idle (init Idle). Pick
-// will be queue, SubConn 3 will Connect(), SubConn 4 and 5 (in TransientFailure)
-// will not queue a Connect.
-func (s) TestPickerPickTriggerTFWithIdle(t *testing.T) {
-	ring := newTestRing([]connectivity.State{
-		connectivity.TransientFailure, connectivity.TransientFailure, connectivity.Idle, connectivity.TransientFailure, connectivity.TransientFailure,
-	})
-	p := newPicker(ring, igrpclog.NewPrefixLogger(grpclog.Component("xds"), "rh_test"))
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
-	_, err := p.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, 5)})
-	if err == balancer.ErrNoSubConnAvailable {
-		t.Fatalf("Pick() error = %v, want %v", err, balancer.ErrNoSubConnAvailable)
-	}
-	// The first 2 SubConns, all in TransientFailure, should be queued to
-	// connect.
-	for i := 0; i < 2; i++ {
-		it := ring.items[i]
-		if !it.sc.connectQueued {
-			t.Errorf("the %d-th SubConn is not queued for connect", i)
-		}
-	}
-	// SubConn 3 was in Idle, so should Connect()
-	select {
-	case <-testSubConns[2].ConnectCh:
-	case <-time.After(defaultTestShortTimeout):
-		t.Errorf("timeout waiting for Connect() from SubConn %v", testSubConns[2])
-	}
-	// The other SubConns, after the first Idle, should not be queued to
-	// connect.
-	for i := 3; i < len(ring.items); i++ {
-		it := ring.items[i]
-		if it.sc.connectQueued {
-			t.Errorf("the %d-th SubConn is unexpected queued for connect", i)
-		}
-	}
-}
-
-func (s) TestNextSkippingDuplicatesNoDup(t *testing.T) {
-	testRing := newTestRing([]connectivity.State{connectivity.Idle, connectivity.Idle})
-	tests := []struct {
-		name string
-		ring *ring
-		cur  *ringEntry
-		want *ringEntry
-	}{
-		{
-			name: "no dup",
-			ring: testRing,
-			cur:  testRing.items[0],
-			want: testRing.items[1],
-		},
-		{
-			name: "only one entry",
-			ring: &ring{items: []*ringEntry{testRing.items[0]}},
-			cur:  testRing.items[0],
-			want: nil,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := nextSkippingDuplicates(tt.ring, tt.cur); !cmp.Equal(got, tt.want, cmpOpts) {
-				t.Errorf("nextSkippingDuplicates() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-// addDups adds duplicates of items[0] to the ring.
-func addDups(r *ring, count int) *ring {
-	var (
-		items []*ringEntry
-		idx   int
-	)
-	for i, it := range r.items {
-		itt := *it
-		itt.idx = idx
-		items = append(items, &itt)
-		idx++
-		if i == 0 {
-			// Add duplicate of items[0] to the ring
-			for j := 0; j < count; j++ {
-				itt2 := *it
-				itt2.idx = idx
-				items = append(items, &itt2)
-				idx++
-			}
-		}
-	}
-	return &ring{items: items}
-}
-
-func (s) TestNextSkippingDuplicatesMoreDup(t *testing.T) {
-	testRing := newTestRing([]connectivity.State{connectivity.Idle, connectivity.Idle})
-	// Make a new ring with duplicate SubConns.
-	dupTestRing := addDups(testRing, 3)
-	if got := nextSkippingDuplicates(dupTestRing, dupTestRing.items[0]); !cmp.Equal(got, dupTestRing.items[len(dupTestRing.items)-1], cmpOpts) {
-		t.Errorf("nextSkippingDuplicates() = %v, want %v", got, dupTestRing.items[len(dupTestRing.items)-1])
-	}
-}
-
-func (s) TestNextSkippingDuplicatesOnlyDup(t *testing.T) {
-	testRing := newTestRing([]connectivity.State{connectivity.Idle})
-	// Make a new ring with only duplicate SubConns.
-	dupTestRing := addDups(testRing, 3)
-	// This ring only has duplicates of items[0], should return nil.
-	if got := nextSkippingDuplicates(dupTestRing, dupTestRing.items[0]); got != nil {
-		t.Errorf("nextSkippingDuplicates() = %v, want nil", got)
-	}
-}
diff --git a/xds/internal/balancer/ringhash/ring_test.go b/xds/internal/balancer/ringhash/ring_test.go
index 8bca19ba..e714a54e 100644
--- a/xds/internal/balancer/ringhash/ring_test.go
+++ b/xds/internal/balancer/ringhash/ring_test.go
@@ -28,23 +28,24 @@ import (
 	"google.golang.org/grpc/resolver"
 )
 
-var testAddrs []resolver.Address
-var testSubConnMap *resolver.AddressMap
+var testEndpoints []resolver.Endpoint
+var testEndpointStateMap *resolver.EndpointMap
 
 func init() {
-	testAddrs = []resolver.Address{
-		testAddr("a", 3),
-		testAddr("b", 3),
-		testAddr("c", 4),
+	testEndpoints = []resolver.Endpoint{
+		testEndpoint("a", 3),
+		testEndpoint("b", 3),
+		testEndpoint("c", 4),
 	}
-	testSubConnMap = resolver.NewAddressMap()
-	testSubConnMap.Set(testAddrs[0], &subConn{addr: "a", weight: 3})
-	testSubConnMap.Set(testAddrs[1], &subConn{addr: "b", weight: 3})
-	testSubConnMap.Set(testAddrs[2], &subConn{addr: "c", weight: 4})
+	testEndpointStateMap = resolver.NewEndpointMap()
+	testEndpointStateMap.Set(testEndpoints[0], &endpointState{firstAddr: "a", weight: 3})
+	testEndpointStateMap.Set(testEndpoints[1], &endpointState{firstAddr: "b", weight: 3})
+	testEndpointStateMap.Set(testEndpoints[2], &endpointState{firstAddr: "c", weight: 4})
 }
 
-func testAddr(addr string, weight uint32) resolver.Address {
-	return weightedroundrobin.SetAddrInfo(resolver.Address{Addr: addr}, weightedroundrobin.AddrInfo{Weight: weight})
+func testEndpoint(addr string, weight uint32) resolver.Endpoint {
+	ep := resolver.Endpoint{Addresses: []resolver.Address{{Addr: addr}}}
+	return weightedroundrobin.SetAddrInfoInEndpoint(ep, weightedroundrobin.AddrInfo{Weight: weight})
 }
 
 func (s) TestRingNew(t *testing.T) {
@@ -52,20 +53,20 @@ func (s) TestRingNew(t *testing.T) {
 	for _, min := range []uint64{3, 4, 6, 8} {
 		for _, max := range []uint64{20, 8} {
 			t.Run(fmt.Sprintf("size-min-%v-max-%v", min, max), func(t *testing.T) {
-				r := newRing(testSubConnMap, min, max, nil)
+				r := newRing(testEndpointStateMap, min, max, nil)
 				totalCount := len(r.items)
 				if totalCount < int(min) || totalCount > int(max) {
 					t.Fatalf("unexpected size %v, want min %v, max %v", totalCount, min, max)
 				}
-				for _, a := range testAddrs {
+				for _, e := range testEndpoints {
 					var count int
 					for _, ii := range r.items {
-						if ii.sc.addr == a.Addr {
+						if ii.firstAddr == e.Addresses[0].Addr {
 							count++
 						}
 					}
 					got := float64(count) / float64(totalCount)
-					want := float64(getWeightAttribute(a)) / totalWeight
+					want := float64(getWeightAttribute(e)) / totalWeight
 					if !equalApproximately(got, want) {
 						t.Fatalf("unexpected item weight in ring: %v != %v", got, want)
 					}
@@ -82,7 +83,7 @@ func equalApproximately(x, y float64) bool {
 }
 
 func (s) TestRingPick(t *testing.T) {
-	r := newRing(testSubConnMap, 10, 20, nil)
+	r := newRing(testEndpointStateMap, 10, 20, nil)
 	for _, h := range []uint64{xxhash.Sum64String("1"), xxhash.Sum64String("2"), xxhash.Sum64String("3"), xxhash.Sum64String("4")} {
 		t.Run(fmt.Sprintf("picking-hash-%v", h), func(t *testing.T) {
 			e := r.pick(h)
@@ -100,7 +101,7 @@ func (s) TestRingPick(t *testing.T) {
 }
 
 func (s) TestRingNext(t *testing.T) {
-	r := newRing(testSubConnMap, 10, 20, nil)
+	r := newRing(testEndpointStateMap, 10, 20, nil)
 
 	for _, e := range r.items {
 		ne := r.next(e)
diff --git a/xds/internal/balancer/ringhash/ringhash_test.go b/xds/internal/balancer/ringhash/ringhash_test.go
index 45d216ce..73c3fd0d 100644
--- a/xds/internal/balancer/ringhash/ringhash_test.go
+++ b/xds/internal/balancer/ringhash/ringhash_test.go
@@ -24,9 +24,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
-	"google.golang.org/grpc/attributes"
 	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/balancer/weightedroundrobin"
 	"google.golang.org/grpc/connectivity"
@@ -36,14 +33,6 @@ import (
 	"google.golang.org/grpc/xds/internal"
 )
 
-var (
-	cmpOpts = cmp.Options{
-		cmp.AllowUnexported(testutils.TestSubConn{}, ringEntry{}, subConn{}),
-		cmpopts.IgnoreFields(subConn{}, "mu"),
-		cmpopts.IgnoreFields(testutils.TestSubConn{}, "connectCalled"),
-	}
-)
-
 const (
 	defaultTestTimeout      = 10 * time.Second
 	defaultTestShortTimeout = 10 * time.Millisecond
@@ -63,7 +52,7 @@ func init() {
 }
 
 // setupTest creates the balancer, and does an initial sanity check.
-func setupTest(t *testing.T, addrs []resolver.Address) (*testutils.BalancerClientConn, balancer.Balancer, balancer.Picker) {
+func setupTest(t *testing.T, endpoints []resolver.Endpoint) (*testutils.BalancerClientConn, balancer.Balancer, balancer.Picker) {
 	t.Helper()
 	cc := testutils.NewBalancerClientConn(t)
 	builder := balancer.Get(Name)
@@ -72,28 +61,32 @@ func setupTest(t *testing.T, addrs []resolver.Address) (*testutils.BalancerClien
 		t.Fatalf("builder.Build(%s) failed and returned nil", Name)
 	}
 	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState:  resolver.State{Addresses: addrs},
+		ResolverState:  resolver.State{Endpoints: endpoints},
 		BalancerConfig: testConfig,
 	}); err != nil {
 		t.Fatalf("UpdateClientConnState returned err: %v", err)
 	}
 
-	for _, addr := range addrs {
-		addr1 := <-cc.NewSubConnAddrsCh
-		if want := []resolver.Address{addr}; !cmp.Equal(addr1, want, cmp.AllowUnexported(attributes.Attributes{})) {
-			t.Fatalf("got unexpected new subconn addrs: %v", cmp.Diff(addr1, want, cmp.AllowUnexported(attributes.Attributes{})))
-		}
-		sc1 := <-cc.NewSubConnCh
-		// All the SubConns start in Idle, and should not Connect().
-		select {
-		case <-sc1.ConnectCh:
-			t.Errorf("unexpected Connect() from SubConn %v", sc1)
-		case <-time.After(defaultTestShortTimeout):
-		}
+	// The leaf pickfirst are created lazily, only when their endpoint is picked
+	// or other endpoints are in TF. No SubConns should be created immediately.
+	select {
+	case sc := <-cc.NewSubConnCh:
+		t.Errorf("unexpected SubConn creation: %v", sc)
+	case <-time.After(defaultTestShortTimeout):
 	}
 
-	// Should also have a picker, with all SubConns in Idle.
+	// Should also have a picker, with all endpoints in Idle.
 	p1 := <-cc.NewPickerCh
+
+	ringHashPicker := p1.(*picker)
+	if got, want := len(ringHashPicker.endpointStates), len(endpoints); got != want {
+		t.Errorf("Number of child balancers = %d, want = %d", got, want)
+	}
+	for firstAddr, bs := range ringHashPicker.endpointStates {
+		if got, want := bs.ConnectivityState, connectivity.Idle; got != want {
+			t.Errorf("Child balancer connectivity state for address %q = %v, want = %v", firstAddr, got, want)
+		}
+	}
 	return cc, b, p1
 }
 
@@ -113,15 +106,15 @@ func (s) TestUpdateClientConnState_NewRingSize(t *testing.T) {
 	origMinRingSize, origMaxRingSize := 1, 10 // Configured from `testConfig` in `setupTest`
 	newMinRingSize, newMaxRingSize := 20, 100
 
-	addrs := []resolver.Address{{Addr: testBackendAddrStrs[0]}}
-	cc, b, p1 := setupTest(t, addrs)
+	endpoints := []resolver.Endpoint{{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[0]}}}}
+	cc, b, p1 := setupTest(t, endpoints)
 	ring1 := p1.(*picker).ring
 	if ringSize := len(ring1.items); ringSize < origMinRingSize || ringSize > origMaxRingSize {
 		t.Fatalf("Ring created with size %d, want between [%d, %d]", ringSize, origMinRingSize, origMaxRingSize)
 	}
 
 	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState:  resolver.State{Addresses: addrs},
+		ResolverState:  resolver.State{Endpoints: endpoints},
 		BalancerConfig: &LBConfig{MinRingSize: uint64(newMinRingSize), MaxRingSize: uint64(newMaxRingSize)},
 	}); err != nil {
 		t.Fatalf("UpdateClientConnState returned err: %v", err)
@@ -139,23 +132,30 @@ func (s) TestUpdateClientConnState_NewRingSize(t *testing.T) {
 	}
 }
 
-func (s) TestOneSubConn(t *testing.T) {
+func (s) TestOneEndpoint(t *testing.T) {
 	wantAddr1 := resolver.Address{Addr: testBackendAddrStrs[0]}
-	cc, _, p0 := setupTest(t, []resolver.Address{wantAddr1})
+	cc, _, p0 := setupTest(t, []resolver.Endpoint{{Addresses: []resolver.Address{wantAddr1}}})
 	ring0 := p0.(*picker).ring
 
 	firstHash := ring0.items[0].hash
 	// firstHash-1 will pick the first (and only) SubConn from the ring.
 	testHash := firstHash - 1
-	// The first pick should be queued, and should trigger Connect() on the only
-	// SubConn.
+	// The first pick should be queued, and should trigger a connection to the
+	// only Endpoint which has a single address.
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-
 	if _, err := p0.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, testHash)}); err != balancer.ErrNoSubConnAvailable {
 		t.Fatalf("first pick returned err %v, want %v", err, balancer.ErrNoSubConnAvailable)
 	}
-	sc0 := ring0.items[0].sc.sc.(*testutils.TestSubConn)
+	var sc0 *testutils.TestSubConn
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for SubConn creation.")
+	case sc0 = <-cc.NewSubConnCh:
+	}
+	if got, want := sc0.Addresses[0].Addr, wantAddr1.Addr; got != want {
+		t.Fatalf("SubConn.Addresses = %v, want = %v", got, want)
+	}
 	select {
 	case <-sc0.ConnectCh:
 	case <-time.After(defaultTestTimeout):
@@ -165,6 +165,9 @@ func (s) TestOneSubConn(t *testing.T) {
 	// Send state updates to Ready.
 	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
 	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+	if err := cc.WaitForConnectivityState(ctx, connectivity.Ready); err != nil {
+		t.Fatal(err)
+	}
 
 	// Test pick with one backend.
 	p1 := <-cc.NewPickerCh
@@ -180,18 +183,23 @@ func (s) TestOneSubConn(t *testing.T) {
 // same hash always pick the same SubConn. When the one picked is down, another
 // one will be picked.
 func (s) TestThreeSubConnsAffinity(t *testing.T) {
-	wantAddrs := []resolver.Address{
-		{Addr: testBackendAddrStrs[0]},
-		{Addr: testBackendAddrStrs[1]},
-		{Addr: testBackendAddrStrs[2]},
+	endpoints := []resolver.Endpoint{
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[0]}}},
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[1]}}},
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[2]}}},
 	}
-	cc, _, p0 := setupTest(t, wantAddrs)
+	remainingAddrs := map[string]bool{
+		testBackendAddrStrs[0]: true,
+		testBackendAddrStrs[1]: true,
+		testBackendAddrStrs[2]: true,
+	}
+	cc, _, p0 := setupTest(t, endpoints)
 	// This test doesn't update addresses, so this ring will be used by all the
 	// pickers.
-	ring0 := p0.(*picker).ring
+	ring := p0.(*picker).ring
 
-	firstHash := ring0.items[0].hash
-	// firstHash+1 will pick the second SubConn from the ring.
+	firstHash := ring.items[0].hash
+	// firstHash+1 will pick the second endpoint from the ring.
 	testHash := firstHash + 1
 	// The first pick should be queued, and should trigger Connect() on the only
 	// SubConn.
@@ -200,73 +208,119 @@ func (s) TestThreeSubConnsAffinity(t *testing.T) {
 	if _, err := p0.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, testHash)}); err != balancer.ErrNoSubConnAvailable {
 		t.Fatalf("first pick returned err %v, want %v", err, balancer.ErrNoSubConnAvailable)
 	}
-	// The picked SubConn should be the second in the ring.
-	sc0 := ring0.items[1].sc.sc.(*testutils.TestSubConn)
+
+	// The picked endpoint should be the second in the ring.
+	var subConns [3]*testutils.TestSubConn
 	select {
-	case <-sc0.ConnectCh:
-	case <-time.After(defaultTestTimeout):
-		t.Errorf("timeout waiting for Connect() from SubConn %v", sc0)
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for SubConn creation.")
+	case subConns[1] = <-cc.NewSubConnCh:
 	}
-
-	// Send state updates to Ready.
-	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
-	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
-	p1 := <-cc.NewPickerCh
-	for i := 0; i < 5; i++ {
-		gotSCSt, _ := p1.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, testHash)})
-		if gotSCSt.SubConn != sc0 {
-			t.Fatalf("picker.Pick, got %v, want SubConn=%v", gotSCSt, sc0)
-		}
+	if got, want := subConns[1].Addresses[0].Addr, ring.items[1].firstAddr; got != want {
+		t.Fatalf("SubConn.Address = %v, want = %v", got, want)
 	}
+	select {
+	case <-subConns[1].ConnectCh:
+	case <-time.After(defaultTestTimeout):
+		t.Errorf("timeout waiting for Connect() from SubConn %v", subConns[1])
+	}
+	delete(remainingAddrs, ring.items[1].firstAddr)
 
 	// Turn down the subConn in use.
-	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
-	p2 := <-cc.NewPickerCh
-	// Pick with the same hash should be queued, because the SubConn after the
-	// first picked is Idle.
-	if _, err := p2.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, testHash)}); err != balancer.ErrNoSubConnAvailable {
-		t.Fatalf("first pick returned err %v, want %v", err, balancer.ErrNoSubConnAvailable)
+	subConns[1].UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+	subConns[1].UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
+
+	// This should trigger a connection to a new endpoint.
+	<-cc.NewPickerCh
+	var sc *testutils.TestSubConn
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for SubConn creation.")
+	case sc = <-cc.NewSubConnCh:
+	}
+	scAddr := sc.Addresses[0].Addr
+	if _, ok := remainingAddrs[scAddr]; !ok {
+		t.Fatalf("New SubConn created with previously used address: %q", scAddr)
+	}
+	delete(remainingAddrs, scAddr)
+	select {
+	case <-sc.ConnectCh:
+	case <-time.After(defaultTestTimeout):
+		t.Errorf("timeout waiting for Connect() from SubConn %v", subConns[1])
+	}
+	if scAddr == ring.items[0].firstAddr {
+		subConns[0] = sc
+	} else if scAddr == ring.items[2].firstAddr {
+		subConns[2] = sc
 	}
 
-	// The third SubConn in the ring should connect.
-	sc1 := ring0.items[2].sc.sc.(*testutils.TestSubConn)
+	// Turning down the SubConn should cause creation of a connection to the
+	// final endpoint.
+	sc.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+	sc.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
 	select {
-	case <-sc1.ConnectCh:
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for SubConn creation.")
+	case sc = <-cc.NewSubConnCh:
+	}
+	scAddr = sc.Addresses[0].Addr
+	if _, ok := remainingAddrs[scAddr]; !ok {
+		t.Fatalf("New SubConn created with previously used address: %q", scAddr)
+	}
+	delete(remainingAddrs, scAddr)
+	select {
+	case <-sc.ConnectCh:
 	case <-time.After(defaultTestTimeout):
-		t.Errorf("timeout waiting for Connect() from SubConn %v", sc1)
+		t.Errorf("timeout waiting for Connect() from SubConn %v", subConns[1])
+	}
+	if scAddr == ring.items[0].firstAddr {
+		subConns[0] = sc
+	} else if scAddr == ring.items[2].firstAddr {
+		subConns[2] = sc
 	}
+	sc.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+	sc.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
 
-	// Send state updates to Ready.
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
-	// New picks should all return this SubConn.
-	p3 := <-cc.NewPickerCh
+	// All endpoints are in TransientFailure. Make the first endpoint in the
+	// ring report Ready. All picks should go to this endpoint which is two
+	// indexes away from the endpoint with the chosen hash.
+	subConns[0].UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Idle})
+	subConns[0].UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+	subConns[0].UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+	if err := cc.WaitForConnectivityState(ctx, connectivity.Ready); err != nil {
+		t.Fatalf("Context timed out while waiting for channel to report Ready.")
+	}
+	p1 := <-cc.NewPickerCh
 	for i := 0; i < 5; i++ {
-		gotSCSt, _ := p3.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, testHash)})
-		if gotSCSt.SubConn != sc1 {
-			t.Fatalf("picker.Pick, got %v, want SubConn=%v", gotSCSt, sc1)
+		gotSCSt, _ := p1.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, testHash)})
+		if gotSCSt.SubConn != subConns[0] {
+			t.Fatalf("picker.Pick, got %v, want SubConn=%v", gotSCSt, subConns[0])
 		}
 	}
 
-	// Now, after backoff, the first picked SubConn will turn Idle.
-	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Idle})
-	// The picks above should have queued Connect() for the first picked
-	// SubConn, so this Idle state change will trigger a Connect().
-	select {
-	case <-sc0.ConnectCh:
-	case <-time.After(defaultTestTimeout):
-		t.Errorf("timeout waiting for Connect() from SubConn %v", sc0)
+	// Make the last endpoint in the ring report Ready. All picks should go to
+	// this endpoint since it is one index away from the chosen hash.
+	subConns[2].UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Idle})
+	subConns[2].UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+	subConns[2].UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+	p2 := <-cc.NewPickerCh
+	for i := 0; i < 5; i++ {
+		gotSCSt, _ := p2.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, testHash)})
+		if gotSCSt.SubConn != subConns[2] {
+			t.Fatalf("picker.Pick, got %v, want SubConn=%v", gotSCSt, subConns[2])
+		}
 	}
 
-	// After the first picked SubConn turn Ready, new picks should return it
-	// again (even though the second picked SubConn is also Ready).
-	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
-	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
-	p4 := <-cc.NewPickerCh
+	// Make the second endpoint in the ring report Ready. All picks should go to
+	// this endpoint as it is the one with the chosen hash.
+	subConns[1].UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Idle})
+	subConns[1].UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+	subConns[1].UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+	p3 := <-cc.NewPickerCh
 	for i := 0; i < 5; i++ {
-		gotSCSt, _ := p4.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, testHash)})
-		if gotSCSt.SubConn != sc0 {
-			t.Fatalf("picker.Pick, got %v, want SubConn=%v", gotSCSt, sc0)
+		gotSCSt, _ := p3.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, testHash)})
+		if gotSCSt.SubConn != subConns[1] {
+			t.Fatalf("picker.Pick, got %v, want SubConn=%v", gotSCSt, subConns[1])
 		}
 	}
 }
@@ -274,13 +328,13 @@ func (s) TestThreeSubConnsAffinity(t *testing.T) {
 // TestThreeBackendsAffinity covers that there are 3 SubConns, RPCs with the
 // same hash always pick the same SubConn. Then try different hash to pick
 // another backend, and verify the first hash still picks the first backend.
-func (s) TestThreeSubConnsAffinityMultiple(t *testing.T) {
-	wantAddrs := []resolver.Address{
-		{Addr: testBackendAddrStrs[0]},
-		{Addr: testBackendAddrStrs[1]},
-		{Addr: testBackendAddrStrs[2]},
+func (s) TestThreeBackendsAffinityMultiple(t *testing.T) {
+	wantEndpoints := []resolver.Endpoint{
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[0]}}},
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[1]}}},
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[2]}}},
 	}
-	cc, _, p0 := setupTest(t, wantAddrs)
+	cc, _, p0 := setupTest(t, wantEndpoints)
 	// This test doesn't update addresses, so this ring will be used by all the
 	// pickers.
 	ring0 := p0.(*picker).ring
@@ -292,11 +346,19 @@ func (s) TestThreeSubConnsAffinityMultiple(t *testing.T) {
 	// SubConn.
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-
 	if _, err := p0.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, testHash)}); err != balancer.ErrNoSubConnAvailable {
 		t.Fatalf("first pick returned err %v, want %v", err, balancer.ErrNoSubConnAvailable)
 	}
-	sc0 := ring0.items[1].sc.sc.(*testutils.TestSubConn)
+	// The picked SubConn should be the second in the ring.
+	var sc0 *testutils.TestSubConn
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for SubConn creation.")
+	case sc0 = <-cc.NewSubConnCh:
+	}
+	if got, want := sc0.Addresses[0].Addr, ring0.items[1].firstAddr; got != want {
+		t.Fatalf("SubConn.Address = %v, want = %v", got, want)
+	}
 	select {
 	case <-sc0.ConnectCh:
 	case <-time.After(defaultTestTimeout):
@@ -306,6 +368,9 @@ func (s) TestThreeSubConnsAffinityMultiple(t *testing.T) {
 	// Send state updates to Ready.
 	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
 	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+	if err := cc.WaitForConnectivityState(ctx, connectivity.Ready); err != nil {
+		t.Fatal(err)
+	}
 
 	// First hash should always pick sc0.
 	p1 := <-cc.NewPickerCh
@@ -322,7 +387,15 @@ func (s) TestThreeSubConnsAffinityMultiple(t *testing.T) {
 	if _, err := p0.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, testHash2)}); err != balancer.ErrNoSubConnAvailable {
 		t.Fatalf("first pick returned err %v, want %v", err, balancer.ErrNoSubConnAvailable)
 	}
-	sc1 := ring0.items[2].sc.sc.(*testutils.TestSubConn)
+	var sc1 *testutils.TestSubConn
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for SubConn creation.")
+	case sc1 = <-cc.NewSubConnCh:
+	}
+	if got, want := sc1.Addresses[0].Addr, ring0.items[2].firstAddr; got != want {
+		t.Fatalf("SubConn.Address = %v, want = %v", got, want)
+	}
 	select {
 	case <-sc1.ConnectCh:
 	case <-time.After(defaultTestTimeout):
@@ -357,104 +430,106 @@ func (s) TestThreeSubConnsAffinityMultiple(t *testing.T) {
 //     sent.  And the new ring should contain the correct number of entries
 //     and weights.
 func (s) TestAddrWeightChange(t *testing.T) {
-	addrs := []resolver.Address{
-		{Addr: testBackendAddrStrs[0]},
-		{Addr: testBackendAddrStrs[1]},
-		{Addr: testBackendAddrStrs[2]},
+	endpoints := []resolver.Endpoint{
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[0]}}},
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[1]}}},
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[2]}}},
 	}
-	cc, b, p0 := setupTest(t, addrs)
+	cc, b, p0 := setupTest(t, endpoints)
 	ring0 := p0.(*picker).ring
 
-	// Update with the same addresses, should not send a new Picker.
+	// Update with the same addresses, it will result in a new picker, but with
+	// the same ring.
 	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState:  resolver.State{Addresses: addrs},
+		ResolverState:  resolver.State{Endpoints: endpoints},
 		BalancerConfig: testConfig,
 	}); err != nil {
 		t.Fatalf("UpdateClientConnState returned err: %v", err)
 	}
+	var p1 balancer.Picker
 	select {
-	case <-cc.NewPickerCh:
-		t.Fatalf("unexpected picker after UpdateClientConn with the same addresses")
-	case <-time.After(defaultTestShortTimeout):
+	case p1 = <-cc.NewPickerCh:
+	case <-time.After(defaultTestTimeout):
+		t.Fatalf("timeout waiting for picker after UpdateClientConn with same addresses")
+	}
+	ring1 := p1.(*picker).ring
+	if ring1 != ring0 {
+		t.Fatalf("new picker with same address has a different ring than before, want same")
 	}
 
 	// Delete an address, should send a new Picker.
 	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: resolver.State{Addresses: []resolver.Address{
-			{Addr: testBackendAddrStrs[0]},
-			{Addr: testBackendAddrStrs[1]},
-		}},
+		ResolverState:  resolver.State{Endpoints: endpoints[:2]},
 		BalancerConfig: testConfig,
 	}); err != nil {
 		t.Fatalf("UpdateClientConnState returned err: %v", err)
 	}
-	var p1 balancer.Picker
+	var p2 balancer.Picker
 	select {
-	case p1 = <-cc.NewPickerCh:
+	case p2 = <-cc.NewPickerCh:
 	case <-time.After(defaultTestTimeout):
 		t.Fatalf("timeout waiting for picker after UpdateClientConn with different addresses")
 	}
-	ring1 := p1.(*picker).ring
-	if ring1 == ring0 {
+	ring2 := p2.(*picker).ring
+	if ring2 == ring0 {
 		t.Fatalf("new picker after removing address has the same ring as before, want different")
 	}
 
 	// Another update with the same addresses, but different weight.
 	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: resolver.State{Addresses: []resolver.Address{
-			{Addr: testBackendAddrStrs[0]},
-			weightedroundrobin.SetAddrInfo(
-				resolver.Address{Addr: testBackendAddrStrs[1]},
+		ResolverState: resolver.State{Endpoints: []resolver.Endpoint{
+			endpoints[0],
+			weightedroundrobin.SetAddrInfoInEndpoint(
+				endpoints[1],
 				weightedroundrobin.AddrInfo{Weight: 2}),
 		}},
 		BalancerConfig: testConfig,
 	}); err != nil {
 		t.Fatalf("UpdateClientConnState returned err: %v", err)
 	}
-	var p2 balancer.Picker
+	var p3 balancer.Picker
 	select {
-	case p2 = <-cc.NewPickerCh:
+	case p3 = <-cc.NewPickerCh:
 	case <-time.After(defaultTestTimeout):
 		t.Fatalf("timeout waiting for picker after UpdateClientConn with different addresses")
 	}
-	if p2.(*picker).ring == ring1 {
+	if p3.(*picker).ring == ring2 {
 		t.Fatalf("new picker after changing address weight has the same ring as before, want different")
 	}
 	// With the new update, the ring must look like this:
 	//   [
-	//     {idx:0 sc: {addr: testBackendAddrStrs[0], weight: 1}},
-	//     {idx:1 sc: {addr: testBackendAddrStrs[1], weight: 2}},
-	//     {idx:2 sc: {addr: testBackendAddrStrs[2], weight: 2}},
+	//     {idx:0 endpoint: {addr: testBackendAddrStrs[0], weight: 1}},
+	//     {idx:1 endpoint: {addr: testBackendAddrStrs[1], weight: 2}},
+	//     {idx:2 endpoint: {addr: testBackendAddrStrs[2], weight: 1}},
 	//   ].
-	if len(p2.(*picker).ring.items) != 3 {
-		t.Fatalf("new picker after changing address weight has %d entries, want 3", len(p2.(*picker).ring.items))
+	if len(p3.(*picker).ring.items) != 3 {
+		t.Fatalf("new picker after changing address weight has %d entries, want 3", len(p3.(*picker).ring.items))
 	}
-	for _, i := range p2.(*picker).ring.items {
-		if i.sc.addr == testBackendAddrStrs[0] {
-			if i.sc.weight != 1 {
-				t.Fatalf("new picker after changing address weight has weight %d for %v, want 1", i.sc.weight, i.sc.addr)
+	for _, i := range p3.(*picker).ring.items {
+		if i.firstAddr == testBackendAddrStrs[0] {
+			if i.weight != 1 {
+				t.Fatalf("new picker after changing address weight has weight %d for %v, want 1", i.weight, i.firstAddr)
 			}
 		}
-		if i.sc.addr == testBackendAddrStrs[1] {
-			if i.sc.weight != 2 {
-				t.Fatalf("new picker after changing address weight has weight %d for %v, want 2", i.sc.weight, i.sc.addr)
+		if i.firstAddr == testBackendAddrStrs[1] {
+			if i.weight != 2 {
+				t.Fatalf("new picker after changing address weight has weight %d for %v, want 2", i.weight, i.firstAddr)
 			}
 		}
 	}
 }
 
-// TestSubConnToConnectWhenOverallTransientFailure covers the situation when the
-// overall state is TransientFailure, the SubConns turning Idle will trigger the
-// next SubConn in the ring to Connect(). But not when the overall state is not
-// TransientFailure.
-func (s) TestSubConnToConnectWhenOverallTransientFailure(t *testing.T) {
-	wantAddrs := []resolver.Address{
-		{Addr: testBackendAddrStrs[0]},
-		{Addr: testBackendAddrStrs[1]},
-		{Addr: testBackendAddrStrs[2]},
-	}
-	_, _, p0 := setupTest(t, wantAddrs)
-	ring0 := p0.(*picker).ring
+// TestAutoConnectEndpointOnTransientFailure covers the situation when an
+// endpoint fails. It verifies that a new endpoint is automatically tried
+// (without a pick) when there is no endpoint already in Connecting state.
+func (s) TestAutoConnectEndpointOnTransientFailure(t *testing.T) {
+	wantEndpoints := []resolver.Endpoint{
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[0]}}},
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[1]}}},
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[2]}}},
+		{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[3]}}},
+	}
+	cc, _, p0 := setupTest(t, wantEndpoints)
 
 	// ringhash won't tell SCs to connect until there is an RPC, so simulate
 	// one now.
@@ -462,14 +537,33 @@ func (s) TestSubConnToConnectWhenOverallTransientFailure(t *testing.T) {
 	defer cancel()
 	p0.Pick(balancer.PickInfo{Ctx: ctx})
 
-	// Turn the first subconn to transient failure.
-	sc0 := ring0.items[0].sc.sc.(*testutils.TestSubConn)
+	// The picked SubConn should be the second in the ring.
+	var sc0 *testutils.TestSubConn
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for SubConn creation.")
+	case sc0 = <-cc.NewSubConnCh:
+	}
+	select {
+	case <-sc0.ConnectCh:
+	case <-time.After(defaultTestTimeout):
+		t.Errorf("timeout waiting for Connect() from SubConn %v", sc0)
+	}
+
+	// Turn the first subconn to transient failure. This should set the overall
+	// connectivity state to CONNECTING.
+	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
 	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
-	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Idle})
+	cc.WaitForConnectivityState(ctx, connectivity.Connecting)
 
-	// It will trigger the second subconn to connect (because overall state is
-	// Connect (when one subconn is TF)).
-	sc1 := ring0.items[1].sc.sc.(*testutils.TestSubConn)
+	// It will trigger the second subconn to connect since there is only one
+	// endpoint, which is in TF.
+	var sc1 *testutils.TestSubConn
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for SubConn creation.")
+	case sc1 = <-cc.NewSubConnCh:
+	}
 	select {
 	case <-sc1.ConnectCh:
 	case <-time.After(defaultTestShortTimeout):
@@ -477,35 +571,49 @@ func (s) TestSubConnToConnectWhenOverallTransientFailure(t *testing.T) {
 	}
 
 	// Turn the second subconn to TF. This will set the overall state to TF.
+	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
 	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
-	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Idle})
+	cc.WaitForConnectivityState(ctx, connectivity.TransientFailure)
 
 	// It will trigger the third subconn to connect.
-	sc2 := ring0.items[2].sc.sc.(*testutils.TestSubConn)
+	var sc2 *testutils.TestSubConn
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for SubConn creation.")
+	case sc2 = <-cc.NewSubConnCh:
+	}
 	select {
 	case <-sc2.ConnectCh:
 	case <-time.After(defaultTestShortTimeout):
 		t.Fatalf("timeout waiting for Connect() from SubConn %v", sc2)
 	}
 
-	// Turn the third subconn to TF. This will set the overall state to TF.
-	sc2.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
-	sc2.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Idle})
+	sc2.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
 
-	// It will trigger the first subconn to connect.
+	// Send the first SubConn into CONNECTING. To do this, first make it READY,
+	// then CONNECTING.
+	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+	cc.WaitForConnectivityState(ctx, connectivity.Ready)
+	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Idle})
+	// Since one endpoint is in TF and one in CONNECTING, the aggregated state
+	// will be CONNECTING.
+	cc.WaitForConnectivityState(ctx, connectivity.Connecting)
+	p1 := <-cc.NewPickerCh
+	p1.Pick(balancer.PickInfo{Ctx: ctx})
 	select {
 	case <-sc0.ConnectCh:
-	case <-time.After(defaultTestShortTimeout):
-		t.Fatalf("timeout waiting for Connect() from SubConn %v", sc0)
+	case <-time.After(defaultTestTimeout):
+		t.Errorf("timeout waiting for Connect() from SubConn %v", sc0)
 	}
+	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
 
-	// Turn the third subconn to TF again.
+	// This will not trigger any new SubCOnns to be created, because sc0 is
+	// still attempting to connect, and we only need one SubConn to connect.
 	sc2.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
-	sc2.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Idle})
 
-	// This will not trigger any new Connect() on the SubConns, because sc0 is
-	// still attempting to connect, and we only need one SubConn to connect.
 	select {
+	case sc := <-cc.NewSubConnCh:
+		t.Fatalf("unexpected SubConn creation: %v", sc)
 	case <-sc0.ConnectCh:
 		t.Fatalf("unexpected Connect() from SubConn %v", sc0)
 	case <-sc1.ConnectCh:
@@ -516,52 +624,49 @@ func (s) TestSubConnToConnectWhenOverallTransientFailure(t *testing.T) {
 	}
 }
 
-func (s) TestConnectivityStateEvaluatorRecordTransition(t *testing.T) {
+func (s) TestAggregatedConnectivityState(t *testing.T) {
 	tests := []struct {
-		name     string
-		from, to []connectivity.State
-		want     connectivity.State
+		name           string
+		endpointStates []connectivity.State
+		want           connectivity.State
 	}{
 		{
-			name: "one ready",
-			from: []connectivity.State{connectivity.Idle},
-			to:   []connectivity.State{connectivity.Ready},
-			want: connectivity.Ready,
+			name:           "one ready",
+			endpointStates: []connectivity.State{connectivity.Ready},
+			want:           connectivity.Ready,
 		},
 		{
-			name: "one connecting",
-			from: []connectivity.State{connectivity.Idle},
-			to:   []connectivity.State{connectivity.Connecting},
-			want: connectivity.Connecting,
+			name:           "one connecting",
+			endpointStates: []connectivity.State{connectivity.Connecting},
+			want:           connectivity.Connecting,
 		},
 		{
-			name: "one ready one transient failure",
-			from: []connectivity.State{connectivity.Idle, connectivity.Idle},
-			to:   []connectivity.State{connectivity.Ready, connectivity.TransientFailure},
-			want: connectivity.Ready,
+			name:           "one ready one transient failure",
+			endpointStates: []connectivity.State{connectivity.Ready, connectivity.TransientFailure},
+			want:           connectivity.Ready,
 		},
 		{
-			name: "one connecting one transient failure",
-			from: []connectivity.State{connectivity.Idle, connectivity.Idle},
-			to:   []connectivity.State{connectivity.Connecting, connectivity.TransientFailure},
-			want: connectivity.Connecting,
+			name:           "one connecting one transient failure",
+			endpointStates: []connectivity.State{connectivity.Connecting, connectivity.TransientFailure},
+			want:           connectivity.Connecting,
 		},
 		{
-			name: "one connecting two transient failure",
-			from: []connectivity.State{connectivity.Idle, connectivity.Idle, connectivity.Idle},
-			to:   []connectivity.State{connectivity.Connecting, connectivity.TransientFailure, connectivity.TransientFailure},
-			want: connectivity.TransientFailure,
+			name:           "one connecting two transient failure",
+			endpointStates: []connectivity.State{connectivity.Connecting, connectivity.TransientFailure, connectivity.TransientFailure},
+			want:           connectivity.TransientFailure,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			cse := &connectivityStateEvaluator{}
-			var got connectivity.State
-			for i, fff := range tt.from {
-				ttt := tt.to[i]
-				got = cse.recordTransition(fff, ttt)
+			bal := &ringhashBalancer{endpointStates: resolver.NewEndpointMap()}
+			for i, cs := range tt.endpointStates {
+				es := &endpointState{
+					state: balancer.State{ConnectivityState: cs},
+				}
+				ep := resolver.Endpoint{Addresses: []resolver.Address{{Addr: fmt.Sprintf("%d.%d.%d.%d:%d", i, i, i, i, i)}}}
+				bal.endpointStates.Set(ep, es)
 			}
-			if got != tt.want {
+			if got := bal.aggregatedStateLocked(); got != tt.want {
 				t.Errorf("recordTransition() = %v, want %v", got, tt.want)
 			}
 		})
@@ -570,16 +675,38 @@ func (s) TestConnectivityStateEvaluatorRecordTransition(t *testing.T) {
 
 // TestAddrBalancerAttributesChange tests the case where the ringhash balancer
 // receives a ClientConnUpdate with the same config and addresses as received in
-// the previous update. Although the `BalancerAttributes` contents are the same,
-// the pointer is different. This test verifies that subConns are not recreated
-// in this scenario.
+// the previous update. Although the `BalancerAttributes` and endpoint
+// attributes contents are the same, the pointers are different. This test
+// verifies that subConns are not recreated in this scenario.
 func (s) TestAddrBalancerAttributesChange(t *testing.T) {
-	addrs1 := []resolver.Address{internal.SetLocalityID(resolver.Address{Addr: testBackendAddrStrs[0]}, internal.LocalityID{Region: "americas"})}
-	cc, b, _ := setupTest(t, addrs1)
+	locality := internal.LocalityID{Region: "americas"}
+	addrs1 := []resolver.Address{internal.SetLocalityID(resolver.Address{Addr: testBackendAddrStrs[0]}, locality)}
+	wantEndpoints1 := []resolver.Endpoint{
+		internal.SetLocalityIDInEndpoint(resolver.Endpoint{Addresses: addrs1}, locality),
+	}
+	cc, b, p0 := setupTest(t, wantEndpoints1)
+	ring0 := p0.(*picker).ring
+
+	firstHash := ring0.items[0].hash
+	// The first pick should be queued, and should trigger a connection to the
+	// only Endpoint which has a single address.
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if _, err := p0.Pick(balancer.PickInfo{Ctx: SetRequestHash(ctx, firstHash)}); err != balancer.ErrNoSubConnAvailable {
+		t.Fatalf("first pick returned err %v, want %v", err, balancer.ErrNoSubConnAvailable)
+	}
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for SubConn creation.")
+	case <-cc.NewSubConnCh:
+	}
 
-	addrs2 := []resolver.Address{internal.SetLocalityID(resolver.Address{Addr: testBackendAddrStrs[0]}, internal.LocalityID{Region: "americas"})}
+	addrs2 := []resolver.Address{internal.SetLocalityID(resolver.Address{Addr: testBackendAddrStrs[0]}, locality)}
+	wantEndpoints2 := []resolver.Endpoint{
+		internal.SetLocalityIDInEndpoint(resolver.Endpoint{Addresses: addrs2}, locality),
+	}
 	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState:  resolver.State{Addresses: addrs2},
+		ResolverState:  resolver.State{Endpoints: wantEndpoints2},
 		BalancerConfig: testConfig,
 	}); err != nil {
 		t.Fatalf("UpdateClientConnState returned err: %v", err)