ronantakizawa/codereview-bench huggingface dataset:
Repository: grpc/grpc-go
File: internal/balancergroup/balancergroup_test.go (Go)

Code before:

	cc := testutils.NewBalancerClientConn(t)
	bg := New(Options{
		CC:              cc,
		BuildOpts:       balancer.BuildOptions{},
		StateAggregator: nil,
		Logger:          nil,
	})
	defer bg.Close()

	bg.Add(testIDs[0], balancer.Get(balancerOne))
	bg.Add(testIDs[1], balancer.Get(balancerTwo))
	bg.Add(testIDs[2], balancer.Get(balancerThree))

	bg.ExitIdle()

	called := make(map[string]bool)
	for i := 0; i < len(balancerNames); i++ {
		select {
		case name := <-exitIdleCh:
			if called[name] {
				t.Fatalf("ExitIdle was called multiple times for sub-balancer %q", name)
			}
			called[name] = true
		case <-time.After(time.Second):
			t.Fatalf("Timeout: ExitIdle not called for all sub-balancers, got %d/%d", len(called), len(balancerNames))
		}
	}
}

func (s) TestBalancerGroup_ExitIdle_AfterClose(t *testing.T) {
	balancerName := "stub-balancer-test-balancer-group-exit-idle-after-close"
	exitIdleCh := make(chan struct{})

	stub.Register(balancerName, stub.BalancerFuncs{
		ExitIdle: func(_ *stub.BalancerData) {
			close(exitIdleCh)
		},
	})

	bg := New(Options{
		CC:              testutils.NewBalancerClientConn(t),
		BuildOpts:       balancer.BuildOptions{},
		StateAggregator: nil,
		Logger:          nil,
	})

	bg.Add(testBalancerIDs[0], balancer.Get(balancerName))
	bg.Close()
	bg.ExitIdle()


Review comment (suggestion):
The error can be changed to give specific name of the balancer for which it was not called.
eg:
`t.Fatalf("Timeout waiting for ExitIdle to be called for %v , name")`

Code after:
		},
	})

	cc := testutils.NewBalancerClientConn(t)
	bg := New(Options{
		CC:              cc,
		BuildOpts:       balancer.BuildOptions{},
		StateAggregator: nil,
		Logger:          nil,
	})
	defer bg.Close()

	bg.Add(testID1, balancer.Get(balancer1))
	bg.Add(testID2, balancer.Get(balancer2))

	bg.ExitIdle()

	errCh := make(chan error, 2)

	go func() {
		select {
		case <-exitIdleCh1:
			errCh <- nil
		case <-time.After(defaultTestTimeout):
			errCh <- fmt.Errorf("timeout waiting for ExitIdle on balancer1")
		}
	}()

	go func() {
		select {
		case <-exitIdleCh2:
			errCh <- nil
		case <-time.After(defaultTestTimeout):
			errCh <- fmt.Errorf("timeout waiting for ExitIdle on balancer2")
		}
	}()

	for i := 0; i < 2; i++ {
		if err := <-errCh; err != nil {
			t.Fatal(err)
		}
	}
}

func (s) TestBalancerGroup_ExitIdle_AfterClose(t *testing.T) {
	balancerName := t.Name()
	exitIdleCh := make(chan struct{}, 1)

	stub.Register(balancerName, stub.BalancerFuncs{
		ExitIdle: func(_ *stub.BalancerData) {
			exitIdleCh <- struct{}{}
Repository: grpc/grpc-go
File: internal/balancergroup/balancergroup_test.go (Go)

Code before:
			}
			return nil
		},
	})
	cc := testutils.NewBalancerClientConn(t)
	bg := New(Options{
		CC:              cc,
		BuildOpts:       bOpts,
		StateAggregator: nil,
		Logger:          nil,
	})

	// Add the stub balancer build above as a child policy.
	balancerBuilder := balancer.Get(balancerName)
	bg.Add(testBalancerIDs[0], balancerBuilder)

	// Send an empty clientConn state change. This should trigger the
	// verification of the buildOptions being passed to the child policy.
	if err := bg.UpdateClientConnState(testBalancerIDs[0], balancer.ClientConnState{}); err != nil {
		t.Fatal(err)
	}
}

func (s) TestBalancerGroup_UpdateClientConnState_AfterClose(t *testing.T) {
	balancerName := t.Name()
	exitIdleCh := make(chan struct{}, 1)

	stub.Register(balancerName, stub.BalancerFuncs{
		UpdateClientConnState: func(_ *stub.BalancerData, _ balancer.ClientConnState) error {
			exitIdleCh <- struct{}{}
			return nil
		},
	})

	bg := New(Options{
		CC:              testutils.NewBalancerClientConn(t),
		BuildOpts:       balancer.BuildOptions{},
		StateAggregator: nil,
		Logger:          nil,
	})

	bg.Add(testBalancerIDs[0], balancer.Get(balancerName))
	bg.Close()

	if err := bg.UpdateClientConnState(testBalancerIDs[0], balancer.ClientConnState{}); err != nil {
		t.Fatalf("Expected nil error, got %v", err)
	}

	select {
	case <-exitIdleCh:
		t.Fatalf("UpdateClientConnState was called after BalancerGroup was closed")

Review comment (suggestion):
This is not a channel used to indicate that `ExitIdle` was called. Can you please rename it appropriately. Thanks.

Code after:
			}
			return nil
		},
	})
	cc := testutils.NewBalancerClientConn(t)
	bg := New(Options{
		CC:              cc,
		BuildOpts:       bOpts,
		StateAggregator: nil,
		Logger:          nil,
	})

	// Add the stub balancer build above as a child policy.
	balancerBuilder := balancer.Get(balancerName)
	bg.Add(testBalancerIDs[0], balancerBuilder)

	// Send an empty clientConn state change. This should trigger the
	// verification of the buildOptions being passed to the child policy.
	if err := bg.UpdateClientConnState(testBalancerIDs[0], balancer.ClientConnState{}); err != nil {
		t.Fatal(err)
	}
}

func (s) TestBalancerGroup_UpdateClientConnState_AfterClose(t *testing.T) {
	balancerName := t.Name()
	clientConnStateCh := make(chan struct{}, 1)

	stub.Register(balancerName, stub.BalancerFuncs{
		UpdateClientConnState: func(_ *stub.BalancerData, _ balancer.ClientConnState) error {
			clientConnStateCh <- struct{}{}
			return nil
		},
	})

	bg := New(Options{
		CC:              testutils.NewBalancerClientConn(t),
		BuildOpts:       balancer.BuildOptions{},
		StateAggregator: nil,
		Logger:          nil,
	})

	bg.Add(testBalancerIDs[0], balancer.Get(balancerName))
	bg.Close()

	if err := bg.UpdateClientConnState(testBalancerIDs[0], balancer.ClientConnState{}); err != nil {
		t.Fatalf("Expected nil error, got %v", err)
	}

	select {
	case <-clientConnStateCh:
		t.Fatalf("UpdateClientConnState was called after BalancerGroup was closed")
Repository: grpc/grpc-go
File: internal/balancergroup/balancergroup_test.go (Go)

Code before:
		case <-exitIdleCh1:
			errCh <- nil
		case <-time.After(defaultTestTimeout):
			errCh <- fmt.Errorf("timeout waiting for ExitIdle on balancer1")
		}
	}()

	go func() {
		select {
		case <-exitIdleCh2:
			errCh <- nil
		case <-time.After(defaultTestTimeout):
			errCh <- fmt.Errorf("timeout waiting for ExitIdle on balancer2")
		}
	}()

	for i := 0; i < 2; i++ {
		if err := <-errCh; err != nil {
			t.Fatal(err)
		}
	}
}

func (s) TestBalancerGroup_ExitIdle_AfterClose(t *testing.T) {
	balancerName := t.Name()
	exitIdleCh := make(chan struct{})

	stub.Register(balancerName, stub.BalancerFuncs{
		ExitIdle: func(_ *stub.BalancerData) {
			exitIdleCh <- struct{}{}
		},
	})

	bg := New(Options{
		CC:              testutils.NewBalancerClientConn(t),
		BuildOpts:       balancer.BuildOptions{},
		StateAggregator: nil,
		Logger:          nil,
	})

	bg.Add(testBalancerIDs[0], balancer.Get(balancerName))
	bg.Close()
	bg.ExitIdle()

	select {
	case <-exitIdleCh:
		t.Fatalf("ExitIdle was called on sub-balancer even after BalancerGroup was closed")
	case <-time.After(defaultTestShortTimeout):
	}
}


Review comment (suggestion):
Oops. Sorry, didn't catch this earlier. This channel needs to have a buffer of `1`. Otherwise, the writer would block if there is no reader.

Code after:
		case <-exitIdleCh1:
			errCh <- nil
		case <-time.After(defaultTestTimeout):
			errCh <- fmt.Errorf("timeout waiting for ExitIdle on balancer1")
		}
	}()

	go func() {
		select {
		case <-exitIdleCh2:
			errCh <- nil
		case <-time.After(defaultTestTimeout):
			errCh <- fmt.Errorf("timeout waiting for ExitIdle on balancer2")
		}
	}()

	for i := 0; i < 2; i++ {
		if err := <-errCh; err != nil {
			t.Fatal(err)
		}
	}
}

func (s) TestBalancerGroup_ExitIdle_AfterClose(t *testing.T) {
	balancerName := t.Name()
	exitIdleCh := make(chan struct{}, 1)

	stub.Register(balancerName, stub.BalancerFuncs{
		ExitIdle: func(_ *stub.BalancerData) {
			exitIdleCh <- struct{}{}
		},
	})

	bg := New(Options{
		CC:              testutils.NewBalancerClientConn(t),
		BuildOpts:       balancer.BuildOptions{},
		StateAggregator: nil,
		Logger:          nil,
	})

	bg.Add(testBalancerIDs[0], balancer.Get(balancerName))
	bg.Close()
	bg.ExitIdle()

	select {
	case <-exitIdleCh:
		t.Fatalf("ExitIdle was called on sub-balancer even after BalancerGroup was closed")
	case <-time.After(defaultTestShortTimeout):
	}
}
Repository: grpc/grpc-go
File: test/end2end_test.go (Go)

Code before:
		t.Fatalf("cc.NewStream() failed unexpectedly: %v", err)
	}

	if err := stream.SendMsg(&testpb.Empty{}); err != nil {
		t.Errorf("stream.SendMsg() = %v, want <nil>", err)
	}

	if err := stream.SendMsg(&testpb.Empty{}); err != nil {
		t.Errorf("stream.SendMsg() = %v, want <nil>", err)
	}

	if err := stream.RecvMsg(&testpb.Empty{}); status.Code(err) != codes.Internal {
		t.Errorf("stream.RecvMsg() = %v, want error %v", status.Code(err), codes.Internal)
	}
}

// Tests the behavior for server-side streaming RPC when client sends zero request message.
func (s) TestServerStreaming_ClientSendsZeroRequest(t *testing.T) {
	lis, err := testutils.LocalTCPListener()
	if err != nil {
		t.Fatal(err)
	}
	defer lis.Close()

	s := grpc.NewServer()
	serviceDesc := grpc.ServiceDesc{
		ServiceName: "grpc.testing.TestService",
		HandlerType: (*any)(nil),
		Methods:     []grpc.MethodDesc{},
		Streams: []grpc.StreamDesc{
			{
				StreamName: "ServerStreaming",
				Handler: func(_ any, stream grpc.ServerStream) error {
					if err = stream.RecvMsg(&testpb.Empty{}); status.Code(err) != codes.Internal {
						t.Errorf("stream.RecvMsg() = %v, want error %v", status.Code(err), codes.Internal)
					}
					return nil
				},
				ClientStreams: false,
				ServerStreams: true,
			},
		},
	}
	s.RegisterService(&serviceDesc, &testServer{})
	go s.Serve(lis)
	defer s.Stop()

	// s := grpc.NewServer(ss)
	go s.Serve(lis)
	defer s.Stop()


Review comment (suggestion):
I believe we don't need to change the stream descriptors on the server side for this test. If so, please use a stubserver and override one of the handlers.

Code after:
	if _, err := stream.CloseAndRecv(); status.Code(err) != codes.Internal {
		t.Fatalf("stream.CloseAndRecv() = %v, want error with status code %s", err, codes.Internal)
	}
}

func (s) TestExceedMaxStreamsLimit(t *testing.T) {
	for _, e := range listTestEnv() {
		testExceedMaxStreamsLimit(t, e)
	}
}

func testExceedMaxStreamsLimit(t *testing.T, e env) {
	te := newTest(t, e)
	te.declareLogNoise(
		"http2Client.notifyError got notified that the client transport was broken",
		"Conn.resetTransport failed to create client transport",
		"grpc: the connection is closing",
	)
	te.maxStream = 1 // Only allows 1 live stream per server transport.
	te.startServer(&testServer{security: e.security})
	defer te.tearDown()

	cc := te.clientConn()
	tc := testgrpc.NewTestServiceClient(cc)

	_, err := tc.StreamingInputCall(te.ctx)
	if err != nil {
		t.Fatalf("%v.StreamingInputCall(_) = _, %v, want _, <nil>", tc, err)
	}
	// Loop until receiving the new max stream setting from the server.
	for {
		ctx, cancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
		defer cancel()
		_, err := tc.StreamingInputCall(ctx)
		if err == nil {
			time.Sleep(50 * time.Millisecond)
			continue
		}
		if status.Code(err) == codes.DeadlineExceeded {
			break
		}
		t.Fatalf("%v.StreamingInputCall(_) = _, %v, want _, %s", tc, err, codes.DeadlineExceeded)
	}
}

func (s) TestStreamsQuotaRecovery(t *testing.T) {
	for _, e := range listTestEnv() {
		testStreamsQuotaRecovery(t, e)
	}
}
Repository: grpc/grpc-go
File: stream.go (Go)

Code before:
	// done. It returns io.EOF when the client has performed a CloseSend. On
	// any non-EOF error, the stream is aborted and the error contains the
	// RPC status.
	//
	// It is safe to have a goroutine calling SendMsg and another goroutine
	// calling RecvMsg on the same stream at the same time, but it is not
	// safe to call RecvMsg on the same stream in different goroutines.
	RecvMsg(m any) error
}

// serverStream implements a server side Stream.
type serverStream struct {
	ctx   context.Context
	s     *transport.ServerStream
	p     *parser
	codec baseCodec
	desc  *StreamDesc

	compressorV0   Compressor
	compressorV1   encoding.Compressor
	decompressorV0 Decompressor
	decompressorV1 encoding.Compressor

	sendCompressorName string

	recvFirstMsg bool // recv frist msg from client

	maxReceiveMessageSize int
	maxSendMessageSize    int
	trInfo                *traceInfo

	statsHandler []stats.Handler

	binlogs []binarylog.MethodLogger
	// serverHeaderBinlogged indicates whether server header has been logged. It
	// will happen when one of the following two happens: stream.SendHeader(),
	// stream.Send().
	//
	// It's only checked in send and sendHeader, doesn't need to be
	// synchronized.
	serverHeaderBinlogged bool

	mu sync.Mutex // protects trInfo.tr after the service handler runs.
}

func (ss *serverStream) Context() context.Context {
	return ss.ctx
}

func (ss *serverStream) SetHeader(md metadata.MD) error {
	if md.Len() == 0 {

Review comment (suggestion):
This comment doesn't really say anything.  How about:

```suggestion
	recvFirstMsg bool // set after the first message is received
```

Code after:
	// done. It returns io.EOF when the client has performed a CloseSend. On
	// any non-EOF error, the stream is aborted and the error contains the
	// RPC status.
	//
	// It is safe to have a goroutine calling SendMsg and another goroutine
	// calling RecvMsg on the same stream at the same time, but it is not
	// safe to call RecvMsg on the same stream in different goroutines.
	RecvMsg(m any) error
}

// serverStream implements a server side Stream.
type serverStream struct {
	ctx   context.Context
	s     *transport.ServerStream
	p     *parser
	codec baseCodec
	desc  *StreamDesc

	compressorV0   Compressor
	compressorV1   encoding.Compressor
	decompressorV0 Decompressor
	decompressorV1 encoding.Compressor

	sendCompressorName string

	recvFirstMsg bool // set after the first message is received

	maxReceiveMessageSize int
	maxSendMessageSize    int
	trInfo                *traceInfo

	statsHandler []stats.Handler

	binlogs []binarylog.MethodLogger
	// serverHeaderBinlogged indicates whether server header has been logged. It
	// will happen when one of the following two happens: stream.SendHeader(),
	// stream.Send().
	//
	// It's only checked in send and sendHeader, doesn't need to be
	// synchronized.
	serverHeaderBinlogged bool

	mu sync.Mutex // protects trInfo.tr after the service handler runs.
}

func (ss *serverStream) Context() context.Context {
	return ss.ctx
}

func (ss *serverStream) SetHeader(md metadata.MD) error {
	if md.Len() == 0 {
Repository: grpc/grpc-go
File: xds/internal/xdsclient/tests/fallback_test.go (Go)

Code before:
	secondaryConn, err := secondaryWrappedLis.NewConnCh.Receive(ctx)
	if err != nil {
		t.Fatalf("Expected new connection to secondary but got error: %v", err)
	}
	if _, err := tertiaryConn.(*testutils.ConnWrapper).CloseCh.Receive(ctx); err != nil {
		t.Fatalf("Expected tertiary connection to close after promotion to secondary, got error: %v", err)
	}
	if err := waitForRPCsToReachBackend(ctx, client, backend2.Address); err != nil {
		t.Fatal(err)
	}

	// Primary becomes available, RPCs go to backend1.
	primaryLis.Restart()
	primaryManagementServer.Update(ctx, e2e.DefaultClientResources(e2e.ResourceParams{
		DialTarget: serviceName,
		NodeID:     nodeID,
		Host:       "localhost",
		Port:       testutils.ParsePort(t, backend1.Address),
		SecLevel:   e2e.SecurityLevelNone,
	}))

	if _, err := primaryWrappedLis.NewConnCh.Receive(ctx); err != nil {
		t.Fatalf("Expected new connection to primary but got error: %v", err)
	}
	if _, err := secondaryConn.(*testutils.ConnWrapper).CloseCh.Receive(ctx); err != nil {
		t.Fatalf("Expected secondary connection to close after promotion to primary, got error: %v", err)
	}
	if err := waitForRPCsToReachBackend(ctx, client, backend1.Address); err != nil {
		t.Fatal(err)
	}
}


Review comment (suggestion):
For all test error messages which currently start with "Expected ... ", please reframe it with this style guide recommendation: https://google.github.io/styleguide/go/decisions#got-before-want

For example, this could be rewritten as "Timeout when waiting for connection to the secondary to be closed after promotion to primary".

Code after:
	secondaryConn, err := secondaryWrappedLis.NewConnCh.Receive(ctx)
	if err != nil {
		t.Fatalf("Timeout when waiting for new connection to secondary: %v", err)
	}
	if _, err := tertiaryConn.(*testutils.ConnWrapper).CloseCh.Receive(ctx); err != nil {
		t.Fatalf("Timeout when waiting for connection to the tertiary to be closed after promotion to secondary: %v", err)
	}
	if err := waitForRPCsToReachBackend(ctx, client, backend2.Address); err != nil {
		t.Fatal(err)
	}

	// Primary becomes available, RPCs go to backend1.
	primaryLis.Restart()
	primaryManagementServer.Update(ctx, e2e.DefaultClientResources(e2e.ResourceParams{
		DialTarget: serviceName,
		NodeID:     nodeID,
		Host:       "localhost",
		Port:       testutils.ParsePort(t, backend1.Address),
		SecLevel:   e2e.SecurityLevelNone,
	}))

	if _, err := primaryWrappedLis.NewConnCh.Receive(ctx); err != nil {
		t.Fatalf("Timeout when waiting for new connection to primary: %v", err)
	}
	if _, err := secondaryConn.(*testutils.ConnWrapper).CloseCh.Receive(ctx); err != nil {
		t.Fatalf("Timeout when waiting for connection to the secondary to be closed after promotion to primary: %v", err)
	}
	if err := waitForRPCsToReachBackend(ctx, client, backend1.Address); err != nil {
		t.Fatal(err)
	}
}
Repository: grpc/grpc-go
File: test/end2end_test.go (Go)

Code before:
		Addresses: []resolver.Address{
			{Addr: ss.Address},
		},
		ServiceConfig: sc,
	})

	cc, err := grpc.NewClient(mr.Scheme()+":///", grpc.WithResolvers(mr), grpc.WithStatsHandler(sh), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatalf("grpc.NewClient() failed: %v", err)
	}
	defer cc.Close()
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	testServiceClient := testgrpc.NewTestServiceClient(cc)
	if _, err := testServiceClient.UnaryCall(ctx, &testpb.SimpleRequest{}); err != nil {
		t.Fatalf("Unexpected error from UnaryCall: %v", err)
	}

	var delayedPickCompleteCount uint
	for _, stat := range sh.s {
		if _, ok := stat.(*stats.DelayedPickComplete); ok {
			delayedPickCompleteCount++
		}
	}
	if delayedPickCompleteCount != 1 {
		t.Fatalf("sh.delayedPickComplete count: %v, want: %v", delayedPickCompleteCount, 2)
	}
}


Review comment (suggestion):
We compare `delayedPickCompleteCount` to `1`, but the log mentions `2`. I think the log message needs to be fixed.

Code after:
		Addresses: []resolver.Address{
			{Addr: ss.Address},
		},
		ServiceConfig: sc,
	})

	cc, err := grpc.NewClient(mr.Scheme()+":///", grpc.WithResolvers(mr), grpc.WithStatsHandler(sh), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatalf("grpc.NewClient() failed: %v", err)
	}
	defer cc.Close()
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	testServiceClient := testgrpc.NewTestServiceClient(cc)
	if _, err := testServiceClient.UnaryCall(ctx, &testpb.SimpleRequest{}); err != nil {
		t.Fatalf("Unexpected error from UnaryCall: %v", err)
	}

	var delayedPickCompleteCount int
	for _, stat := range sh.s {
		if _, ok := stat.(*stats.DelayedPickComplete); ok {
			delayedPickCompleteCount++
		}
	}
	if got, want := delayedPickCompleteCount, 1; got != want {
		t.Fatalf("sh.delayedPickComplete count: %v, want: %v", got, want)
	}
}
Repository: grpc/grpc-go
File: xds/internal/clients/xdsclient/authority.go (Go)

Code before:
}

func serviceStatusToProto(serviceStatus xdsresource.ServiceStatus) v3adminpb.ClientResourceStatus {
	switch serviceStatus {
	case xdsresource.ServiceStatusUnknown:
		return v3adminpb.ClientResourceStatus_UNKNOWN
	case xdsresource.ServiceStatusRequested:
		return v3adminpb.ClientResourceStatus_REQUESTED
	case xdsresource.ServiceStatusNotExist:
		return v3adminpb.ClientResourceStatus_DOES_NOT_EXIST
	case xdsresource.ServiceStatusACKed:
		return v3adminpb.ClientResourceStatus_ACKED
	case xdsresource.ServiceStatusNACKed:
		return v3adminpb.ClientResourceStatus_NACKED
	default:
		return v3adminpb.ClientResourceStatus_UNKNOWN
	}
}

func (a *authority) resourceWatchStateForTesting(rType ResourceType, resourceName string) (state xdsresource.ResourceWatchState, err error) {
	done := make(chan struct{})
	a.xdsClientSerializer.ScheduleOr(func(context.Context) {
		state, err = a.activeXDSChannel.channel.ads.adsResourceWatchStateForTesting(rType, resourceName)
		close(done)
	}, func() { close(done) })
	<-done

	return state, err
}


Review comment (suggestion):
Should the scheduler have a function like:

```go
// Schedules f and waits for it to be executed.
func (cs *CallbackSerializer) ScheduleAndWait(f func(ctx context.Context))
```

Maybe even something nice with generics:

```go
// Schedules f and waits for it to be executed.  Returns what T returned or nil and
// ErrSerializerClosedBeforeItCouldRunFOrWhatever.
func (cs *CallbackSerializer) ScheduleAndWait[T any](f func(ctx context.Context) T) (T, error)
```

(I think we have this pattern multiple times?)

Just an idea to discuss, not for this PR.

Code after:
	}
}

func serviceStatusToProto(serviceStatus xdsresource.ServiceStatus) v3adminpb.ClientResourceStatus {
	switch serviceStatus {
	case xdsresource.ServiceStatusUnknown:
		return v3adminpb.ClientResourceStatus_UNKNOWN
	case xdsresource.ServiceStatusRequested:
		return v3adminpb.ClientResourceStatus_REQUESTED
	case xdsresource.ServiceStatusNotExist:
		return v3adminpb.ClientResourceStatus_DOES_NOT_EXIST
	case xdsresource.ServiceStatusACKed:
		return v3adminpb.ClientResourceStatus_ACKED
	case xdsresource.ServiceStatusNACKed:
		return v3adminpb.ClientResourceStatus_NACKED
	default:
		return v3adminpb.ClientResourceStatus_UNKNOWN
	}
}

func (a *authority) resourceWatchStateForTesting(rType ResourceType, resourceName string) (state xdsresource.ResourceWatchState, err error) {
	done := make(chan struct{})
	a.xdsClientSerializer.ScheduleOr(func(context.Context) {
		state, err = a.activeXDSChannel.channel.ads.adsResourceWatchStateForTesting(rType, resourceName)
		close(done)
	}, func() {
		err = errors.New("failed to retrieve resource watch state because the xDS client is closed")
		close(done)
	})
	<-done

	return state, err
}