milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix rmq (#6147) 

Signed-off-by: yefu.chen <yefu.chen@zilliz.com>
pull/6149/head^2
neza2017 2021-06-26 22:26:20 +08:00 committed by GitHub
parent b04f08b0c4
commit b22ab71222
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 10 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
internal/msgstream/mq_factory.go
View File

@ -99,9 +99,11 @@ func (f *RmsFactory) NewTtMsgStream(ctx context.Context) (MsgStream, error) {
}
func (f *RmsFactory) NewQueryMsgStream(ctx context.Context) (MsgStream, error) {
InitMmq()
return NewMemMsgStream(ctx, f.ReceiveBufSize)
rmqClient, err := mqclient.NewRmqClient(rocksmq.ClientOptions{Server: rocksmqserver.Rmq})
if err != nil {
return nil, err
}
return NewMqMsgStream(ctx, f.ReceiveBufSize, f.RmqBufSize, rmqClient, f.dispatcherFactory.NewUnmarshalDispatcher())
}
func NewRmsFactory(rocksmqPath string) Factory {

7
internal/rootcoord/meta_table.go
View File

@ -451,16 +451,19 @@ func (mt *metaTable) GetCollectionBySegmentID(segID typeutil.UniqueID) (*pb.Coll
func (mt *metaTable) ListCollections(ts typeutil.Timestamp) (map[string]typeutil.UniqueID, error) {
mt.ddLock.RLock()
defer mt.ddLock.RUnlock()
colls := make(map[string]typeutil.UniqueID)
if ts == 0 {
return mt.collName2ID, nil
for k, v := range mt.collName2ID {
colls[k] = v
}
return colls, nil
}
_, vals, err := mt.client.LoadWithPrefix(CollectionMetaPrefix, ts)
if err != nil {
log.Debug("load with prefix error", zap.Uint64("timestamp", ts), zap.Error(err))
return nil, nil
}
colls := make(map[string]typeutil.UniqueID)
for _, val := range vals {
collMeta := pb.CollectionInfo{}
err := proto.UnmarshalText(val, &collMeta)