mirror of https://github.com/milvus-io/milvus.git
Make the `UnsubDmChannel` rpc unsuccessful when the releaseCollectionTask fails (#21469)
Signed-off-by: SimFG <bang.fu@zilliz.com>pull/21486/head
parent
772e5a4a3b
commit
289df1cc3d
|
@ -433,19 +433,19 @@ func (node *QueryNode) UnsubDmChannel(ctx context.Context, req *querypb.UnsubDmC
|
|||
}
|
||||
log.Info("unsubDmChannel(ReleaseCollection) enqueue done", zap.Int64("collectionID", req.GetCollectionID()))
|
||||
|
||||
func() {
|
||||
err = dct.WaitToFinish()
|
||||
if err != nil {
|
||||
log.Warn("failed to do subscribe channel task successfully", zap.Error(err))
|
||||
return
|
||||
return &commonpb.Status{
|
||||
ErrorCode: commonpb.ErrorCode_UnexpectedError,
|
||||
Reason: err.Error(),
|
||||
}, nil
|
||||
}
|
||||
log.Info("unsubDmChannel(ReleaseCollection) WaitToFinish done", zap.Int64("collectionID", req.GetCollectionID()))
|
||||
}()
|
||||
|
||||
status := &commonpb.Status{
|
||||
log.Info("unsubDmChannel(ReleaseCollection) WaitToFinish done", zap.Int64("collectionID", req.GetCollectionID()))
|
||||
return &commonpb.Status{
|
||||
ErrorCode: commonpb.ErrorCode_Success,
|
||||
}
|
||||
return status, nil
|
||||
}, nil
|
||||
}
|
||||
|
||||
// LoadSegments load historical data into query node, historical data can be vector data or index
|
||||
|
|
|
@ -212,6 +212,54 @@ func TestImpl_UnsubDmChannel(t *testing.T) {
|
|||
node, err := genSimpleQueryNode(ctx)
|
||||
assert.NoError(t, err)
|
||||
|
||||
t.Run("normal run", func(t *testing.T) {
|
||||
schema := genTestCollectionSchema()
|
||||
req := &queryPb.WatchDmChannelsRequest{
|
||||
Base: &commonpb.MsgBase{
|
||||
MsgType: commonpb.MsgType_WatchDmChannels,
|
||||
MsgID: rand.Int63(),
|
||||
TargetID: node.session.ServerID,
|
||||
},
|
||||
NodeID: 0,
|
||||
CollectionID: defaultCollectionID,
|
||||
PartitionIDs: []UniqueID{defaultPartitionID},
|
||||
Schema: schema,
|
||||
Infos: []*datapb.VchannelInfo{
|
||||
{
|
||||
CollectionID: 1000,
|
||||
ChannelName: "1000-dmc0",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
status, err := node.WatchDmChannels(ctx, req)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, commonpb.ErrorCode_Success, status.ErrorCode)
|
||||
|
||||
{
|
||||
req := &queryPb.UnsubDmChannelRequest{
|
||||
Base: &commonpb.MsgBase{
|
||||
MsgType: commonpb.MsgType_UnsubDmChannel,
|
||||
MsgID: rand.Int63(),
|
||||
TargetID: node.session.ServerID,
|
||||
},
|
||||
NodeID: 0,
|
||||
CollectionID: defaultCollectionID,
|
||||
ChannelName: "1000-dmc0",
|
||||
}
|
||||
originMetaReplica := node.metaReplica
|
||||
node.metaReplica = newMockReplicaInterface()
|
||||
status, err := node.UnsubDmChannel(ctx, req)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, commonpb.ErrorCode_UnexpectedError, status.ErrorCode)
|
||||
|
||||
node.metaReplica = originMetaReplica
|
||||
status, err = node.UnsubDmChannel(ctx, req)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, commonpb.ErrorCode_Success, status.ErrorCode)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("target not match", func(t *testing.T) {
|
||||
req := &queryPb.UnsubDmChannelRequest{
|
||||
Base: &commonpb.MsgBase{
|
||||
|
|
Loading…
Reference in New Issue