diff --git a/balancer/weightedtarget/weightedtarget_test.go b/balancer/weightedtarget/weightedtarget_test.go
index d78db53..8468892 100644
--- a/balancer/weightedtarget/weightedtarget_test.go
+++ b/balancer/weightedtarget/weightedtarget_test.go
@@ -1348,7 +1348,7 @@
 	return ret
 }
 
-const initIdleBalancerName = "test-init-Idle-balancer"
+const initIdleBalancerName = "test-init-idle-balancer"
 
 var errTestInitIdle = fmt.Errorf("init Idle balancer error 0")
 
@@ -1388,7 +1388,7 @@
   "targets": {
     "cluster_1": {
       "weight":1,
-      "childPolicy": [{"test-init-Idle-balancer": ""}]
+      "childPolicy": [{"test-init-idle-balancer": ""}]
     }
   }
 }`))
diff --git a/internal/balancergroup/balancergroup.go b/internal/balancergroup/balancergroup.go
index cd1a9ed..8acb942 100644
--- a/internal/balancergroup/balancergroup.go
+++ b/internal/balancergroup/balancergroup.go
@@ -338,6 +338,16 @@
 // closed after timeout. Cleanup work (closing sub-balancer and removing
 // subconns) will be done after timeout.
 func (bg *BalancerGroup) Remove(id string) {
+	bg.removeInternal(id, true)
+}
+
+// RemoveImmediately removes and closes the balancer with id from the group
+// immediately.
+func (bg *BalancerGroup) RemoveImmediately(id string) {
+	bg.removeInternal(id, false)
+}
+
+func (bg *BalancerGroup) removeInternal(id string, withCaching bool) {
 	bg.logger.Infof("Removing child policy for child %q", id)
 
 	bg.outgoingMu.Lock()
@@ -356,32 +366,40 @@
 	// Unconditionally remove the sub-balancer config from the map.
 	delete(bg.idToBalancerConfig, id)
 
-	if bg.deletedBalancerCache != nil {
-		if bg.logger.V(2) {
-			bg.logger.Infof("Adding child policy for child %q to the balancer cache", id)
-			bg.logger.Infof("Number of items remaining in the balancer cache: %d", bg.deletedBalancerCache.Len())
-		}
-
-		bg.deletedBalancerCache.Add(id, sbToRemove, func() {
+	if withCaching {
+		if bg.deletedBalancerCache != nil {
 			if bg.logger.V(2) {
-				bg.logger.Infof("Removing child policy for child %q from the balancer cache after timeout", id)
+				bg.logger.Infof("Adding child policy for child %q to the balancer cache", id)
+			}
+			bg.deletedBalancerCache.Add(id, sbToRemove, func() {
+				if bg.logger.V(2) {
+					bg.logger.Infof("Removing child policy for child %q from the balancer cache after timeout", id)
+					bg.logger.Infof("Number of items remaining in the balancer cache: %d", bg.deletedBalancerCache.Len())
+				}
+
+				// A sub-balancer evicted from the timeout cache needs to closed
+				// and its subConns need to removed, unconditionally. There is a
+				// possibility that a sub-balancer might be removed (thereby
+				// moving it to the cache) around the same time that the
+				// balancergroup is closed, and by the time we get here the
+				// balancergroup might be closed.  Check for `outgoingStarted ==
+				// true` at that point can lead to a leaked sub-balancer.
+				bg.outgoingMu.Lock()
+				sbToRemove.stopBalancer()
+				bg.outgoingMu.Unlock()
+				bg.cleanupSubConns(sbToRemove)
+			})
+			if bg.logger.V(2) {
 				bg.logger.Infof("Number of items remaining in the balancer cache: %d", bg.deletedBalancerCache.Len())
 			}
-
-			// A sub-balancer evicted from the timeout cache needs to closed
-			// and its subConns need to removed, unconditionally. There is a
-			// possibility that a sub-balancer might be removed (thereby
-			// moving it to the cache) around the same time that the
-			// balancergroup is closed, and by the time we get here the
-			// balancergroup might be closed.  Check for `outgoingStarted ==
-			// true` at that point can lead to a leaked sub-balancer.
-			bg.outgoingMu.Lock()
-			sbToRemove.stopBalancer()
 			bg.outgoingMu.Unlock()
-			bg.cleanupSubConns(sbToRemove)
-		})
-		bg.outgoingMu.Unlock()
-		return
+			return
+		}
+
+		// Fall through to remove the sub-balancer with immediate effect if we are not caching.
+		if bg.logger.V(2) {
+			bg.logger.Infof("Child policy for child %q was requested to be cached before eventual removal. No such cache exists. Removing right away.", id)
+		}
 	}
 
 	// Remove the sub-balancer with immediate effect if we are not caching.
@@ -481,7 +499,7 @@
 // from map. Delete sc from the map only when state changes to Shutdown. Since
 // it's just forwarding the action, there's no need for a removeSubConn()
 // wrapper function.
-func (bg *BalancerGroup) newSubConn(config *subBalancerWrapper, addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
+func (bg *BalancerGroup) newSubConn(sbw *subBalancerWrapper, addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
 	// NOTE: if balancer with id was already removed, this should also return
 	// error. But since we call balancer.stopBalancer when removing the balancer, this
 	// shouldn't happen.
@@ -493,12 +511,12 @@
 	var sc balancer.SubConn
 	oldListener := opts.StateListener
 	opts.StateListener = func(state balancer.SubConnState) { bg.updateSubConnState(sc, state, oldListener) }
-	sc, err := bg.cc.NewSubConn(addrs, opts)
+	sc, err := sbw.ClientConn.NewSubConn(addrs, opts)
 	if err != nil {
 		bg.incomingMu.Unlock()
 		return nil, err
 	}
-	bg.scToSubBalancer[sc] = config
+	bg.scToSubBalancer[sc] = sbw
 	bg.incomingMu.Unlock()
 	return sc, nil
 }
diff --git a/internal/balancergroup/balancergroup_ext_test.go b/internal/balancergroup/balancergroup_ext_test.go
new file mode 100644
index 0000000..bf53d99
--- /dev/null
+++ b/internal/balancergroup/balancergroup_ext_test.go
@@ -0,0 +1,124 @@
+/*
+ * Copyright 2026 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package balancergroup_test
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc/balancer"
+	"google.golang.org/grpc/internal/balancer/stub"
+	"google.golang.org/grpc/internal/balancergroup"
+	"google.golang.org/grpc/internal/grpctest"
+	"google.golang.org/grpc/internal/testutils"
+)
+
+type s struct {
+	grpctest.Tester
+}
+
+func Test(t *testing.T) {
+	grpctest.RunSubTests(t, s{})
+}
+
+const (
+	defaultTestTimeout      = 5 * time.Second
+	defaultTestShortTimeout = 100 * time.Millisecond
+)
+
+// Tests verifies that the RemoveImmediately method of balancergroup closes the
+// child balancer immediately, while the Remove method does not close the child
+// balancer immediately, when the cache is enabled.
+func (s) TestBalancerGroup_RemoveImmediately(t *testing.T) {
+	// Channels to track child balancer creation and closure.
+	childLBCreated := make(chan string, 1)
+	childLBClosed := make(chan string, 1)
+
+	childLBName1 := strings.ToLower(t.Name()) + "-child-1"
+	t.Logf("Registering a child balancer with name %q", childLBName1)
+	stub.Register(childLBName1, stub.BalancerFuncs{
+		Init: func(*stub.BalancerData) {
+			childLBCreated <- childLBName1
+		},
+		Close: func(*stub.BalancerData) {
+			childLBClosed <- childLBName1
+		},
+	})
+
+	childLBName2 := strings.ToLower(t.Name()) + "-child-2"
+	t.Logf("Registering a child balancer with name %q", childLBName2)
+	stub.Register(childLBName2, stub.BalancerFuncs{
+		Init: func(*stub.BalancerData) {
+			childLBCreated <- childLBName2
+		},
+		Close: func(*stub.BalancerData) {
+			childLBClosed <- childLBName2
+		},
+	})
+
+	t.Logf("Creating a balancergroup with cache enabled")
+	tcc := testutils.NewBalancerClientConn(t)
+	bg := balancergroup.New(balancergroup.Options{
+		CC:                      tcc,
+		BuildOpts:               balancer.BuildOptions{},
+		SubBalancerCloseTimeout: defaultTestTimeout,
+	})
+
+	t.Logf("Adding a child balancer with name %q to the group", "child-1")
+	bg.AddWithClientConn("child-1", childLBName1, tcc)
+	select {
+	case <-childLBCreated:
+	case <-time.After(defaultTestTimeout):
+		t.Fatalf("Timeout when waiting for child LB to be created")
+	}
+
+	t.Logf("Adding a child balancer with name %q to the group", "child-2")
+	bg.AddWithClientConn("child-2", childLBName2, tcc)
+	select {
+	case <-childLBCreated:
+	case <-time.After(defaultTestTimeout):
+		t.Fatalf("Timeout when waiting for child LB to be created")
+	}
+
+	t.Logf("Removing the child balancer with name %q from the group with immediate effect", "child-1")
+	bg.RemoveImmediately("child-1")
+	select {
+	case <-childLBClosed:
+	case <-time.After(defaultTestTimeout):
+		t.Fatalf("Timeout when waiting for child LB to be closed")
+	}
+
+	t.Logf("Removing the child balancer with name %q from the group with caching", "child-2")
+	bg.Remove("child-2")
+	sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
+	defer sCancel()
+	select {
+	case <-childLBClosed:
+		t.Fatalf("Child LB closed when expected to be cached")
+	case <-sCtx.Done():
+	}
+
+	t.Logf("Closing the balancergroup, which should close the second child balancer")
+	bg.Close()
+	select {
+	case <-childLBClosed:
+	case <-time.After(defaultTestTimeout):
+		t.Fatalf("Timeout when waiting for child LB to be closed")
+	}
+}
diff --git a/internal/envconfig/envconfig.go b/internal/envconfig/envconfig.go
index e3adb91..3ae45fa 100644
--- a/internal/envconfig/envconfig.go
+++ b/internal/envconfig/envconfig.go
@@ -119,6 +119,13 @@
 	// A future release will remove this environment variable, enabling strict
 	// path checking behavior unconditionally.
 	DisableStrictPathChecking = boolFromEnv("GRPC_GO_EXPERIMENTAL_DISABLE_STRICT_PATH_CHECKING", false)
+
+	// EnablePriorityLBChildPolicyCache controls whether the priority balancer
+	// should cache child balancers that are removed from the LB policy config,
+	// for a period of 15 minutes. This is disabled by default, but can be
+	// enabled by setting the env variable
+	// GRPC_EXPERIMENTAL_ENABLE_PRIORITY_LB_CHILD_POLICY_CACHE to true.
+	EnablePriorityLBChildPolicyCache = boolFromEnv("GRPC_EXPERIMENTAL_ENABLE_PRIORITY_LB_CHILD_POLICY_CACHE", false)
 )
 
 func boolFromEnv(envVar string, def bool) bool {
diff --git a/internal/xds/balancer/cdsbalancer/configbuilder_childname.go b/internal/xds/balancer/cdsbalancer/configbuilder_childname.go
index 8f118d4..2b0fedd 100644
--- a/internal/xds/balancer/cdsbalancer/configbuilder_childname.go
+++ b/internal/xds/balancer/cdsbalancer/configbuilder_childname.go
@@ -31,9 +31,10 @@
 // struct keeps state between generate() calls, and a later generate() might
 // return names returned by the previous call.
 type nameGenerator struct {
-	existingNames map[clients.Locality]string
-	prefix        uint64
-	nextID        uint64
+	prevLocalitiesToChildNames map[clients.Locality]string // locality to child name mapping generated for the previous update
+	prevChildNames             []string                    // prioritized list of child names generated for the previous update
+	prefix                     uint64
+	nextID                     uint64
 }
 
 func newNameGenerator(prefix uint64) *nameGenerator {
@@ -53,36 +54,58 @@
 // - update 3: [[L1, L2], [L3]] --> ["0", "2"]   (Two priorities were merged)
 // - update 4: [[L1], [L4]] --> ["0", "3",]      (A priority was split, and a new priority was added)
 func (ng *nameGenerator) generate(priorities [][]xdsresource.Locality) []string {
-	var ret []string
+	ret := make([]string, len(priorities))
 	usedNames := make(map[string]bool)
 	newNames := make(map[clients.Locality]string)
-	for _, priority := range priorities {
-		var nameFound string
+
+	// Pass 1: Same priority index match.
+	for i, priority := range priorities {
+		if i >= len(ng.prevChildNames) {
+			continue
+		}
+		targetName := ng.prevChildNames[i]
 		for _, locality := range priority {
-			if name, ok := ng.existingNames[locality.ID]; ok {
+			if name, ok := ng.prevLocalitiesToChildNames[locality.ID]; ok && name == targetName {
+				ret[i] = targetName
+				usedNames[targetName] = true
+				break
+			}
+		}
+	}
+
+	// Pass 2: Greedy reuse.
+	for i, priority := range priorities {
+		if ret[i] != "" {
+			continue
+		}
+		for _, locality := range priority {
+			if name, ok := ng.prevLocalitiesToChildNames[locality.ID]; ok {
 				if !usedNames[name] {
-					nameFound = name
-					// Found a name to use. No need to process the remaining
-					// localities.
+					ret[i] = name
+					usedNames[name] = true
 					break
 				}
 			}
 		}
-
-		if nameFound == "" {
-			// No appropriate used name is found. Make a new name.
-			nameFound = fmt.Sprintf("priority-%d-%d", ng.prefix, ng.nextID)
-			ng.nextID++
-		}
-
-		ret = append(ret, nameFound)
-		// All localities in this priority share the same name. Add them all to
-		// the new map.
-		for _, l := range priority {
-			newNames[l.ID] = nameFound
-		}
-		usedNames[nameFound] = true
 	}
-	ng.existingNames = newNames
+
+	// Pass 3: New name.
+	for i, name := range ret {
+		if name == "" {
+			newID := fmt.Sprintf("priority-%d-%d", ng.prefix, ng.nextID)
+			ng.nextID++
+			ret[i] = newID
+			usedNames[newID] = true
+		}
+	}
+
+	// Update state.
+	for i, priority := range priorities {
+		for _, l := range priority {
+			newNames[l.ID] = ret[i]
+		}
+	}
+	ng.prevLocalitiesToChildNames = newNames
+	ng.prevChildNames = ret
 	return ret
 }
diff --git a/internal/xds/balancer/cdsbalancer/configbuilder_childname_test.go b/internal/xds/balancer/cdsbalancer/configbuilder_childname_test.go
index e534d54..7275f55 100644
--- a/internal/xds/balancer/cdsbalancer/configbuilder_childname_test.go
+++ b/internal/xds/balancer/cdsbalancer/configbuilder_childname_test.go
@@ -29,82 +29,282 @@
 	tests := []struct {
 		name   string
 		prefix uint64
-		input1 [][]xdsresource.Locality
-		input2 [][]xdsresource.Locality
-		want   []string
+		steps  []struct {
+			input [][]xdsresource.Locality
+			want  []string
+		}
 	}{
 		{
 			name:   "init, two new priorities",
 			prefix: 3,
-			input1: nil,
-			input2: [][]xdsresource.Locality{
-				{{ID: clients.Locality{Zone: "L0"}}},
-				{{ID: clients.Locality{Zone: "L1"}}},
+			steps: []struct {
+				input [][]xdsresource.Locality
+				want  []string
+			}{
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}},
+						{{ID: clients.Locality{Zone: "L1"}}},
+					},
+					want: []string{"priority-3-0", "priority-3-1"},
+				},
 			},
-			want: []string{"priority-3-0", "priority-3-1"},
 		},
 		{
 			name:   "one new priority",
 			prefix: 1,
-			input1: [][]xdsresource.Locality{
-				{{ID: clients.Locality{Zone: "L0"}}},
+			steps: []struct {
+				input [][]xdsresource.Locality
+				want  []string
+			}{
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}},
+					},
+					want: []string{"priority-1-0"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}},
+						{{ID: clients.Locality{Zone: "L1"}}},
+					},
+					want: []string{"priority-1-0", "priority-1-1"},
+				},
 			},
-			input2: [][]xdsresource.Locality{
-				{{ID: clients.Locality{Zone: "L0"}}},
-				{{ID: clients.Locality{Zone: "L1"}}},
-			},
-			want: []string{"priority-1-0", "priority-1-1"},
 		},
 		{
 			name:   "merge two priorities",
 			prefix: 4,
-			input1: [][]xdsresource.Locality{
-				{{ID: clients.Locality{Zone: "L0"}}},
-				{{ID: clients.Locality{Zone: "L1"}}},
-				{{ID: clients.Locality{Zone: "L2"}}},
+			steps: []struct {
+				input [][]xdsresource.Locality
+				want  []string
+			}{
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}},
+						{{ID: clients.Locality{Zone: "L1"}}},
+						{{ID: clients.Locality{Zone: "L2"}}},
+					},
+					want: []string{"priority-4-0", "priority-4-1", "priority-4-2"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+						{{ID: clients.Locality{Zone: "L2"}}},
+					},
+					want: []string{"priority-4-0", "priority-4-2"},
+				},
 			},
-			input2: [][]xdsresource.Locality{
-				{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
-				{{ID: clients.Locality{Zone: "L2"}}},
-			},
-			want: []string{"priority-4-0", "priority-4-2"},
 		},
 		{
 			name: "swap two priorities",
-			input1: [][]xdsresource.Locality{
-				{{ID: clients.Locality{Zone: "L0"}}},
-				{{ID: clients.Locality{Zone: "L1"}}},
-				{{ID: clients.Locality{Zone: "L2"}}},
+			steps: []struct {
+				input [][]xdsresource.Locality
+				want  []string
+			}{
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}},
+						{{ID: clients.Locality{Zone: "L1"}}},
+						{{ID: clients.Locality{Zone: "L2"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1", "priority-0-2"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L1"}}},
+						{{ID: clients.Locality{Zone: "L0"}}},
+						{{ID: clients.Locality{Zone: "L2"}}},
+					},
+					want: []string{"priority-0-1", "priority-0-0", "priority-0-2"},
+				},
 			},
-			input2: [][]xdsresource.Locality{
-				{{ID: clients.Locality{Zone: "L1"}}},
-				{{ID: clients.Locality{Zone: "L0"}}},
-				{{ID: clients.Locality{Zone: "L2"}}},
-			},
-			want: []string{"priority-0-1", "priority-0-0", "priority-0-2"},
 		},
 		{
 			name: "split priority",
-			input1: [][]xdsresource.Locality{
-				{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
-				{{ID: clients.Locality{Zone: "L2"}}},
+			steps: []struct {
+				input [][]xdsresource.Locality
+				want  []string
+			}{
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+						{{ID: clients.Locality{Zone: "L2"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}},
+						{{ID: clients.Locality{Zone: "L1"}}},
+						{{ID: clients.Locality{Zone: "L2"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-2", "priority-0-1"},
+				},
 			},
-			input2: [][]xdsresource.Locality{
-				{{ID: clients.Locality{Zone: "L0"}}},
-				{{ID: clients.Locality{Zone: "L1"}}}, // This gets a newly generated name, since "0-0" was already picked.
-				{{ID: clients.Locality{Zone: "L2"}}},
+		},
+		{
+			name: "priority index preference",
+			steps: []struct {
+				input [][]xdsresource.Locality
+				want  []string
+			}{
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}},
+						{{ID: clients.Locality{Zone: "L1"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L1"}}, {ID: clients.Locality{Zone: "L0"}}},
+					},
+					want: []string{"priority-0-0"},
+				},
 			},
-			want: []string{"priority-0-0", "priority-0-2", "priority-0-1"},
+		},
+		{
+			name: "merge partial",
+			steps: []struct {
+				input [][]xdsresource.Locality
+				want  []string
+			}{
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+						{{ID: clients.Locality{Zone: "L2"}}, {ID: clients.Locality{Zone: "L3"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}, {ID: clients.Locality{Zone: "L2"}}},
+						{{ID: clients.Locality{Zone: "L3"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+						{{ID: clients.Locality{Zone: "L2"}}, {ID: clients.Locality{Zone: "L3"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+			},
+		},
+		{
+			name: "swap shift",
+			steps: []struct {
+				input [][]xdsresource.Locality
+				want  []string
+			}{
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+						{{ID: clients.Locality{Zone: "L2"}}, {ID: clients.Locality{Zone: "L3"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L2"}}},
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+					},
+					want: []string{"priority-0-1", "priority-0-0"},
+				},
+			},
+		},
+		{
+			name: "replace priority",
+			steps: []struct {
+				input [][]xdsresource.Locality
+				want  []string
+			}{
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+						{{ID: clients.Locality{Zone: "L2"}}, {ID: clients.Locality{Zone: "L3"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+						{{ID: clients.Locality{Zone: "L5"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-2"},
+				},
+			},
+		},
+		{
+			name: "reordered merge",
+			steps: []struct {
+				input [][]xdsresource.Locality
+				want  []string
+			}{
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L1"}}, {ID: clients.Locality{Zone: "L2"}}},
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L3"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}, {ID: clients.Locality{Zone: "L2"}}},
+						{{ID: clients.Locality{Zone: "L3"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L1"}}, {ID: clients.Locality{Zone: "L2"}}},
+						{{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L3"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+			},
+		},
+		{
+			name: "three-step shift stability",
+			steps: []struct {
+				input [][]xdsresource.Locality
+				want  []string
+			}{
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}},
+						{{ID: clients.Locality{Zone: "L1"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L2"}}},
+						{{ID: clients.Locality{Zone: "L0"}}},
+						{{ID: clients.Locality{Zone: "L1"}}},
+					},
+					want: []string{"priority-0-2", "priority-0-0", "priority-0-1"},
+				},
+				{
+					input: [][]xdsresource.Locality{
+						{{ID: clients.Locality{Zone: "L0"}}},
+						{{ID: clients.Locality{Zone: "L1"}}},
+					},
+					want: []string{"priority-0-0", "priority-0-1"},
+				},
+			},
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			ng := newNameGenerator(tt.prefix)
-			got1 := ng.generate(tt.input1)
-			t.Logf("%v", got1)
-			got := ng.generate(tt.input2)
-			if diff := cmp.Diff(got, tt.want); diff != "" {
-				t.Errorf("generate() = got: %v, want: %v, diff (-got +want): %s", got, tt.want, diff)
+			for i, step := range tt.steps {
+				got := ng.generate(step.input)
+				if diff := cmp.Diff(got, step.want); diff != "" {
+					t.Errorf("step %d: generate() = got: %v, want: %v, diff (-got +want): %s", i, got, step.want, diff)
+				}
 			}
 		})
 	}
diff --git a/internal/xds/balancer/cdsbalancer/e2e_test/aggregate_cluster_test.go b/internal/xds/balancer/cdsbalancer/e2e_test/aggregate_cluster_test.go
index 76de30a..8b75dbe 100644
--- a/internal/xds/balancer/cdsbalancer/e2e_test/aggregate_cluster_test.go
+++ b/internal/xds/balancer/cdsbalancer/e2e_test/aggregate_cluster_test.go
@@ -23,6 +23,7 @@
 	"slices"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -697,8 +698,8 @@
 	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
 
 	// Start two test backends.
-	servers, cleanup3 := startTestServiceBackends(t, 2)
-	defer cleanup3()
+	servers, cleanup := startTestServiceBackends(t, 2)
+	defer cleanup()
 	addrs, _ := backendAddressesAndPorts(t, servers)
 
 	// Configure an aggregate cluster pointing to an EDS and LOGICAL_DNS
@@ -795,8 +796,28 @@
 // the DNS resolver pushes an update, the test verifies that we switch to the
 // DNS cluster and can make a successful RPC.
 func (s) TestAggregateCluster_BadEDSFromError_GoodToBadDNS(t *testing.T) {
-	// Start an xDS management server.
-	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	// Start an xDS management server and block it from continuously re-sending
+	// the same EDS response that triggers a NACK. This is to ensure that the
+	// test can proceed to the point of verifying that we fall back to the
+	// LOGICAL_DNS cluster. More more details see:
+	// https://github.com/grpc/grpc-go/issues/8994.
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	edsResponseSeen := atomic.Bool{}
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		OnStreamResponse: func(_ context.Context, _ int64, _ *v3discoverypb.DiscoveryRequest, resp *v3discoverypb.DiscoveryResponse) {
+			if resp.GetTypeUrl() != version.V3EndpointsURL {
+				return
+			}
+			if edsResponseSeen.Load() {
+				t.Logf("Received EDS response again, blocking to prevent test from busylooping on the same EDS response: %v", resp)
+				<-ctx.Done()
+				return
+			}
+			edsResponseSeen.Store(true)
+		},
+		AllowResourceSubset: true,
+	})
 
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
@@ -834,8 +855,6 @@
 		Endpoints:      []*v3endpointpb.ClusterLoadAssignment{nackEndpointResource},
 		SkipValidation: true,
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	if err := managementServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
@@ -1093,8 +1112,29 @@
 // the LOGICAL_DNS cluster, because it is supposed to treat the bad EDS response
 // as though it received an update with no endpoints.
 func (s) TestAggregateCluster_Fallback_EDSNackedWithoutPreviousGoodUpdate(t *testing.T) {
-	// Start an xDS management server.
-	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	// Start an xDS management server and block it from continuously re-sending
+	// the same EDS response that triggers a NACK. This is to ensure that the
+	// test can proceed to the point of verifying that we fall back to the
+	// LOGICAL_DNS cluster. More more details see:
+	// https://github.com/grpc/grpc-go/issues/8994.
+	edsResponseSeen := atomic.Bool{}
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		OnStreamResponse: func(_ context.Context, _ int64, _ *v3discoverypb.DiscoveryRequest, resp *v3discoverypb.DiscoveryResponse) {
+			if resp.GetTypeUrl() != version.V3EndpointsURL {
+				return
+			}
+			if edsResponseSeen.Load() {
+				t.Logf("Received EDS response again, blocking to prevent test from busylooping on the same EDS response: %v", resp)
+				<-ctx.Done()
+				return
+			}
+			edsResponseSeen.Store(true)
+		},
+		AllowResourceSubset: true,
+	})
 
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
@@ -1103,8 +1143,8 @@
 	// Start two test backends and extract their host and port. The first
 	// backend is used for the EDS cluster and the second backend is used for
 	// the LOGICAL_DNS cluster.
-	servers, cleanup3 := startTestServiceBackends(t, 2)
-	defer cleanup3()
+	servers, cleanup := startTestServiceBackends(t, 2)
+	defer cleanup()
 	addrs, ports := backendAddressesAndPorts(t, servers)
 	dnsHostName, dnsPort := hostAndPortFromAddress(t, servers[1].Address)
 
@@ -1128,11 +1168,7 @@
 
 	// Set a load balancing weight of 0 for the backend in the EDS resource.
 	// This is expected to be NACKed by the xDS client. Since the
-	// cluster_resolver LB policy has no previously received good EDS resource,
-	// it will treat this as though it received an update with no endpoints.
 	resources.Endpoints[0].Endpoints[0].LbEndpoints[0].LoadBalancingWeight = &wrapperspb.UInt32Value{Value: 0}
-	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
-	defer cancel()
 	if err := managementServer.Update(ctx, resources); err != nil {
 		t.Fatal(err)
 	}
diff --git a/internal/xds/balancer/cdsbalancer/e2e_test/eds_impl_test.go b/internal/xds/balancer/cdsbalancer/e2e_test/eds_impl_test.go
index a92a2ac..ee45ca3 100644
--- a/internal/xds/balancer/cdsbalancer/e2e_test/eds_impl_test.go
+++ b/internal/xds/balancer/cdsbalancer/e2e_test/eds_impl_test.go
@@ -236,8 +236,8 @@
 	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
 
 	// Start backend servers which provide an implementation of the TestService.
-	servers, cleanup2 := startTestServiceBackends(t, 4)
-	defer cleanup2()
+	servers, cleanup := startTestServiceBackends(t, 4)
+	defer cleanup()
 	addrs, ports := backendAddressesAndPorts(t, servers)
 
 	// Create xDS resources for consumption by the test. We start off with two
diff --git a/internal/xds/balancer/clustermanager/clustermanager_test.go b/internal/xds/balancer/clustermanager/clustermanager_test.go
index 17a3026..576f440 100644
--- a/internal/xds/balancer/clustermanager/clustermanager_test.go
+++ b/internal/xds/balancer/clustermanager/clustermanager_test.go
@@ -513,7 +513,7 @@
 	}
 }
 
-const initIdleBalancerName = "test-init-Idle-balancer"
+const initIdleBalancerName = "test-init-idle-balancer"
 
 var errTestInitIdle = fmt.Errorf("init Idle balancer error 0")
 
@@ -551,7 +551,7 @@
 
 	configJSON1 := `{
 "children": {
-	"cds:cluster_1":{ "childPolicy": [{"test-init-Idle-balancer":""}] }
+	"cds:cluster_1":{ "childPolicy": [{"test-init-idle-balancer":""}] }
 }
 }`
 	config1, err := parser.ParseConfig([]byte(configJSON1))
diff --git a/internal/xds/balancer/priority/balancer.go b/internal/xds/balancer/priority/balancer.go
index 950cd13..e842920 100644
--- a/internal/xds/balancer/priority/balancer.go
+++ b/internal/xds/balancer/priority/balancer.go
@@ -34,6 +34,7 @@
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/internal/balancergroup"
 	"google.golang.org/grpc/internal/buffer"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/grpclog"
 	"google.golang.org/grpc/internal/grpcsync"
 	"google.golang.org/grpc/internal/hierarchy"
@@ -153,7 +154,7 @@
 		// The balancing policy name is changed, close the old child. But don't
 		// rebuild, rebuild will happen when syncing priorities.
 		if currentChild.balancerName != bb.Name() {
-			currentChild.stop()
+			currentChild.stop(true)
 			currentChild.updateBalancerName(bb.Name())
 		}
 
@@ -169,7 +170,7 @@
 	// Cleanup resources used by children removed from the config.
 	for name, oldChild := range b.children {
 		if _, ok := newConfig.Children[name]; !ok {
-			oldChild.stop()
+			oldChild.stop(!envconfig.EnablePriorityLBChildPolicyCache)
 			delete(b.children, name)
 		}
 	}
@@ -230,7 +231,7 @@
 	// Stop the child policies, this is necessary to stop the init timers in the
 	// children.
 	for _, child := range b.children {
-		child.stop()
+		child.stop(true)
 	}
 }
 
diff --git a/internal/xds/balancer/priority/balancer_child.go b/internal/xds/balancer/priority/balancer_child.go
index 5e76391..ad5b5a8 100644
--- a/internal/xds/balancer/priority/balancer_child.go
+++ b/internal/xds/balancer/priority/balancer_child.go
@@ -133,12 +133,16 @@
 // It doesn't do it directly. It asks the balancer group to remove it.
 //
 // Note that the underlying balancer group could keep the child in a cache.
-func (cb *childBalancer) stop() {
+func (cb *childBalancer) stop(immediate bool) {
 	if !cb.started {
 		return
 	}
 	cb.stopInitTimer()
-	cb.parent.bg.Remove(cb.name)
+	if immediate {
+		cb.parent.bg.RemoveImmediately(cb.name)
+	} else {
+		cb.parent.bg.Remove(cb.name)
+	}
 	cb.started = false
 	cb.state = balancer.State{
 		ConnectivityState: connectivity.Connecting,
diff --git a/internal/xds/balancer/priority/balancer_priority.go b/internal/xds/balancer/priority/balancer_priority.go
index a57df7d..3181ce2 100644
--- a/internal/xds/balancer/priority/balancer_priority.go
+++ b/internal/xds/balancer/priority/balancer_priority.go
@@ -128,7 +128,7 @@
 			b.logger.Warningf("Priority name %q is not found in list of child policies", name)
 			continue
 		}
-		child.stop()
+		child.stop(false)
 	}
 }
 
diff --git a/internal/xds/balancer/priority/balancer_test.go b/internal/xds/balancer/priority/balancer_test.go
index 070e95c..c1c1572 100644
--- a/internal/xds/balancer/priority/balancer_test.go
+++ b/internal/xds/balancer/priority/balancer_test.go
@@ -29,6 +29,7 @@
 	"google.golang.org/grpc/balancer/roundrobin"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/internal/balancer/stub"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/hierarchy"
 	internalserviceconfig "google.golang.org/grpc/internal/serviceconfig"
@@ -68,9 +69,6 @@
 	for i := 0; i < testBackendAddrsCount; i++ {
 		testBackendAddrStrs = append(testBackendAddrStrs, fmt.Sprintf("%d.%d.%d.%d:%d", i, i, i, i, i))
 	}
-	// Disable sub-balancer caching for all but the tests which exercise the
-	// caching behavior.
-	DefaultSubBalancerCloseTimeout = time.Duration(0)
 	balancer.Register(&anotherRR{Builder: balancer.Get(roundrobin.Name)})
 }
 
@@ -85,6 +83,17 @@
 //
 // Init 0 and 1; 0 is up, use 0; add 2, use 0; remove 2, use 0.
 func (s) TestPriority_HighPriorityReady(t *testing.T) {
+	t.Run("caching_enabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+		testPriorityHighPriorityReady(t)
+	})
+	t.Run("caching_disabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+		testPriorityHighPriorityReady(t)
+	})
+}
+
+func testPriorityHighPriorityReady(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
@@ -199,6 +208,17 @@
 // Init 0 and 1; 0 is up, use 0; 0 is down, 1 is up, use 1; add 2, use 1; 1 is
 // down, use 2; remove 2, use 1.
 func (s) TestPriority_SwitchPriority(t *testing.T) {
+	t.Run("caching_enabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+		testPrioritySwitchPriority(t)
+	})
+	t.Run("caching_disabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+		testPrioritySwitchPriority(t)
+	})
+}
+
+func testPrioritySwitchPriority(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
@@ -340,16 +360,19 @@
 		t.Fatalf("failed to update ClientConn state: %v", err)
 	}
 
-	// p2 SubConns are shut down.
-	scToShutdown := <-cc.ShutdownSubConnCh
-	// The same SubConn is closed by gracefulswitch and pickfirstleaf when they
-	// are closed. Remove duplicate events.
-	// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
-	// workaround once pickfirst is the only leaf policy and responsible for
-	// shutting down SubConns.
-	<-cc.ShutdownSubConnCh
-	if scToShutdown != sc2 {
-		t.Fatalf("ShutdownSubConn, want %v, got %v", sc2, scToShutdown)
+	// p2 will be closed immediately only if the cache is not enabled.
+	if !envconfig.EnablePriorityLBChildPolicyCache {
+		// p2 SubConns are shut down.
+		scToShutdown := <-cc.ShutdownSubConnCh
+		// The same SubConn is closed by gracefulswitch and pickfirstleaf when
+		// they are closed. Remove duplicate events.
+		// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+		// workaround once pickfirst is the only leaf policy and responsible for
+		// shutting down SubConns.
+		<-cc.ShutdownSubConnCh
+		if scToShutdown != sc2 {
+			t.Fatalf("ShutdownSubConn, want %v, got %v", sc2, scToShutdown)
+		}
 	}
 
 	// Should get an update with 1's old transient failure picker, to override
@@ -374,6 +397,10 @@
 //
 // Init 0 and 1; 0 is up, use 0; 0 is connecting, 1 is up, use 1; 0 is ready,
 // use 0.
+//
+// Env var GRPC_EXPERIMENTAL_ENABLE_PRIORITY_LB_CHILD_POLICY_CACHE does not
+// affect whether a lower priority child policy is cached when it is removed
+// from the balancer group.
 func (s) TestPriority_HighPriorityToConnectingFromReady(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
@@ -383,7 +410,7 @@
 	pb := bb.Build(cc, balancer.BuildOptions{})
 	defer pb.Close()
 
-	// Two localities, with priorities [0, 1], each with one backend.
+	t.Log("Two localities, with priorities [0, 1], each with one backend.")
 	if err := pb.UpdateClientConnState(balancer.ClientConnState{
 		ResolverState: resolver.State{
 			Endpoints: []resolver.Endpoint{
@@ -408,7 +435,7 @@
 	}
 	sc0 := <-cc.NewSubConnCh
 
-	// p0 is ready.
+	t.Log("Make p0 ready.")
 	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
 	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
 
@@ -417,16 +444,15 @@
 		t.Fatal(err.Error())
 	}
 
-	// Turn 0 to TransientFailure, will start and use 1.
+	t.Log("Turn down 0, will start and use 1.")
 	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
-
 	// Before 1 gets READY, picker should return NoSubConnAvailable, so RPCs
 	// will retry.
 	if err := cc.WaitForPickerWithErr(ctx, balancer.ErrNoSubConnAvailable); err != nil {
 		t.Fatal(err.Error())
 	}
 
-	// Handle SubConn creation from 1.
+	t.Log("Handle SubConn creation from 1.")
 	addrs1 := <-cc.NewSubConnAddrsCh
 	if got, want := addrs1[0].Addr, testBackendAddrStrs[1]; got != want {
 		t.Fatalf("sc is created with addr %v, want %v", got, want)
@@ -440,22 +466,15 @@
 		t.Fatal(err.Error())
 	}
 
-	// Turn 0 back to Ready.
+	t.Logf("Turn 0 back to Ready.")
 	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
 	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
 
-	// p1 subconn should be shut down.
-	scToShutdown := <-cc.ShutdownSubConnCh
-	// The same SubConn is closed by gracefulswitch and pickfirstleaf when they
-	// are closed. Remove duplicate events.
-	// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
-	// workaround once pickfirst is the only leaf policy and responsible for
-	// shutting down SubConns.
-	<-cc.ShutdownSubConnCh
-	if scToShutdown != sc1 {
-		t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
-	}
-
+	// At this point p1 is removed from the balancer group, but it gets cached.
+	// Therefore its subchannels will not be removed immediately. So, we do not
+	// check for the shutdown of p1's subchannels here. Instead, we check that
+	// p0 is used again after it becomes ready, which indicates that the
+	// priority switch is successful.
 	if err := cc.WaitForRoundRobinPicker(ctx, sc0); err != nil {
 		t.Fatal(err.Error())
 	}
@@ -646,25 +665,6 @@
 	// When 0 becomes ready, 0 should be used, 1 and 2 should all be closed.
 	sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
 
-	// sc1 and sc2 should be shut down.
-	//
-	// With localities caching, the lower priorities are closed after a timeout,
-	// in goroutines. The order is no longer guaranteed.
-	// The same SubConn is closed by gracefulswitch and pickfirstleaf when they
-	// are closed. Remove duplicate events.
-	// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
-	// workaround once pickfirst is the only leaf policy and responsible for
-	// shutting down SubConns.
-	scToShutdown := [2]balancer.SubConn{}
-	scToShutdown[0] = <-cc.ShutdownSubConnCh
-	<-cc.ShutdownSubConnCh
-	scToShutdown[1] = <-cc.ShutdownSubConnCh
-	<-cc.ShutdownSubConnCh
-
-	if !(scToShutdown[0] == sc1 && scToShutdown[1] == sc2) && !(scToShutdown[0] == sc2 && scToShutdown[1] == sc1) {
-		t.Errorf("ShutdownSubConn, want [%v, %v], got %v", sc1, sc2, scToShutdown)
-	}
-
 	// Test pick with 0.
 	if err := cc.WaitForRoundRobinPicker(ctx, sc0); err != nil {
 		t.Fatal(err.Error())
@@ -741,6 +741,17 @@
 
 // EDS removes all priorities, and re-adds them.
 func (s) TestPriority_RemovesAllPriorities(t *testing.T) {
+	t.Run("caching_enabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+		testPriorityRemovesAllPriorities(t)
+	})
+	t.Run("caching_disabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+		testPriorityRemovesAllPriorities(t)
+	})
+}
+
+func testPriorityRemovesAllPriorities(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
@@ -797,16 +808,19 @@
 		t.Fatalf("failed to update ClientConn state: %v", err)
 	}
 
-	// p0 subconn should be shut down.
-	scToShutdown := <-cc.ShutdownSubConnCh
-	// The same SubConn is closed by gracefulswitch and pickfirstleaf when they
-	// are closed. Remove duplicate events.
-	// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
-	// workaround once pickfirst is the only leaf policy and responsible for
-	// shutting down SubConns.
-	<-cc.ShutdownSubConnCh
-	if scToShutdown != sc0 {
-		t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
+	// p0 will be closed immediately only if the cache is not enabled.
+	if !envconfig.EnablePriorityLBChildPolicyCache {
+		// p0 SubConns are shut down.
+		scToShutdown := <-cc.ShutdownSubConnCh
+		// The same SubConn is closed by gracefulswitch and pickfirstleaf when
+		// they are closed. Remove duplicate events.
+		// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+		// workaround once pickfirst is the only leaf policy and responsible for
+		// shutting down SubConns.
+		<-cc.ShutdownSubConnCh
+		if scToShutdown != sc0 {
+			t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
+		}
 	}
 
 	// Test pick return TransientFailure.
@@ -839,6 +853,14 @@
 	}
 	sc01 := <-cc.NewSubConnCh
 
+	if envconfig.EnablePriorityLBChildPolicyCache {
+		// The old p0 SubConn is shut down.
+		scToShutdown := <-cc.ShutdownSubConnCh
+		if scToShutdown != sc0 {
+			t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
+		}
+	}
+
 	// Don't send any update to p0, so to not override the old state of p0.
 	// Later, connect to p1 and then remove p1. This will fallback to p0, and
 	// will send p0's old picker if they are not correctly removed.
@@ -874,16 +896,19 @@
 		t.Fatalf("failed to update ClientConn state: %v", err)
 	}
 
-	// p1 subconn should be shut down.
-	scToShutdown1 := <-cc.ShutdownSubConnCh
-	// The same SubConn is closed by gracefulswitch and pickfirstleaf when they
-	// are closed. Remove duplicate events.
-	// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
-	// workaround once pickfirst is the only leaf policy and responsible for
-	// shutting down SubConns.
-	<-cc.ShutdownSubConnCh
-	if scToShutdown1 != sc11 {
-		t.Fatalf("ShutdownSubConn, want %v, got %v", sc11, scToShutdown1)
+	// p1 will be closed immediately only if the cache is not enabled.
+	if !envconfig.EnablePriorityLBChildPolicyCache {
+		// p1 SubConns are shut down.
+		scToShutdown := <-cc.ShutdownSubConnCh
+		// The same SubConn is closed by gracefulswitch and pickfirstleaf when
+		// they are closed. Remove duplicate events.
+		// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+		// workaround once pickfirst is the only leaf policy and responsible for
+		// shutting down SubConns.
+		<-cc.ShutdownSubConnCh
+		if scToShutdown != sc11 {
+			t.Fatalf("ShutdownSubConn, want %v, got %v", sc11, scToShutdown)
+		}
 	}
 
 	// Test pick return NoSubConn.
@@ -1004,8 +1029,19 @@
 
 // Test the case where the first and only priority is removed.
 func (s) TestPriority_FirstPriorityUnavailable(t *testing.T) {
-	const testPriorityInitTimeout = 200 * time.Millisecond
-	overrideInitTimeout(t, testPriorityInitTimeout)
+	t.Run("caching_enabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+		testPriorityFirstPriorityUnavailable(t)
+	})
+	t.Run("caching_disabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+		testPriorityFirstPriorityUnavailable(t)
+	})
+}
+
+func testPriorityFirstPriorityUnavailable(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
 
 	cc := testutils.NewBalancerClientConn(t)
 	bb := balancer.Get(Name)
@@ -1029,6 +1065,12 @@
 		t.Fatalf("failed to update ClientConn state: %v", err)
 	}
 
+	addrs0 := <-cc.NewSubConnAddrsCh
+	if got, want := addrs0[0].Addr, testBackendAddrStrs[0]; got != want {
+		t.Fatalf("sc is created with addr %v, want %v", got, want)
+	}
+	sc0 := <-cc.NewSubConnCh
+
 	// Remove the only localities.
 	if err := pb.UpdateClientConnState(balancer.ClientConnState{
 		ResolverState: resolver.State{
@@ -1042,14 +1084,42 @@
 		t.Fatalf("failed to update ClientConn state: %v", err)
 	}
 
-	// Wait after double the init timer timeout, to ensure it doesn't panic.
-	time.Sleep(testPriorityInitTimeout * 2)
+	// p0 will be closed immediately only if the cache is not enabled.
+	if !envconfig.EnablePriorityLBChildPolicyCache {
+		// p0 SubConns are shut down.
+		scToShutdown := <-cc.ShutdownSubConnCh
+		// The same SubConn is closed by gracefulswitch and pickfirstleaf when
+		// they are closed. Remove duplicate events.
+		// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+		// workaround once pickfirst is the only leaf policy and responsible for
+		// shutting down SubConns.
+		<-cc.ShutdownSubConnCh
+		if scToShutdown != sc0 {
+			t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
+		}
+	}
+
+	// Test pick return TransientFailure.
+	if err := cc.WaitForPickerWithErr(ctx, ErrAllPrioritiesRemoved); err != nil {
+		t.Fatal(err.Error())
+	}
 }
 
 // When a child is moved from low priority to high.
 //
 // Init a(p0) and b(p1); a(p0) is up, use a; move b to p0, a to p1, use b.
 func (s) TestPriority_MoveChildToHigherPriority(t *testing.T) {
+	t.Run("caching_enabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+		testPriorityMoveChildToHigherPriority(t)
+	})
+	t.Run("caching_disabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+		testPriorityMoveChildToHigherPriority(t)
+	})
+}
+
+func testPriorityMoveChildToHigherPriority(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
@@ -1120,18 +1190,6 @@
 		t.Fatal(err.Error())
 	}
 
-	// Old subconn should be shut down.
-	scToShutdown := <-cc.ShutdownSubConnCh
-	// The same SubConn is closed by gracefulswitch and pickfirstleaf when they
-	// are closed. Remove duplicate events.
-	// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
-	// workaround once pickfirst is the only leaf policy and responsible for
-	// shutting down SubConns.
-	<-cc.ShutdownSubConnCh
-	if scToShutdown != sc1 {
-		t.Fatalf("ShutdownSubConn, want %v, got %v", sc1, scToShutdown)
-	}
-
 	addrs2 := <-cc.NewSubConnAddrsCh
 	if got, want := addrs2[0].Addr, testBackendAddrStrs[1]; got != want {
 		t.Fatalf("sc is created with addr %v, want %v", got, want)
@@ -1153,6 +1211,17 @@
 //
 // Init a(p0) and b(p1); a(p0) is down, use b; move b to p0, a to p1, use b.
 func (s) TestPriority_MoveReadyChildToHigherPriority(t *testing.T) {
+	t.Run("caching_enabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+		testPriorityMoveReadyChildToHigherPriority(t)
+	})
+	t.Run("caching_disabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+		testPriorityMoveReadyChildToHigherPriority(t)
+	})
+}
+
+func testPriorityMoveReadyChildToHigherPriority(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
@@ -1227,18 +1296,6 @@
 		t.Fatalf("failed to update ClientConn state: %v", err)
 	}
 
-	// Old subconn from child-0 should be removed.
-	scToShutdown := <-cc.ShutdownSubConnCh
-	// The same SubConn is closed by gracefulswitch and pickfirstleaf when they
-	// are closed. Remove duplicate events.
-	// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
-	// workaround once pickfirst is the only leaf policy and responsible for
-	// shutting down SubConns.
-	<-cc.ShutdownSubConnCh
-	if scToShutdown != sc0 {
-		t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
-	}
-
 	// Because this was a ready child moved to a higher priority, no new subconn
 	// or picker should be updated.
 	select {
@@ -1255,6 +1312,17 @@
 //
 // Init a(p0) and b(p1); a(p0) is down, use b; move b to p0, a to p1, use b.
 func (s) TestPriority_RemoveReadyLowestChild(t *testing.T) {
+	t.Run("caching_enabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+		testPriorityRemoveReadyLowestChild(t)
+	})
+	t.Run("caching_disabled", func(t *testing.T) {
+		testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+		testPriorityRemoveReadyLowestChild(t)
+	})
+}
+
+func testPriorityRemoveReadyLowestChild(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
@@ -1329,16 +1397,18 @@
 		t.Fatalf("failed to update ClientConn state: %v", err)
 	}
 
-	// Old subconn from child-1 should be shut down.
-	scToShutdown := <-cc.ShutdownSubConnCh
-	// The same SubConn is closed by gracefulswitch and pickfirstleaf when they
-	// are closed. Remove duplicate events.
-	// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
-	// workaround once pickfirst is the only leaf policy and responsible for
-	// shutting down SubConns.
-	<-cc.ShutdownSubConnCh
-	if scToShutdown != sc1 {
-		t.Fatalf("ShutdownSubConn, want %v, got %v", sc1, scToShutdown)
+	if !envconfig.EnablePriorityLBChildPolicyCache {
+		// Old subconn from child-1 should be shut down.
+		scToShutdown := <-cc.ShutdownSubConnCh
+		// The same SubConn is closed by gracefulswitch and pickfirstleaf when they
+		// are closed. Remove duplicate events.
+		// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+		// workaround once pickfirst is the only leaf policy and responsible for
+		// shutting down SubConns.
+		<-cc.ShutdownSubConnCh
+		if scToShutdown != sc1 {
+			t.Fatalf("ShutdownSubConn, want %v, got %v", sc1, scToShutdown)
+		}
 	}
 
 	if err := cc.WaitForErrPicker(ctx); err != nil {
@@ -1358,19 +1428,11 @@
 //
 // Init 0; 0 is up, use 0; remove 0, only picker is updated, no subconn is
 // removed; re-add 0, picker is updated.
-func (s) TestPriority_ReadyChildRemovedButInCache(t *testing.T) {
+func (s) TestPriority_RemoveOnlyReadyChild_CachingEnabled(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
-	const testChildCacheTimeout = time.Second
-	defer func() func() {
-		old := DefaultSubBalancerCloseTimeout
-		DefaultSubBalancerCloseTimeout = testChildCacheTimeout
-		return func() {
-			DefaultSubBalancerCloseTimeout = old
-		}
-	}()()
-
 	cc := testutils.NewBalancerClientConn(t)
 	bb := balancer.Get(Name)
 	pb := bb.Build(cc, balancer.BuildOptions{})
@@ -1408,8 +1470,7 @@
 		t.Fatal(err.Error())
 	}
 
-	// Remove the child, it shouldn't cause any conn changed, but picker should
-	// be different.
+	// Remove the child.
 	if err := pb.UpdateClientConnState(balancer.ClientConnState{
 		ResolverState:  resolver.State{},
 		BalancerConfig: &LBConfig{},
@@ -1430,7 +1491,7 @@
 	case <-time.After(time.Millisecond * 100):
 	}
 
-	// Re-add the child, shouldn't create new connections.
+	// Re-add the child.
 	if err := pb.UpdateClientConnState(balancer.ClientConnState{
 		ResolverState: resolver.State{
 			Endpoints: []resolver.Endpoint{
@@ -1463,6 +1524,109 @@
 	}
 }
 
+// When the only ready child is removed, it is deleted and subchannels are shut
+// down.  Re-adding results in subchannels being recreated.
+//
+// Init 0; 0 is up, use 0; remove 0, re-add 0.
+func (s) TestPriority_RemoveOnlyReadyChild_CachingDisabled(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	cc := testutils.NewBalancerClientConn(t)
+	bb := balancer.Get(Name)
+	pb := bb.Build(cc, balancer.BuildOptions{})
+	defer pb.Close()
+
+	// One children, with priorities [0], with one backend.
+	if err := pb.UpdateClientConnState(balancer.ClientConnState{
+		ResolverState: resolver.State{
+			Endpoints: []resolver.Endpoint{
+				hierarchy.SetInEndpoint(resolver.Endpoint{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[0]}}}, []string{"child-0"}),
+			},
+		},
+		BalancerConfig: &LBConfig{
+			Children: map[string]*Child{
+				"child-0": {Config: &internalserviceconfig.BalancerConfig{Name: roundrobin.Name}},
+			},
+			Priorities: []string{"child-0"},
+		},
+	}); err != nil {
+		t.Fatalf("failed to update ClientConn state: %v", err)
+	}
+
+	addrs1 := <-cc.NewSubConnAddrsCh
+	if got, want := addrs1[0].Addr, testBackendAddrStrs[0]; got != want {
+		t.Fatalf("sc is created with addr %v, want %v", got, want)
+	}
+	sc1 := <-cc.NewSubConnCh
+
+	// p0 is ready.
+	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+
+	// Test roundrobin with only p0 subconns.
+	if err := cc.WaitForRoundRobinPicker(ctx, sc1); err != nil {
+		t.Fatal(err.Error())
+	}
+
+	// Remove the child.
+	if err := pb.UpdateClientConnState(balancer.ClientConnState{
+		ResolverState:  resolver.State{},
+		BalancerConfig: &LBConfig{},
+	}); err != nil {
+		t.Fatalf("failed to update ClientConn state: %v", err)
+	}
+
+	// p0 SubConns are shut down.
+	scToShutdown := <-cc.ShutdownSubConnCh
+	// The same SubConn is closed by gracefulswitch and pickfirstleaf when
+	// they are closed. Remove duplicate events.
+	// TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+	// workaround once pickfirst is the only leaf policy and responsible for
+	// shutting down SubConns.
+	<-cc.ShutdownSubConnCh
+	if scToShutdown != sc1 {
+		t.Fatalf("ShutdownSubConn, want %v, got %v", sc1, scToShutdown)
+	}
+
+	if err := cc.WaitForPickerWithErr(ctx, ErrAllPrioritiesRemoved); err != nil {
+		t.Fatal(err.Error())
+	}
+
+	// Re-add the child.
+	if err := pb.UpdateClientConnState(balancer.ClientConnState{
+		ResolverState: resolver.State{
+			Endpoints: []resolver.Endpoint{
+				hierarchy.SetInEndpoint(resolver.Endpoint{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[0]}}}, []string{"child-0"}),
+			},
+		},
+		BalancerConfig: &LBConfig{
+			Children: map[string]*Child{
+				"child-0": {Config: &internalserviceconfig.BalancerConfig{Name: roundrobin.Name}},
+			},
+			Priorities: []string{"child-0"},
+		},
+	}); err != nil {
+		t.Fatalf("failed to update ClientConn state: %v", err)
+	}
+
+	addrs1 = <-cc.NewSubConnAddrsCh
+	if got, want := addrs1[0].Addr, testBackendAddrStrs[0]; got != want {
+		t.Fatalf("sc is created with addr %v, want %v", got, want)
+	}
+	sc1 = <-cc.NewSubConnCh
+
+	// p0 is ready.
+	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+	sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+
+	// Test roundrobin with only p0 subconns.
+	if err := cc.WaitForRoundRobinPicker(ctx, sc1); err != nil {
+		t.Fatal(err.Error())
+	}
+}
+
 // When the policy of a child is changed.
 //
 // Init 0; 0 is up, use 0; change 0's policy, 0 is used.
@@ -1691,7 +1855,6 @@
 	case <-time.After(time.Second):
 		t.Fatalf("timeout waiting for ResolveNow()")
 	}
-
 }
 
 // TestPriority_IgnoreReresolutionRequestTwoChildren tests the case where the
@@ -1791,7 +1954,7 @@
 	}
 }
 
-const initIdleBalancerName = "test-init-Idle-balancer"
+const initIdleBalancerName = "test-init-idle-balancer"
 
 var errsTestInitIdle = []error{
 	fmt.Errorf("init Idle balancer error 0"),