fix: use pulsar for streaming it

Signed-off-by: chyezh <[email protected]>
milvus-io · Jan 25, 2025 · c8502dd · c8502dd
1 parent 4ae22b5
commit c8502dd
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 1 deletion.
diff --git a/tests/integration/minicluster_v2.go b/tests/integration/minicluster_v2.go
@@ -138,6 +138,12 @@ type MiniClusterV2 struct {
 
 type OptionV2 func(cluster *MiniClusterV2)
 
+func OptionSetParam(k string, v string) OptionV2 {
+	return func(cluster *MiniClusterV2) {
+		cluster.params[k] = v
+	}
+}
+
 func StartMiniClusterV2(ctx context.Context, opts ...OptionV2) (*MiniClusterV2, error) {
 	cluster := &MiniClusterV2{
 		ctx:  ctx,

diff --git a/tests/integration/streaming/hello_streaming_test.go b/tests/integration/streaming/hello_streaming_test.go
@@ -46,6 +46,9 @@ type HelloStreamingSuite struct {
 
 func (s *HelloStreamingSuite) SetupSuite() {
 	streamingutil.SetStreamingServiceEnabled()
+	s.MiniClusterSuite.Opt = []integration.OptionV2{
+		integration.OptionSetParam("mq.type", "pulsar"),
+	}
 	s.MiniClusterSuite.SetupSuite()
 }
 

diff --git a/tests/integration/suite.go b/tests/integration/suite.go
@@ -88,6 +88,7 @@ type MiniClusterSuite struct {
 	EmbedEtcdSuite
 
 	Cluster    *MiniClusterV2
+	Opt        []OptionV2
 	cancelFunc context.CancelFunc
 }
 
@@ -111,10 +112,11 @@ func (s *MiniClusterSuite) SetupTest() {
 	s.T().Log("Setup case timeout", caseTimeout)
 	ctx, cancel := context.WithTimeout(context.Background(), caseTimeout)
 	s.cancelFunc = cancel
-	c, err := StartMiniClusterV2(ctx, func(c *MiniClusterV2) {
+	opts := append(s.Opt, func(c *MiniClusterV2) {
 		// change config etcd endpoints
 		c.params[params.EtcdCfg.Endpoints.Key] = val
 	})
+	c, err := StartMiniClusterV2(ctx, opts...)
 	s.Require().NoError(err)
 	s.Cluster = c