func (s) TestParseConfig(t *testing.T) {
	bb := balancer.Get(cdsName)
	if bb == nil {
		t.Fatalf("balancer.Get(%q) returned nil", cdsName)
	}
	parser, ok := bb.(balancer.ConfigParser)
	if !ok {
		t.Fatalf("balancer %q does not implement the ConfigParser interface", cdsName)
	}
	tests := []struct {
		name    string
		input   json.RawMessage
		wantCfg serviceconfig.LoadBalancingConfig
		wantErr bool
	}{
		{
			name:    "good-lb-config",
			input:   json.RawMessage(`{"Cluster": "cluster1"}`),
			wantCfg: &lbConfig{ClusterName: "cluster1"},
		},
		{
			name:    "unknown-fields-in-lb-config",
			input:   json.RawMessage(`{"Unknown": "foobar"}`),
			wantCfg: &lbConfig{ClusterName: ""},
		},
		{
			name:    "empty-lb-config",
			input:   json.RawMessage(""),
			wantErr: true,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			gotCfg, gotErr := parser.ParseConfig(test.input)
			if (gotErr != nil) != test.wantErr {
				t.Fatalf("ParseConfig(%v) = %v, wantErr %v", string(test.input), gotErr, test.wantErr)
			}
			if test.wantErr {
				return
			}
			if !cmp.Equal(gotCfg, test.wantCfg) {
				t.Fatalf("ParseConfig(%v) = %v, want %v", string(test.input), gotCfg, test.wantCfg)
			}
		})
	}
}
TestParseConfig verifies the ParseConfig() method in the CDS balancer.
package cdsbalancer
import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"testing"
	"time"
	"github.com/google/go-cmp/cmp"
	"github.com/google/go-cmp/cmp/cmpopts"
	"google.golang.org/grpc/balancer"
	"google.golang.org/grpc/connectivity"
	"google.golang.org/grpc/internal"
	"google.golang.org/grpc/internal/grpctest"
	internalserviceconfig "google.golang.org/grpc/internal/serviceconfig"
	"google.golang.org/grpc/internal/testutils"
	"google.golang.org/grpc/resolver"
	"google.golang.org/grpc/serviceconfig"
	"google.golang.org/grpc/xds/internal/balancer/clusterresolver"
	"google.golang.org/grpc/xds/internal/balancer/ringhash"
	xdstestutils "google.golang.org/grpc/xds/internal/testutils"
	"google.golang.org/grpc/xds/internal/testutils/fakeclient"
	"google.golang.org/grpc/xds/internal/xdsclient"
)
const (
	clusterName             = "cluster1"
	serviceName             = "service1"
	defaultTestTimeout      = 5 * time.Second
	defaultTestShortTimeout = 10 * time.Millisecond  
)
type s struct {
	grpctest.Tester
}
func Test(t *testing.T) {
	grpctest.RunSubTests(t, s{})
}
type cdsWatchInfo struct {
	update xdsclient.ClusterUpdate
	err    error
}
func invokeWatchCbAndWait(ctx context.Context, xdsC *fakeclient.Client, cdsW cdsWatchInfo, wantCCS balancer.ClientConnState, edsB *testEDSBalancer) error {
	xdsC.InvokeWatchClusterCallback(cdsW.update, cdsW.err)
	if cdsW.err != nil {
		return edsB.waitForResolverError(ctx, cdsW.err)
	}
	return edsB.waitForClientConnUpdate(ctx, wantCCS)
}
type testEDSBalancer struct {
	ccsCh *testutils.Channel
	scStateCh *testutils.Channel
	resolverErrCh *testutils.Channel
	closeCh    *testutils.Channel
	exitIdleCh *testutils.Channel
	parentCC balancer.ClientConn
}
type subConnWithState struct {
	sc    balancer.SubConn
	state balancer.SubConnState
}
func newTestEDSBalancer() *testEDSBalancer {
	return &testEDSBalancer{
		ccsCh:         testutils.NewChannel(),
		scStateCh:     testutils.NewChannel(),
		resolverErrCh: testutils.NewChannel(),
		closeCh:       testutils.NewChannel(),
		exitIdleCh:    testutils.NewChannel(),
	}
}
func (tb *testEDSBalancer) UpdateClientConnState(ccs balancer.ClientConnState) error {
	tb.ccsCh.Send(ccs)
	return nil
}
func (tb *testEDSBalancer) ResolverError(err error) {
	tb.resolverErrCh.Send(err)
}
func (tb *testEDSBalancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
	tb.scStateCh.Send(subConnWithState{sc: sc, state: state})
}
func (tb *testEDSBalancer) Close() {
	tb.closeCh.Send(struct{}{})
}
func (tb *testEDSBalancer) ExitIdle() {
	tb.exitIdleCh.Send(struct{}{})
}
func (tb *testEDSBalancer) waitForClientConnUpdate(ctx context.Context, wantCCS balancer.ClientConnState) error {
	ccs, err := tb.ccsCh.Receive(ctx)
	if err != nil {
		return err
	}
	gotCCS := ccs.(balancer.ClientConnState)
	if xdsclient.FromResolverState(gotCCS.ResolverState) == nil {
		return fmt.Errorf("want resolver state with XDSClient attached, got one without")
	}
	if diff := cmp.Diff(gotCCS, wantCCS, cmpopts.IgnoreFields(resolver.State{}, "Attributes")); diff != "" {
		return fmt.Errorf("received unexpected ClientConnState, diff (-got +want): %v", diff)
	}
	return nil
}
func (tb *testEDSBalancer) waitForSubConnUpdate(ctx context.Context, wantSCS subConnWithState) error {
	scs, err := tb.scStateCh.Receive(ctx)
	if err != nil {
		return err
	}
	gotSCS := scs.(subConnWithState)
	if !cmp.Equal(gotSCS, wantSCS, cmp.AllowUnexported(subConnWithState{})) {
		return fmt.Errorf("received SubConnState: %+v, want %+v", gotSCS, wantSCS)
	}
	return nil
}
func (tb *testEDSBalancer) waitForResolverError(ctx context.Context, wantErr error) error {
	gotErr, err := tb.resolverErrCh.Receive(ctx)
	if err != nil {
		return err
	}
	if gotErr != wantErr {
		return fmt.Errorf("received resolver error: %v, want %v", gotErr, wantErr)
	}
	return nil
}
func (tb *testEDSBalancer) waitForClose(ctx context.Context) error {
	if _, err := tb.closeCh.Receive(ctx); err != nil {
		return err
	}
	return nil
}
func cdsCCS(cluster string, xdsC xdsclient.XDSClient) balancer.ClientConnState {
	const cdsLBConfig = `{
      "loadBalancingConfig":[
        {
          "cds_experimental":{
            "Cluster": "%s"
          }
        }
      ]
    }`
	jsonSC := fmt.Sprintf(cdsLBConfig, cluster)
	return balancer.ClientConnState{
		ResolverState: xdsclient.SetClient(resolver.State{
			ServiceConfig: internal.ParseServiceConfigForTesting.(func(string) *serviceconfig.ParseResult)(jsonSC),
		}, xdsC),
		BalancerConfig: &lbConfig{ClusterName: clusterName},
	}
}
func edsCCS(service string, countMax *uint32, enableLRS bool, xdslbpolicy *internalserviceconfig.BalancerConfig) balancer.ClientConnState {
	discoveryMechanism := clusterresolver.DiscoveryMechanism{
		Type:                  clusterresolver.DiscoveryMechanismTypeEDS,
		Cluster:               service,
		MaxConcurrentRequests: countMax,
	}
	if enableLRS {
		discoveryMechanism.LoadReportingServerName = new(string)
	}
	lbCfg := &clusterresolver.LBConfig{
		DiscoveryMechanisms: []clusterresolver.DiscoveryMechanism{discoveryMechanism},
		XDSLBPolicy:         xdslbpolicy,
	}
	return balancer.ClientConnState{
		BalancerConfig: lbCfg,
	}
}
func setup(t *testing.T) (*fakeclient.Client, *cdsBalancer, *testEDSBalancer, *xdstestutils.TestClientConn, func()) {
	t.Helper()
	xdsC := fakeclient.NewClient()
	builder := balancer.Get(cdsName)
	if builder == nil {
		t.Fatalf("balancer.Get(%q) returned nil", cdsName)
	}
	tcc := xdstestutils.NewTestClientConn(t)
	cdsB := builder.Build(tcc, balancer.BuildOptions{})
	edsB := newTestEDSBalancer()
	oldEDSBalancerBuilder := newChildBalancer
	newChildBalancer = func(cc balancer.ClientConn, opts balancer.BuildOptions) (balancer.Balancer, error) {
		edsB.parentCC = cc
		return edsB, nil
	}
	return xdsC, cdsB.(*cdsBalancer), edsB, tcc, func() {
		newChildBalancer = oldEDSBalancerBuilder
		xdsC.Close()
	}
}
func setupWithWatch(t *testing.T) (*fakeclient.Client, *cdsBalancer, *testEDSBalancer, *xdstestutils.TestClientConn, func()) {
	t.Helper()
	xdsC, cdsB, edsB, tcc, cancel := setup(t)
	if err := cdsB.UpdateClientConnState(cdsCCS(clusterName, xdsC)); err != nil {
		t.Fatalf("cdsBalancer.UpdateClientConnState failed with error: %v", err)
	}
	ctx, ctxCancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer ctxCancel()
	gotCluster, err := xdsC.WaitForWatchCluster(ctx)
	if err != nil {
		t.Fatalf("xdsClient.WatchCDS failed with error: %v", err)
	}
	if gotCluster != clusterName {
		t.Fatalf("xdsClient.WatchCDS called for cluster: %v, want: %v", gotCluster, clusterName)
	}
	return xdsC, cdsB, edsB, tcc, cancel
}
func (s) TestUpdateClientConnState(t *testing.T) {
	xdsC := fakeclient.NewClient()
	defer xdsC.Close()
	tests := []struct {
		name        string
		ccs         balancer.ClientConnState
		wantErr     error
		wantCluster string
	}{
		{
			name:    "bad-lbCfg-type",
			ccs:     balancer.ClientConnState{BalancerConfig: nil},
			wantErr: balancer.ErrBadResolverState,
		},
		{
			name:    "empty-cluster-in-lbCfg",
			ccs:     balancer.ClientConnState{BalancerConfig: &lbConfig{ClusterName: ""}},
			wantErr: balancer.ErrBadResolverState,
		},
		{
			name:        "happy-good-case",
			ccs:         cdsCCS(clusterName, xdsC),
			wantCluster: clusterName,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			_, cdsB, _, _, cancel := setup(t)
			defer func() {
				cancel()
				cdsB.Close()
			}()
			if err := cdsB.UpdateClientConnState(test.ccs); err != test.wantErr {
				t.Fatalf("cdsBalancer.UpdateClientConnState failed with error: %v", err)
			}
			if test.wantErr != nil {
				return
			}
			ctx, ctxCancel := context.WithTimeout(context.Background(), defaultTestTimeout)
			defer ctxCancel()
			gotCluster, err := xdsC.WaitForWatchCluster(ctx)
			if err != nil {
				t.Fatalf("xdsClient.WatchCDS failed with error: %v", err)
			}
			if gotCluster != test.wantCluster {
				t.Fatalf("xdsClient.WatchCDS called for cluster: %v, want: %v", gotCluster, test.wantCluster)
			}
		})
	}
}
func (s) TestUpdateClientConnStateWithSameState(t *testing.T) {
	xdsC, cdsB, _, _, cancel := setupWithWatch(t)
	defer func() {
		cancel()
		cdsB.Close()
	}()
	if err := cdsB.UpdateClientConnState(cdsCCS(clusterName, xdsC)); err != nil {
		t.Fatalf("cdsBalancer.UpdateClientConnState failed with error: %v", err)
	}
	ctx, ctxCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
	defer ctxCancel()
	if _, err := xdsC.WaitForWatchCluster(ctx); err != context.DeadlineExceeded {
		t.Fatalf("waiting for WatchCluster() should have timed out, but returned error: %v", err)
	}
}
func (s) TestHandleClusterUpdate(t *testing.T) {
	xdsC, cdsB, edsB, _, cancel := setupWithWatch(t)
	defer func() {
		cancel()
		cdsB.Close()
	}()
	tests := []struct {
		name      string
		cdsUpdate xdsclient.ClusterUpdate
		updateErr error
		wantCCS   balancer.ClientConnState
	}{
		{
			name:      "happy-case-with-lrs",
			cdsUpdate: xdsclient.ClusterUpdate{ClusterName: serviceName, EnableLRS: true},
			wantCCS:   edsCCS(serviceName, nil, true, nil),
		},
		{
			name:      "happy-case-without-lrs",
			cdsUpdate: xdsclient.ClusterUpdate{ClusterName: serviceName},
			wantCCS:   edsCCS(serviceName, nil, false, nil),
		},
		{
			name: "happy-case-with-ring-hash-lb-policy",
			cdsUpdate: xdsclient.ClusterUpdate{
				ClusterName: serviceName,
				LBPolicy:    &xdsclient.ClusterLBPolicyRingHash{MinimumRingSize: 10, MaximumRingSize: 100},
			},
			wantCCS: edsCCS(serviceName, nil, false, &internalserviceconfig.BalancerConfig{
				Name:   ringhash.Name,
				Config: &ringhash.LBConfig{MinRingSize: 10, MaxRingSize: 100},
			}),
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			ctx, ctxCancel := context.WithTimeout(context.Background(), defaultTestTimeout)
			defer ctxCancel()
			if err := invokeWatchCbAndWait(ctx, xdsC, cdsWatchInfo{test.cdsUpdate, test.updateErr}, test.wantCCS, edsB); err != nil {
				t.Fatal(err)
			}
		})
	}
}
func (s) TestHandleClusterUpdateError(t *testing.T) {
	xdsC, cdsB, edsB, tcc, cancel := setupWithWatch(t)
	defer func() {
		cancel()
		cdsB.Close()
	}()
	watcherErr := errors.New("cdsBalancer watcher error")
	xdsC.InvokeWatchClusterCallback(xdsclient.ClusterUpdate{}, watcherErr)
	sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
	defer sCancel()
	if _, err := xdsC.WaitForCancelClusterWatch(sCtx); err != context.DeadlineExceeded {
		t.Fatal("cluster watch cancelled for a non-resource-not-found-error")
	}
	sCtx, sCancel = context.WithTimeout(context.Background(), defaultTestShortTimeout)
	defer sCancel()
	if err := edsB.waitForResolverError(sCtx, watcherErr); err != context.DeadlineExceeded {
		t.Fatal("eds balancer shouldn't get error (shouldn't be built yet)")
	}
	ctx, ctxCancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer ctxCancel()
	select {
	case <-ctx.Done():
		t.Fatalf("timeout when waiting for an error picker")
	case picker := <-tcc.NewPickerCh:
		if _, perr := picker.Pick(balancer.PickInfo{}); perr == nil {
			t.Fatalf("CDS balancer returned a picker which is not an error picker")
		}
	}
	cdsUpdate := xdsclient.ClusterUpdate{ClusterName: serviceName}
	wantCCS := edsCCS(serviceName, nil, false, nil)
	if err := invokeWatchCbAndWait(ctx, xdsC, cdsWatchInfo{cdsUpdate, nil}, wantCCS, edsB); err != nil {
		t.Fatal(err)
	}
	xdsC.InvokeWatchClusterCallback(xdsclient.ClusterUpdate{}, watcherErr)
	sCtx, sCancel = context.WithTimeout(context.Background(), defaultTestShortTimeout)
	defer sCancel()
	if _, err := xdsC.WaitForCancelClusterWatch(sCtx); err != context.DeadlineExceeded {
		t.Fatal("cluster watch cancelled for a non-resource-not-found-error")
	}
	if err := edsB.waitForResolverError(ctx, watcherErr); err != nil {
		t.Fatalf("Watch callback error is not forwarded to EDS balancer")
	}
	resourceErr := xdsclient.NewErrorf(xdsclient.ErrorTypeResourceNotFound, "cdsBalancer resource not found error")
	xdsC.InvokeWatchClusterCallback(xdsclient.ClusterUpdate{}, resourceErr)
	sCtx, sCancel = context.WithTimeout(context.Background(), defaultTestShortTimeout)
	defer sCancel()
	if _, err := xdsC.WaitForCancelClusterWatch(sCtx); err != context.DeadlineExceeded {
		t.Fatal("cluster watch cancelled for a resource-not-found-error")
	}
	if err := edsB.waitForResolverError(ctx, resourceErr); err != nil {
		t.Fatalf("Watch callback error is not forwarded to EDS balancer")
	}
}
func (s) TestResolverError(t *testing.T) {
	xdsC, cdsB, edsB, tcc, cancel := setupWithWatch(t)
	defer func() {
		cancel()
		cdsB.Close()
	}()
	resolverErr := errors.New("cdsBalancer resolver error")
	cdsB.ResolverError(resolverErr)
	sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
	defer sCancel()
	if _, err := xdsC.WaitForCancelClusterWatch(sCtx); err != context.DeadlineExceeded {
		t.Fatal("cluster watch cancelled for a non-resource-not-found-error")
	}
	sCtx, sCancel = context.WithTimeout(context.Background(), defaultTestShortTimeout)
	defer sCancel()
	if err := edsB.waitForResolverError(sCtx, resolverErr); err != context.DeadlineExceeded {
		t.Fatal("eds balancer shouldn't get error (shouldn't be built yet)")
	}
	ctx, ctxCancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer ctxCancel()
	select {
	case <-ctx.Done():
		t.Fatalf("timeout when waiting for an error picker")
	case picker := <-tcc.NewPickerCh:
		if _, perr := picker.Pick(balancer.PickInfo{}); perr == nil {
			t.Fatalf("CDS balancer returned a picker which is not an error picker")
		}
	}
	cdsUpdate := xdsclient.ClusterUpdate{ClusterName: serviceName}
	wantCCS := edsCCS(serviceName, nil, false, nil)
	if err := invokeWatchCbAndWait(ctx, xdsC, cdsWatchInfo{cdsUpdate, nil}, wantCCS, edsB); err != nil {
		t.Fatal(err)
	}
	cdsB.ResolverError(resolverErr)
	sCtx, sCancel = context.WithTimeout(context.Background(), defaultTestShortTimeout)
	defer sCancel()
	if _, err := xdsC.WaitForCancelClusterWatch(sCtx); err != context.DeadlineExceeded {
		t.Fatal("cluster watch cancelled for a non-resource-not-found-error")
	}
	if err := edsB.waitForResolverError(ctx, resolverErr); err != nil {
		t.Fatalf("ResolverError() not forwarded to EDS balancer")
	}
	resourceErr := xdsclient.NewErrorf(xdsclient.ErrorTypeResourceNotFound, "cdsBalancer resource not found error")
	cdsB.ResolverError(resourceErr)
	if _, err := xdsC.WaitForCancelClusterWatch(ctx); err != nil {
		t.Fatalf("want watch to be canceled, watchForCancel failed: %v", err)
	}
	if err := edsB.waitForResolverError(ctx, resourceErr); err != nil {
		t.Fatalf("eds balancer should get resource-not-found error, waitForError failed: %v", err)
	}
}
func (s) TestUpdateSubConnState(t *testing.T) {
	xdsC, cdsB, edsB, _, cancel := setupWithWatch(t)
	defer func() {
		cancel()
		cdsB.Close()
	}()
	cdsUpdate := xdsclient.ClusterUpdate{ClusterName: serviceName}
	wantCCS := edsCCS(serviceName, nil, false, nil)
	ctx, ctxCancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer ctxCancel()
	if err := invokeWatchCbAndWait(ctx, xdsC, cdsWatchInfo{cdsUpdate, nil}, wantCCS, edsB); err != nil {
		t.Fatal(err)
	}
	var sc balancer.SubConn
	state := balancer.SubConnState{ConnectivityState: connectivity.Ready}
	cdsB.UpdateSubConnState(sc, state)
	if err := edsB.waitForSubConnUpdate(ctx, subConnWithState{sc: sc, state: state}); err != nil {
		t.Fatal(err)
	}
}
func (s) TestCircuitBreaking(t *testing.T) {
	xdsC, cdsB, edsB, _, cancel := setupWithXDSCreds(t)
	defer func() {
		cancel()
		cdsB.Close()
	}()
	var maxRequests uint32 = 1
	cdsUpdate := xdsclient.ClusterUpdate{ClusterName: clusterName, MaxRequests: &maxRequests}
	wantCCS := edsCCS(clusterName, &maxRequests, false, nil)
	ctx, ctxCancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer ctxCancel()
	if err := invokeWatchCbAndWait(ctx, xdsC, cdsWatchInfo{cdsUpdate, nil}, wantCCS, edsB); err != nil {
		t.Fatal(err)
	}
	counter := xdsclient.GetClusterRequestsCounter(clusterName, "")
	if err := counter.StartRequest(maxRequests); err != nil {
		t.Fatal(err)
	}
	if err := counter.StartRequest(maxRequests); err == nil {
		t.Fatal("unexpected success on start request over max")
	}
	counter.EndRequest()
}
func (s) TestClose(t *testing.T) {
	xdsC, cdsB, edsB, _, cancel := setupWithWatch(t)
	defer cancel()
	cdsUpdate := xdsclient.ClusterUpdate{ClusterName: serviceName}
	wantCCS := edsCCS(serviceName, nil, false, nil)
	ctx, ctxCancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer ctxCancel()
	if err := invokeWatchCbAndWait(ctx, xdsC, cdsWatchInfo{cdsUpdate, nil}, wantCCS, edsB); err != nil {
		t.Fatal(err)
	}
	cdsB.Close()
	if _, err := xdsC.WaitForCancelClusterWatch(ctx); err != nil {
		t.Fatal(err)
	}
	xdsC.InvokeWatchClusterCallback(cdsUpdate, nil)
	sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
	defer sCancel()
	if err := edsB.waitForClientConnUpdate(sCtx, balancer.ClientConnState{}); err != context.DeadlineExceeded {
		t.Fatalf("ClusterUpdate after close forwaded to EDS balancer")
	}
	if err := edsB.waitForClose(ctx); err != nil {
		t.Fatal(err)
	}
	if err := cdsB.UpdateClientConnState(cdsCCS(clusterName, xdsC)); err != errBalancerClosed {
		t.Fatalf("UpdateClientConnState() after close returned %v, want %v", err, errBalancerClosed)
	}
	cdsB.UpdateSubConnState(&xdstestutils.TestSubConn{}, balancer.SubConnState{})
	sCtx, sCancel = context.WithTimeout(context.Background(), defaultTestShortTimeout)
	defer sCancel()
	if err := edsB.waitForSubConnUpdate(sCtx, subConnWithState{}); err != context.DeadlineExceeded {
		t.Fatal("UpdateSubConnState() forwarded to EDS balancer after Close()")
	}
	rErr := errors.New("cdsBalancer resolver error")
	cdsB.ResolverError(rErr)
	sCtx, sCancel = context.WithTimeout(context.Background(), defaultTestShortTimeout)
	defer sCancel()
	if err := edsB.waitForResolverError(sCtx, rErr); err != context.DeadlineExceeded {
		t.Fatal("ResolverError() forwarded to EDS balancer after Close()")
	}
}
func (s) TestExitIdle(t *testing.T) {
	xdsC, cdsB, edsB, _, cancel := setupWithWatch(t)
	defer func() {
		cancel()
		cdsB.Close()
	}()
	cdsUpdate := xdsclient.ClusterUpdate{ClusterName: serviceName}
	wantCCS := edsCCS(serviceName, nil, false, nil)
	ctx, ctxCancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer ctxCancel()
	if err := invokeWatchCbAndWait(ctx, xdsC, cdsWatchInfo{cdsUpdate, nil}, wantCCS, edsB); err != nil {
		t.Fatal(err)
	}
	cdsB.ExitIdle()
	edsB.exitIdleCh.Receive(ctx)
}