fix: unit test related to kafka

zilliztech · Oct 15, 2024 · f55afd3 · f55afd3
1 parent b995773
commit f55afd3
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 2 deletions.
diff --git a/core/reader/replicate_channel_manager_test.go b/core/reader/replicate_channel_manager_test.go
@@ -421,7 +421,6 @@ func TestStartReadCollectionForKafka(t *testing.T) {
 	})
 
 	realManager := manager.(*replicateChannelManager)
-
 	stream := msgstream.NewMockMsgStream(t)
 	streamChan := make(chan *msgstream.MsgPack)
 
@@ -479,7 +478,7 @@ func TestStartReadCollectionForKafka(t *testing.T) {
 	})
 
 	t.Run("collection and partition", func(t *testing.T) {
-		// start collection
+		// start read collection
 		{
 			err := realManager.StartReadCollection(context.Background(), &pb.CollectionInfo{
 				ID: 31001,
@@ -495,6 +494,8 @@ func TestStartReadCollectionForKafka(t *testing.T) {
 				VirtualChannelNames:  []string{"collection-partition-p1_v0"},
 			}, nil)
 			assert.NoError(t, err)
+			event := <-realManager.GetEventChan()
+			assert.Equal(t, api.ReplicateCreateCollection, event.EventType)
 		}
 
 		// partition not found

diff --git a/server/model/request/base.go b/server/model/request/base.go
@@ -59,9 +59,12 @@ func GetTask(taskInfo *meta.TaskInfo) Task {
 	taskInfo.MilvusConnectParam.Username = ""
 	taskInfo.MilvusConnectParam.Password = ""
 	taskInfo.MilvusConnectParam.Token = ""
+	taskInfo.KafkaConnectParam.SASL.Username = ""
+	taskInfo.KafkaConnectParam.SASL.Password = ""
 	return Task{
 		TaskID:             taskInfo.TaskID,
 		MilvusConnectParam: taskInfo.MilvusConnectParam,
+		KafkaConnectParam:  taskInfo.KafkaConnectParam,
 		CollectionInfos:    taskInfo.CollectionInfos,
 		State:              taskInfo.State.String(),
 		LastPauseReason:    taskInfo.Reason,