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

Remove QueryMsgStream in MqFactory interface (#26374) 

Signed-off-by: Enwei Jiao <enwei.jiao@zilliz.com>
pull/26372/head
Enwei Jiao 2023-08-16 10:29:33 +08:00 committed by GitHub
parent 4742049ecf
commit 78bc688d16
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 7 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/developer_guides/chap04_message_stream.md
View File

@ -216,7 +216,6 @@ type Factory interface {
Init(params *paramtable.ComponentParam) error
NewMsgStream(ctx context.Context) (MsgStream, error)
NewTtMsgStream(ctx context.Context) (MsgStream, error)
NewQueryMsgStream(ctx context.Context) (MsgStream, error)
}
// Pulsar

4
internal/datanode/flow_graph_dmstream_input_node_test.go
View File

@ -56,10 +56,6 @@ func (mm *mockMsgStreamFactory) NewTtMsgStream(ctx context.Context) (msgstream.M
return &mockTtMsgStream{}, nil
}
func (mm *mockMsgStreamFactory) NewQueryMsgStream(ctx context.Context) (msgstream.MsgStream, error) {
return nil, nil
}
func (mm *mockMsgStreamFactory) NewMsgStreamDisposer(ctx context.Context) func([]string, string) error {
return nil
}

5
internal/indexnode/chunkmgr_mock.go
View File

@ -245,11 +245,6 @@ func (f *mockFactory) NewTtMsgStream(context.Context) (msgstream.MsgStream, erro
return nil, errNotImplErr
}
func (f *mockFactory) NewQueryMsgStream(context.Context) (msgstream.MsgStream, error) {
// TODO
return nil, errNotImplErr
}
func (f *mockFactory) NewMsgStreamDisposer(ctx context.Context) func([]string, string) error {
// TODO
return nil

5
internal/mq/msgstream/mq_factory_test.go
View File

@ -41,9 +41,4 @@ func TestRmsFactory(t *testing.T) {
_, err = rmsFactory.NewTtMsgStream(ctx)
assert.NoError(t, err)
_, err = rmsFactory.NewQueryMsgStream(ctx)
assert.NoError(t, err)
err = rmsFactory.NewMsgStreamDisposer(ctx)([]string{"hello"}, "xx")
assert.NoError(t, err)
}

23
internal/proxy/channels_mgr.go
View File

@ -113,21 +113,12 @@ func getDmlChannelsFunc(ctx context.Context, rc types.RootCoord) getChannelsFunc
}
}
// streamType indicates which type of message stream should be created.
type streamType int
const (
dmlStreamType streamType = iota
dqlStreamType
)
type singleTypeChannelsMgr struct {
infos map[UniqueID]streamInfos // collection id -> stream infos
mu sync.RWMutex
getChannelsFunc getChannelsFuncType
repackFunc repackFuncType
singleStreamType streamType
msgStreamFactory msgstream.Factory
}
@ -184,15 +175,11 @@ func (mgr *singleTypeChannelsMgr) streamExistPrivate(collectionID UniqueID) bool
return ok && streamInfos.stream != nil
}
func createStream(factory msgstream.Factory, streamType streamType, pchans []pChan, repack repackFuncType) (msgstream.MsgStream, error) {
func createStream(factory msgstream.Factory, pchans []pChan, repack repackFuncType) (msgstream.MsgStream, error) {
var stream msgstream.MsgStream
var err error
if streamType == dqlStreamType {
stream, err = factory.NewQueryMsgStream(context.Background())
} else {
stream, err = factory.NewMsgStream(context.Background())
}
stream, err = factory.NewMsgStream(context.Background())
if err != nil {
return nil, err
@ -240,7 +227,7 @@ func (mgr *singleTypeChannelsMgr) createMsgStream(collectionID UniqueID) (msgstr
return nil, err
}
stream, err := createStream(mgr.msgStreamFactory, mgr.singleStreamType, channelInfos.pchans, mgr.repackFunc)
stream, err := createStream(mgr.msgStreamFactory, channelInfos.pchans, mgr.repackFunc)
if err != nil {
// What if stream created by other goroutines?
log.Error("failed to create message stream", zap.Error(err), zap.Int64("collection", collectionID))
@ -309,13 +296,11 @@ func newSingleTypeChannelsMgr(
getChannelsFunc getChannelsFuncType,
msgStreamFactory msgstream.Factory,
repackFunc repackFuncType,
singleStreamType streamType,
) *singleTypeChannelsMgr {
return &singleTypeChannelsMgr{
infos: make(map[UniqueID]streamInfos),
getChannelsFunc: getChannelsFunc,
repackFunc: repackFunc,
singleStreamType: singleStreamType,
msgStreamFactory: msgStreamFactory,
}
}
@ -355,6 +340,6 @@ func newChannelsMgrImpl(
msgStreamFactory msgstream.Factory,
) *channelsMgrImpl {
return &channelsMgrImpl{
dmlChannelsMgr: newSingleTypeChannelsMgr(getDmlChannelsFunc, msgStreamFactory, dmlRepackFunc, dmlStreamType),
dmlChannelsMgr: newSingleTypeChannelsMgr(getDmlChannelsFunc, msgStreamFactory, dmlRepackFunc),
}
}

9
internal/proxy/channels_mgr_test.go
View File

@ -213,7 +213,7 @@ func Test_createStream(t *testing.T) {
factory.fQStream = func(ctx context.Context) (msgstream.MsgStream, error) {
return nil, errors.New("mock")
}
_, err := createStream(factory, dmlStreamType, nil, nil)
_, err := createStream(factory, nil, nil)
assert.Error(t, err)
})
@ -222,7 +222,7 @@ func Test_createStream(t *testing.T) {
factory.f = func(ctx context.Context) (msgstream.MsgStream, error) {
return nil, errors.New("mock")
}
_, err := createStream(factory, dqlStreamType, nil, nil)
_, err := createStream(factory, nil, nil)
assert.Error(t, err)
})
@ -231,7 +231,7 @@ func Test_createStream(t *testing.T) {
factory.f = func(ctx context.Context) (msgstream.MsgStream, error) {
return newMockMsgStream(), nil
}
_, err := createStream(factory, dmlStreamType, []string{"111"}, func(tsMsgs []msgstream.TsMsg, hashKeys [][]int32) (map[int32]*msgstream.MsgPack, error) {
_, err := createStream(factory, []string{"111"}, func(tsMsgs []msgstream.TsMsg, hashKeys [][]int32) (map[int32]*msgstream.MsgPack, error) {
return nil, nil
})
assert.NoError(t, err)
@ -271,7 +271,6 @@ func Test_singleTypeChannelsMgr_createMsgStream(t *testing.T) {
return channelInfos{vchans: []string{"111", "222"}, pchans: []string{"111"}}, nil
},
msgStreamFactory: factory,
singleStreamType: dmlStreamType,
repackFunc: nil,
}
_, err := m.createMsgStream(100)
@ -289,7 +288,6 @@ func Test_singleTypeChannelsMgr_createMsgStream(t *testing.T) {
return channelInfos{vchans: []string{"111", "222"}, pchans: []string{"111"}}, nil
},
msgStreamFactory: factory,
singleStreamType: dmlStreamType,
repackFunc: nil,
}
stream, err := m.createMsgStream(100)
@ -356,7 +354,6 @@ func Test_singleTypeChannelsMgr_getStream(t *testing.T) {
return channelInfos{vchans: []string{"111", "222"}, pchans: []string{"111"}}, nil
},
msgStreamFactory: factory,
singleStreamType: dmlStreamType,
repackFunc: nil,
}
stream, err := m.getOrCreateStream(100)

7
internal/proxy/mock_msgstream_test.go
View File

@ -58,13 +58,6 @@ func (m *mockMsgStreamFactory) NewTtMsgStream(ctx context.Context) (msgstream.Ms
return nil, errors.New("mock")
}
func (m *mockMsgStreamFactory) NewQueryMsgStream(ctx context.Context) (msgstream.MsgStream, error) {
if m.fQStream != nil {
return m.fQStream(ctx)
}
return nil, errors.New("mock")
}
func newMockMsgStreamFactory() *mockMsgStreamFactory {
return &mockMsgStreamFactory{}
}

4
internal/proxy/mock_test.go
View File

@ -326,10 +326,6 @@ func (factory *simpleMockMsgStreamFactory) NewTtMsgStream(ctx context.Context) (
return newSimpleMockMsgStream(), nil
}
func (factory *simpleMockMsgStreamFactory) NewQueryMsgStream(ctx context.Context) (msgstream.MsgStream, error) {
return newSimpleMockMsgStream(), nil
}
func (factory *simpleMockMsgStreamFactory) NewMsgStreamDisposer(ctx context.Context) func([]string, string) error {
return nil
}

4
internal/util/dependency/factory.go
View File

@ -144,10 +144,6 @@ func (f *DefaultFactory) NewTtMsgStream(ctx context.Context) (msgstream.MsgStrea
return f.msgStreamFactory.NewTtMsgStream(ctx)
}
func (f *DefaultFactory) NewQueryMsgStream(ctx context.Context) (msgstream.MsgStream, error) {
return f.msgStreamFactory.NewQueryMsgStream(ctx)
}
func (f *DefaultFactory) NewMsgStreamDisposer(ctx context.Context) func([]string, string) error {
return f.msgStreamFactory.NewMsgStreamDisposer(ctx)
}

10
pkg/mq/msgstream/common_mq_factory.go
View File

@ -40,16 +40,6 @@ func (f *CommonFactory) NewTtMsgStream(ctx context.Context) (ms MsgStream, err e
return NewMqTtMsgStream(ctx, f.ReceiveBufSize, f.MQBufSize, cli, f.DispatcherFactory.NewUnmarshalDispatcher())
}
// NewQueryMsgStream is used to generate a new QueryMsgstream object
func (f *CommonFactory) NewQueryMsgStream(ctx context.Context) (ms MsgStream, err error) {
defer wrapError(&err, "NewQueryMsgStream")
cli, err := f.Newer()
if err != nil {
return nil, err
}
return NewMqMsgStream(ctx, f.ReceiveBufSize, f.MQBufSize, cli, f.DispatcherFactory.NewUnmarshalDispatcher())
}
// NewMsgStreamDisposer returns a function that can be used to dispose of a message stream.
// The returned function takes a slice of channel names and a subscription name, and
// disposes of the message stream associated with those arguments.

3
pkg/mq/msgstream/factory_test.go
View File

@ -82,9 +82,6 @@ func testFactoryCommonOperation(t *testing.T, f Factory) {
_, err = f.NewTtMsgStream(ctx)
assert.NoError(t, err)
_, err = f.NewQueryMsgStream(ctx)
assert.NoError(t, err)
err = f.NewMsgStreamDisposer(ctx)([]string{"hello"}, "xx")
assert.NoError(t, err)
}

9
pkg/mq/msgstream/mq_factory.go
View File

@ -110,11 +110,6 @@ func (f *PmsFactory) getAuthentication() (pulsar.Authentication, error) {
return auth, nil
}
// NewQueryMsgStream is used to generate a new QueryMsgstream object
func (f *PmsFactory) NewQueryMsgStream(ctx context.Context) (MsgStream, error) {
return f.NewMsgStream(ctx)
}
func (f *PmsFactory) NewMsgStreamDisposer(ctx context.Context) func([]string, string) error {
return func(channels []string, subname string) error {
// try to delete the old subscription
@ -166,10 +161,6 @@ func (f *KmsFactory) NewTtMsgStream(ctx context.Context) (MsgStream, error) {
return NewMqTtMsgStream(ctx, f.ReceiveBufSize, f.MQBufSize, kafkaClient, f.dispatcherFactory.NewUnmarshalDispatcher())
}
func (f *KmsFactory) NewQueryMsgStream(ctx context.Context) (MsgStream, error) {
return f.NewMsgStream(ctx)
}
func (f *KmsFactory) NewMsgStreamDisposer(ctx context.Context) func([]string, string) error {
return func(channels []string, subname string) error {
msgstream, err := f.NewMsgStream(ctx)

12
pkg/mq/msgstream/mq_factory_test.go
View File

@ -34,9 +34,6 @@ func TestPmsFactory(t *testing.T) {
_, err = pmsFactory.NewTtMsgStream(ctx)
assert.NoError(t, err)
_, err = pmsFactory.NewQueryMsgStream(ctx)
assert.NoError(t, err)
err = pmsFactory.NewMsgStreamDisposer(ctx)([]string{"hello"}, "xx")
assert.NoError(t, err)
}
@ -58,9 +55,6 @@ func TestPmsFactoryWithAuth(t *testing.T) {
_, err = pmsFactory.NewTtMsgStream(ctx)
assert.NoError(t, err)
_, err = pmsFactory.NewQueryMsgStream(ctx)
assert.NoError(t, err)
Params.Save(Params.PulsarCfg.AuthParams.Key, "")
pmsFactory = NewPmsFactory(config)
@ -71,9 +65,6 @@ func TestPmsFactoryWithAuth(t *testing.T) {
_, err = pmsFactory.NewTtMsgStream(ctx)
assert.Error(t, err)
_, err = pmsFactory.NewQueryMsgStream(ctx)
assert.Error(t, err)
}
func TestKafkaFactory(t *testing.T) {
@ -86,9 +77,6 @@ func TestKafkaFactory(t *testing.T) {
_, err = kmsFactory.NewTtMsgStream(ctx)
assert.NoError(t, err)
_, err = kmsFactory.NewQueryMsgStream(ctx)
assert.NoError(t, err)
// err = kmsFactory.NewMsgStreamDisposer(ctx)([]string{"hello"}, "xx")
// assert.NoError(t, err)
}

1
pkg/mq/msgstream/msgstream.go
View File

@ -73,6 +73,5 @@ type MsgStream interface {
type Factory interface {
NewMsgStream(ctx context.Context) (MsgStream, error)
NewTtMsgStream(ctx context.Context) (MsgStream, error)
NewQueryMsgStream(ctx context.Context) (MsgStream, error)
NewMsgStreamDisposer(ctx context.Context) func([]string, string) error
}