clusterimpl: update picker synchronously on config update (#7652)

xds/internal/balancer/clusterimpl/balancer_test.go

@@ -943,6 +943,74 @@ func (s) TestFailedToParseChildPolicyConfig(t *testing.T) {
 	}
 }
 
+// Test verify that the case picker is updated synchronously on receipt of
+// configuration update.
+func (s) TestPickerUpdatedSynchronouslyOnConfigUpdate(t *testing.T) {
+	// Override the pickerUpdateHook to be notified that picker was updated.
+	pickerUpdated := make(chan struct{}, 1)
+	origNewPickerUpdated := pickerUpdateHook
+	pickerUpdateHook = func() {
+		pickerUpdated <- struct{}{}
+	}
+	defer func() { pickerUpdateHook = origNewPickerUpdated }()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	// Override the clientConnUpdateHook to ensure client conn was updated.
+	clientConnUpdateDone := make(chan struct{}, 1)
+	origClientConnUpdateHook := clientConnUpdateHook
+	clientConnUpdateHook = func() {
+		// Verify that picker was updated before the completion of
+		// client conn update.
+		select {
+		case <-pickerUpdated:
+		case <-ctx.Done():
+			t.Fatal("Client conn update completed before picker update.")
+		}
+		clientConnUpdateDone <- struct{}{}
+	}
+	defer func() { clientConnUpdateHook = origClientConnUpdateHook }()
+
+	defer xdsclient.ClearCounterForTesting(testClusterName, testServiceName)
+	xdsC := fakeclient.NewClient()
+
+	builder := balancer.Get(Name)
+	cc := testutils.NewBalancerClientConn(t)
+	b := builder.Build(cc, balancer.BuildOptions{})
+	defer b.Close()
+
+	// Create a stub balancer which waits for the cluster_impl policy to be
+	// closed before sending a picker update (upon receipt of a resolver
+	// update).
+	stub.Register(t.Name(), stub.BalancerFuncs{
+		UpdateClientConnState: func(bd *stub.BalancerData, _ balancer.ClientConnState) error {
+			bd.ClientConn.UpdateState(balancer.State{
+				Picker: base.NewErrPicker(errors.New("dummy error picker")),
+			})
+			return nil
+		},
+	})
+
+	if err := b.UpdateClientConnState(balancer.ClientConnState{
+		ResolverState: xdsclient.SetClient(resolver.State{Addresses: testBackendAddrs}, xdsC),
+		BalancerConfig: &LBConfig{
+			Cluster:        testClusterName,
+			EDSServiceName: testServiceName,
+			ChildPolicy: &internalserviceconfig.BalancerConfig{
+				Name: t.Name(),
+			},
+		},
+	}); err != nil {
+		t.Fatalf("Unexpected error from UpdateClientConnState: %v", err)
+	}
+
+	select {
+	case <-clientConnUpdateDone:
+	case <-ctx.Done():
+		t.Fatal("Timed out waiting for client conn update to be completed.")
+	}
+}
+
 func assertString(f func() (string, error)) string {
 	s, err := f()
 	if err != nil {