test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-12-02 03:48:37 +08:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity

fix: grpc client check session skipped due to role not match (#29356)

Browse Source 
Related to #28815

---------

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
This commit is contained in:
congqixia 2023-12-21 10:12:51 +08:00 committed by GitHub
parent f457b9f7c9
commit f699be79f7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 13 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
internal/util/grpcclient/client.go
View File

@ -106,8 +106,10 @@ type ClientBase[T interface {
encryption bool
addr atomic.String
// conn *grpc.ClientConn
grpcClientMtx sync.RWMutex
role string
grpcClientMtx sync.RWMutex
role string
isNode bool // pre-calculated is node flag
ClientMaxSendSize int
ClientMaxRecvSize int
CompressionEnabled bool
@ -159,6 +161,12 @@ func NewClientBase[T interface {
// SetRole sets role of client
func (c *ClientBase[T]) SetRole(role string) {
c.role = role
if strings.HasPrefix(role, typeutil.DataNodeRole) ||
strings.HasPrefix(role, typeutil.IndexNodeRole) ||
strings.HasPrefix(role, typeutil.QueryNodeRole) ||
strings.HasPrefix(role, typeutil.ProxyRole) {
c.isNode = true
}
}
// GetRole returns role of client
@ -417,8 +425,7 @@ func (c *ClientBase[T]) checkGrpcErr(ctx context.Context, err error) (needRetry,
}
func (c *ClientBase[T]) checkNodeSessionExist(ctx context.Context) (bool, error) {
switch c.GetRole() {
case typeutil.DataNodeRole, typeutil.IndexNodeRole, typeutil.QueryNodeRole, typeutil.ProxyRole:
if c.isNode {
err := c.verifySession(ctx)
if errors.Is(err, merr.ErrNodeNotFound) {
log.Warn("failed to verify node session", zap.Error(err))

2
internal/util/grpcclient/client_test.go
View File

@ -98,6 +98,7 @@ func TestClientBase_NodeSessionNotExist(t *testing.T) {
base := ClientBase[*mockClient]{
maxCancelError: 10,
MaxAttempts: 3,
isNode: true,
}
base.SetGetAddrFunc(func() (string, error) {
return "", errors.New("mocked address error")
@ -148,6 +149,7 @@ func testCall(t *testing.T, compressed bool) {
base := ClientBase[*mockClient]{
maxCancelError: 10,
MaxAttempts: 3,
isNode: true,
}
base.CompressionEnabled = compressed
initClient := func() {