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 bug: data race in MsgFactory (#14839)

Browse Source 
Signed-off-by: zhenshan.cao <zhenshan.cao@zilliz.com>
This commit is contained in:
zhenshan.cao 2022-01-06 14:57:20 +08:00 committed by GitHub
parent 92b8eddb5c
commit a5f8bdeb42
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 14 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
internal/datanode/data_node.go
View File

@ -220,9 +220,19 @@ func (node *DataNode) Init() error {
}
Params.DataNodeCfg.Refresh()
m := map[string]interface{}{
"PulsarAddress": Params.DataNodeCfg.PulsarAddress,
"ReceiveBufSize": 1024,
"PulsarBufSize": 1024,
}
if err := node.msFactory.SetParams(m); err != nil {
log.Warn("DataNode Init msFactory SetParams failed, use default",
zap.Error(err))
return err
}
log.Debug("DataNode Init",
zap.String("MsgChannelSubName", Params.DataNodeCfg.MsgChannelSubName),
)
zap.String("MsgChannelSubName", Params.DataNodeCfg.MsgChannelSubName))
return nil
}

13
internal/datanode/data_sync_service.go
View File

@ -137,18 +137,6 @@ func (dsService *dataSyncService) close() {
// initNodes inits a TimetickedFlowGraph
func (dsService *dataSyncService) initNodes(vchanInfo *datapb.VchannelInfo) error {
dsService.fg = flowgraph.NewTimeTickedFlowGraph(dsService.ctx)
m := map[string]interface{}{
"PulsarAddress": Params.DataNodeCfg.PulsarAddress,
"ReceiveBufSize": 1024,
"PulsarBufSize": 1024,
}
err := dsService.msFactory.SetParams(m)
if err != nil {
return err
}
// initialize flush manager for DataSync Service
dsService.flushManager = NewRendezvousFlushManager(dsService.idAllocator, dsService.blobKV, dsService.replica,
flushNotifyFunc(dsService), dropVirtualChannelFunc(dsService))
@ -216,6 +204,7 @@ func (dsService *dataSyncService) initNodes(vchanInfo *datapb.VchannelInfo) erro
parallelConfig: newParallelConfig(),
}
var err error
var dmStreamNode Node
dmStreamNode, err = newDmInputNode(dsService.ctx, vchanInfo.GetSeekPosition(), c)
if err != nil {

2
internal/datanode/data_sync_service_test.go
View File

@ -96,7 +96,7 @@ func TestDataSyncService_newDataSyncService(te *testing.T) {
ctx := context.Background()
tests := []*testInfo{
{false, false, &mockMsgStreamFactory{false, true},
{true, false, &mockMsgStreamFactory{false, true},
0, "by-dev-rootcoord-dml-test_v0",
0, 0, "", 0,
0, 0, "", 0,