Repository: lf-edge/ekuiper
Pull request #3055 (2024-07-25 03:13:27)

fix(rest): do not override content type (#3024)

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3055?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 67.94%. Comparing base [(`c11b697`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/c11b6970bbedb200441f43caa8a4735e5b0777df?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) to head [(`e067985`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/e067985bb40b68875354e24c5286afb22cdcb006?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #3055      +/-   ##
==========================================
+ Coverage   67.84%   67.94%   +0.10%     
==========================================
  Files         390      390              
  Lines       37135    37130       -5     
==========================================
+ Hits        25191    25225      +34     
+ Misses       9667     9623      -44     
- Partials     2277     2282       +5     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3055?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/internal/io/http/client.go b/internal/io/http/client.go
index 498c38b1..90c37be7 100644
--- a/internal/io/http/client.go
+++ b/internal/io/http/client.go
@@ -74,7 +74,8 @@ type RawConf struct {
 	Timeout     cast.DurationConf `json:"timeout"`
 	Incremental bool              `json:"incremental"`
 
-	OAuth map[string]map[string]interface{} `json:"oauth"`
+	OAuth      map[string]map[string]interface{} `json:"oauth"`
+	SendSingle bool                              `json:"sendSingle"`
 	// Could be code or body
 	ResponseType string `json:"responseType"`
 	Compression  string `json:"compression"` // Compression specifies the algorithms used to payload compression
diff --git a/internal/io/http/rest_sink.go b/internal/io/http/rest_sink.go
index f10df430..7e1e0d24 100644
--- a/internal/io/http/rest_sink.go
+++ b/internal/io/http/rest_sink.go
@@ -44,28 +44,19 @@ func (r *RestSink) Connect(ctx api.StreamContext) error {
 }
 
 func (r *RestSink) Collect(ctx api.StreamContext, item api.MessageTuple) error {
-	return r.collect(ctx, item)
+	return r.collect(ctx, item, item.ToMap())
 }
 
 func (r *RestSink) CollectList(ctx api.StreamContext, items api.MessageTupleList) error {
-	var err error
-	items.RangeOfTuples(func(_ int, tuple api.MessageTuple) bool {
-		err = r.collect(ctx, tuple)
-		if err != nil {
-			return false
-		}
-		return true
-	})
-	return err
+	return r.collect(ctx, items, items.ToMaps())
 }
 
-func (r *RestSink) collect(ctx api.StreamContext, item api.MessageTuple) error {
+func (r *RestSink) collect(ctx api.StreamContext, item any, data any) error {
 	logger := ctx.GetLogger()
 	headers := r.config.Headers
 	bodyType := r.config.BodyType
 	method := r.config.Method
 	u := r.config.Url
-	data := item.ToMap()
 	if dp, ok := item.(api.HasDynamicProps); ok {
 		for k := range headers {
 			nv, ok := dp.DynamicProps(k)
@@ -86,7 +77,7 @@ func (r *RestSink) collect(ctx api.StreamContext, item api.MessageTuple) error {
 			u = nu
 		}
 	}
-	resp, err := httpx.Send(ctx.GetLogger(), r.client, bodyType, method, u, headers, true, data)
+	resp, err := httpx.Send(ctx.GetLogger(), r.client, bodyType, method, u, headers, r.config.SendSingle, data)
 	failpoint.Inject("recoverAbleErr", func() {
 		err = errors.New("connection reset by peer")
 	})
diff --git a/internal/io/mqtt/client/client.go b/internal/io/mqtt/client/client.go
index fa32eef5..860beda0 100644
--- a/internal/io/mqtt/client/client.go
+++ b/internal/io/mqtt/client/client.go
@@ -114,11 +114,10 @@ func (conn *Connection) Close(ctx api.StreamContext) error {
 }
 
 func (conn *Connection) Ping(ctx api.StreamContext) error {
-	if conn.Client.IsConnected() {
+	if conn.connected.Load() {
 		return nil
-	} else {
-		return errors.New("mqtt ping failed")
 	}
+	return errors.New("failed to connect to broker")
 }
 
 func CreateConnection(ctx api.StreamContext, props map[string]any) (modules.Connection, error) {
diff --git a/internal/pkg/httpx/http.go b/internal/pkg/httpx/http.go
index b2f458a1..531f855b 100644
--- a/internal/pkg/httpx/http.go
+++ b/internal/pkg/httpx/http.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -63,7 +63,9 @@ func Send(logger api.Logger, client *http.Client, bodyType string, method string
 		if err != nil {
 			return nil, fmt.Errorf("fail to create request: %v", err)
 		}
-		req.Header.Set("Content-Type", BodyTypeMap[bodyType])
+		if req.Header.Get("Content-Type") == "" {
+			req.Header.Set("Content-Type", BodyTypeMap[bodyType])
+		}
 	case "form":
 		form := url.Values{}
 		im, err := convertToMap(v, sendSingle)
@@ -89,7 +91,9 @@ func Send(logger api.Logger, client *http.Client, bodyType string, method string
 		if err != nil {
 			return nil, fmt.Errorf("fail to create request: %v", err)
 		}
-		req.Header.Set("Content-Type", "application/x-www-form-urlencoded;param=value")
+		if req.Header.Get("Content-Type") == "" {
+			req.Header.Set("Content-Type", "application/x-www-form-urlencoded;param=value")
+		}
 	default:
 		return nil, fmt.Errorf("unsupported body type %s", bodyType)
 	}
diff --git a/internal/plugin/portable/manager.go b/internal/plugin/portable/manager.go
index fb54c421..6f6b9f53 100644
--- a/internal/plugin/portable/manager.go
+++ b/internal/plugin/portable/manager.go
@@ -422,6 +422,7 @@ func (m *Manager) Delete(name string) error {
 	if err != nil {
 		return fmt.Errorf("fail to kill portable plugin %s process, please try to kill it manually", name)
 	}
+	pm.DeletePluginIns(name)
 	return nil
 }
 
diff --git a/internal/plugin/portable/runtime/plugin_ins_manager.go b/internal/plugin/portable/runtime/plugin_ins_manager.go
index 046660a5..6e8e8fb4 100644
--- a/internal/plugin/portable/runtime/plugin_ins_manager.go
+++ b/internal/plugin/portable/runtime/plugin_ins_manager.go
@@ -22,7 +22,6 @@ import (
 	"os/exec"
 	"path/filepath"
 	"sync"
-	"time"
 
 	"github.com/pingcap/failpoint"
 
@@ -168,28 +167,11 @@ func (i *PluginIns) StopSymbol(ctx api.StreamContext, ctrl *Control) error {
 	}
 	err = i.sendCmd(jsonArg)
 	if err == nil {
-		referred := false
 		i.Lock()
 		delete(i.commands, ctrl.Meta)
 		i.deRef(ctx)
 		i.Unlock()
 		ctx.GetLogger().Infof("stopped symbol %s", ctrl.SymbolName)
-		if !referred {
-			go func() {
-				// delay to kill the plugin process
-				time.Sleep(1 * time.Second)
-				i.RLock()
-				defer i.RUnlock()
-				if len(i.commands) == 0 {
-					err := GetPluginInsManager().Kill(i.name)
-					if err != nil {
-						ctx.GetLogger().Errorf("fail to stop plugin %s: %v", i.name, err)
-						return
-					}
-					ctx.GetLogger().Infof("stop plugin %s", i.name)
-				}
-			}()
-		}
 	}
 	return err
 }
@@ -256,6 +238,16 @@ func (p *pluginInsManager) GetPluginInsStatus(name string) (*PluginStatus, bool)
 	return ins.GetStatus(), true
 }
 
+func (p *pluginInsManager) GetAllPluginStatus() map[string]*PluginStatus {
+	want := make(map[string]*PluginStatus)
+	p.RLock()
+	defer p.RUnlock()
+	for key, plugin := range p.instances {
+		want[key] = plugin.GetStatus()
+	}
+	return want
+}
+
 func (p *pluginInsManager) GetPluginIns(name string) (*PluginIns, bool) {
 	p.RLock()
 	defer p.RUnlock()
@@ -331,16 +323,15 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 	}
 	// should only happen for first start, then the ctrl channel will keep running
 	if ins.ctrlChan == nil {
-		conf.Log.Infof("create control channel")
 		ctrlChan, err := CreateControlChannel(pluginMeta.Name)
 		if err != nil {
+			conf.Log.Errorf("plugin %s can't create new control channel: %s", pluginMeta.Name, err.Error())
 			ins.Status.StatusErr(err)
-			return nil, fmt.Errorf("can't create new control channel: %s", err.Error())
+			return nil, fmt.Errorf("plugin %s can't create new control channel: %s", pluginMeta.Name, err.Error())
 		}
 		ins.ctrlChan = ctrlChan
 	}
 	// init or restart all need to run the process
-	conf.Log.Infof("executing plugin")
 	jsonArg, err := json.Marshal(pconf)
 	failpoint.Inject("confErr", func() {
 		err = errors.New("confErr")
@@ -383,7 +374,6 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 	cmd.Stderr = conf.Log.Out
 	cmd.Dir = filepath.Dir(pluginMeta.Executable)
 
-	conf.Log.Println("plugin starting")
 	err = cmd.Start()
 	failpoint.Inject("cmdStartErr", func() {
 		cmd.Process.Kill()
@@ -391,12 +381,14 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 	})
 	if err != nil {
 		ins.Status.StatusErr(err)
-		return nil, fmt.Errorf("plugin executable %s stops with error %v", pluginMeta.Executable, err)
+		conf.Log.Errorf("plugin %s executable %s stops with error %v", pluginMeta.Name, pluginMeta.Executable, err)
+		return nil, fmt.Errorf("plugin %s executable %s stops with error %v", pluginMeta.Name, pluginMeta.Executable, err)
 	}
 	process := cmd.Process
-	conf.Log.Printf("plugin started pid: %d\n", process.Pid)
+	conf.Log.Printf("plugin %s started pid: %d\n", pluginMeta.Name, process.Pid)
 	defer func() {
 		if e != nil {
+			ins.Status.StatusErr(e)
 			_ = process.Kill()
 		}
 	}()
@@ -417,7 +409,6 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 			}
 			ins.process = nil
 			ins.Unlock()
-			p.DeletePluginIns(pluginMeta.Name)
 		}
 		return nil
 	})
@@ -429,10 +420,10 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 	}
 	ins.process = process
 	p.instances[pluginMeta.Name] = ins
-	conf.Log.Println("plugin start running")
-	ins.Status.StartRunning()
+	ins.Status.StartRunning(ins.process.Pid)
+	conf.Log.Infof("plugin %s start running", pluginMeta.Name)
 	// restore symbols by sending commands when restarting plugin
-	conf.Log.Info("restore plugin symbols")
+	conf.Log.Infof("restore plugin %s symbols", pluginMeta.Name)
 	for m, c := range ins.commands {
 		go func(key Meta, jsonArg []byte) {
 			e := ins.sendCmd(jsonArg)
@@ -488,6 +479,7 @@ type PluginStatus struct {
 	RefCount map[string]int `json:"refCount"`
 	Status   string         `json:"status"`
 	ErrMsg   string         `json:"errMsg"`
+	Pid      int            `json:"pid"`
 }
 
 func NewPluginStatus() *PluginStatus {
@@ -502,8 +494,9 @@ func (s *PluginStatus) StatusErr(err error) {
 	s.ErrMsg = err.Error()
 }
 
-func (s *PluginStatus) StartRunning() {
+func (s *PluginStatus) StartRunning(pid int) {
 	s.Status = PluginStatusRunning
+	s.Pid = pid
 	s.ErrMsg = ""
 }
 
diff --git a/internal/server/server.go b/internal/server/server.go
index ffc6743e..6c4a8a5d 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -44,6 +44,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/schedule"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store/definition"
+	"github.com/lf-edge/ekuiper/v2/internal/plugin/portable/runtime"
 	"github.com/lf-edge/ekuiper/v2/internal/processor"
 	"github.com/lf-edge/ekuiper/v2/internal/server/bump"
 	"github.com/lf-edge/ekuiper/v2/internal/server/promMetrics"
@@ -287,9 +288,7 @@ func StartUp(Version string) {
 	wg.Add(2)
 	go func() {
 		conf.Log.Info("start to stop all rules")
-		if err := waitAllRuleStop(ctx); err != nil {
-			conf.Log.Warnf(err.Error())
-		}
+		waitAllRuleStop()
 		wg.Done()
 	}()
 	go func() {
@@ -301,6 +300,8 @@ func StartUp(Version string) {
 		wg.Done()
 	}()
 	wg.Wait()
+	// kill all plugin process
+	runtime.GetPluginInsManager().KillAll()
 
 	// close extend services
 	for k, v := range servers {
@@ -503,42 +504,9 @@ func startCPUProfiling(ctx context.Context) error {
 	return nil
 }
 
-func waitAllRuleStop(ctx context.Context) error {
+func waitAllRuleStop() {
 	rules, _ := ruleProcessor.GetAllRules()
 	for _, r := range rules {
 		stopRuleWhenServerStop(r)
 	}
-	wg := &sync.WaitGroup{}
-	m := &sync.Map{}
-	for _, r := range rules {
-		rs, ok := registry.Load(r)
-		if ok {
-			m.Store(r, struct{}{})
-			wg.Add(1)
-			go func() {
-				defer func() {
-					m.Delete(r)
-					wg.Done()
-				}()
-				rs.Topology.WaitClose()
-			}()
-		}
-	}
-	wgCh := make(chan struct{})
-	go func() {
-		wg.Wait()
-		wgCh <- struct{}{}
-	}()
-
-	select {
-	case <-ctx.Done():
-		timeoutRules := make([]string, 0)
-		m.Range(func(key, value any) bool {
-			timeoutRules = append(timeoutRules, key.(string))
-			return true
-		})
-		return fmt.Errorf("stop rules timeout, remain::%s", timeoutRules)
-	case <-wgCh:
-		return nil
-	}
 }
diff --git a/internal/topo/context/default.go b/internal/topo/context/default.go
index cb8afee8..16eb72f0 100644
--- a/internal/topo/context/default.go
+++ b/internal/topo/context/default.go
@@ -51,9 +51,6 @@ type DefaultContext struct {
 	// cache
 	tpReg sync.Map
 	jpReg sync.Map
-
-	// ops waitGroup
-	opsWg *sync.WaitGroup
 }
 
 func RuleBackground(ruleName string) *DefaultContext {
@@ -87,11 +84,6 @@ func WithValue(parent *DefaultContext, key, val interface{}) *DefaultContext {
 	return parent
 }
 
-func WithWg(parent *DefaultContext, wg *sync.WaitGroup) *DefaultContext {
-	parent.opsWg = wg
-	return parent
-}
-
 // Deadline Implement context interface
 func (c *DefaultContext) Deadline() (deadline time.Time, ok bool) {
 	return c.ctx.Deadline()
@@ -212,7 +204,6 @@ func (c *DefaultContext) WithMeta(ruleId string, opId string, store api.Store) a
 		state:      s,
 		tpReg:      sync.Map{},
 		jpReg:      sync.Map{},
-		opsWg:      c.opsWg,
 	}
 }
 
@@ -223,7 +214,6 @@ func (c *DefaultContext) WithInstance(instanceId int) api.StreamContext {
 		opId:       c.opId,
 		ctx:        c.ctx,
 		state:      c.state,
-		opsWg:      c.opsWg,
 	}
 }
 
@@ -235,7 +225,6 @@ func (c *DefaultContext) WithCancel() (api.StreamContext, context.CancelFunc) {
 		instanceId: c.instanceId,
 		ctx:        ctx,
 		state:      c.state,
-		opsWg:      c.opsWg,
 	}, cancel
 }
 
diff --git a/internal/topo/node/operations.go b/internal/topo/node/operations.go
index b898568f..5ef49c04 100644
--- a/internal/topo/node/operations.go
+++ b/internal/topo/node/operations.go
@@ -15,6 +15,10 @@
 package node
 
 import (
+	"time"
+
+	"github.com/pingcap/failpoint"
+
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
@@ -61,6 +65,9 @@ func (o *UnaryOperator) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}
 	go func() {
 		defer func() {
+			failpoint.Inject("mockTimeConsumingClose", func() {
+				time.Sleep(300 * time.Millisecond)
+			})
 			o.Close()
 		}()
 		err := infra.SafeRun(func() error {
diff --git a/internal/topo/node/window_op.go b/internal/topo/node/window_op.go
index 65aa3c97..45f644ed 100644
--- a/internal/topo/node/window_op.go
+++ b/internal/topo/node/window_op.go
@@ -150,11 +150,11 @@ func (o *WindowOperator) Exec(ctx api.StreamContext, errCh chan<- error) {
 		}
 	}
 	log.Infof("Start with window state triggerTime: %d, msgCount: %d", o.triggerTime.UnixMilli(), o.msgCount)
-	if o.isEventTime {
-		go func() {
-			defer func() {
-				o.Close()
-			}()
+	go func() {
+		defer func() {
+			o.Close()
+		}()
+		if o.isEventTime {
 			err := infra.SafeRun(func() error {
 				o.execEventWindow(ctx, inputs, errCh)
 				return nil
@@ -162,9 +162,7 @@ func (o *WindowOperator) Exec(ctx api.StreamContext, errCh chan<- error) {
 			if err != nil {
 				infra.DrainError(ctx, err, errCh)
 			}
-		}()
-	} else {
-		go func() {
+		} else {
 			err := infra.SafeRun(func() error {
 				o.execProcessingWindow(ctx, inputs, errCh)
 				return nil
@@ -172,8 +170,8 @@ func (o *WindowOperator) Exec(ctx api.StreamContext, errCh chan<- error) {
 			if err != nil {
 				infra.DrainError(ctx, err, errCh)
 			}
-		}()
-	}
+		}
+	}()
 }
 
 func getAlignedWindowEndTime(n time.Time, interval int, timeUnit ast.Token) time.Time {
diff --git a/internal/topo/rule/ruleState.go b/internal/topo/rule/ruleState.go
index 434143cb..fe8d7591 100644
--- a/internal/topo/rule/ruleState.go
+++ b/internal/topo/rule/ruleState.go
@@ -482,6 +482,8 @@ func (rs *RuleState) stop() error {
 	rs.triggered = 0
 	if rs.Topology != nil {
 		rs.Topology.Cancel()
+		// wait all operator close
+		rs.Topology.WaitClose()
 		// de-reference old Topology in order to release data memory
 		rs.Topology = rs.Topology.NewTopoWithSucceededCtx()
 	}
@@ -542,6 +544,7 @@ func (rs *RuleState) Close() (err error) {
 	}
 	if rs.triggered == 1 && rs.Topology != nil {
 		rs.Topology.Cancel()
+		rs.Topology.WaitClose()
 	}
 	rs.triggered = -1
 	rs.stopScheduleRule()
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index 0d62578f..8f96d7fb 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -233,7 +233,7 @@ func (s *Topo) prepareContext() {
 		}
 		ctx := kctx.WithValue(kctx.RuleBackground(s.name), kctx.LoggerKey, contextLogger)
 		ctx = kctx.WithValue(ctx, kctx.RuleStartKey, timex.GetNowInMilli())
-		ctx = kctx.WithWg(ctx, s.opsWg)
+		ctx = kctx.WithValue(ctx, kctx.RuleWaitGroupKey, s.opsWg)
 		s.ctx, s.cancel = ctx.WithCancel()
 	}
 }
@@ -257,19 +257,19 @@ func (s *Topo) Open() <-chan error {
 		if err := s.enableCheckpoint(s.ctx); err != nil {
 			return err
 		}
+		topoStore := s.store
 		// open stream sink, after log sink is ready.
 		for _, snk := range s.sinks {
-			snk.Exec(s.ctx.WithMeta(s.name, snk.GetName(), s.store), s.drain)
+			snk.Exec(s.ctx.WithMeta(s.name, snk.GetName(), topoStore), s.drain)
 		}
 
 		for _, op := range s.ops {
-			op.Exec(s.ctx.WithMeta(s.name, op.GetName(), s.store), s.drain)
+			op.Exec(s.ctx.WithMeta(s.name, op.GetName(), topoStore), s.drain)
 		}
 
 		for _, source := range s.sources {
-			source.Open(s.ctx.WithMeta(s.name, source.GetName(), s.store), s.drain)
+			source.Open(s.ctx.WithMeta(s.name, source.GetName(), topoStore), s.drain)
 		}
-
 		// activate checkpoint
 		if s.coordinator != nil {
 			return s.coordinator.Activate()
diff --git a/pkg/connection/connection_pool.go b/pkg/connection/connection_pool.go
index ecddbae0..bdf5cf58 100644
--- a/pkg/connection/connection_pool.go
+++ b/pkg/connection/connection_pool.go
@@ -69,25 +69,39 @@ func GetConnectionRef(id string) int {
 }
 
 func GetAllConnectionStatus(ctx api.StreamContext) map[string]ConnectionStatus {
+	cws := make(map[string]*ConnWrapper)
 	globalConnectionManager.RLock()
-	defer globalConnectionManager.RUnlock()
-	s := make(map[string]ConnectionStatus)
 	for id, meta := range globalConnectionManager.connectionPool {
-		status := ConnectionStatus{
-			Status: ConnectionRunning,
-		}
-		conn, err := meta.cw.Wait()
-		if err != nil {
-			status.Status = ConnectionFail
-			status.ErrMsg = err.Error()
-		} else {
+		cws[id] = meta.cw
+	}
+	globalConnectionManager.RUnlock()
+	s := make(map[string]ConnectionStatus)
+	for id, cw := range cws {
+		if cw.IsInitialized() {
+			conn, err := cw.Wait()
+			if err != nil {
+				s[id] = ConnectionStatus{
+					Status: ConnectionFail,
+					ErrMsg: err.Error(),
+				}
+				continue
+			}
 			err = conn.Ping(ctx)
 			if err != nil {
-				status.Status = ConnectionFail
-				status.ErrMsg = err.Error()
+				s[id] = ConnectionStatus{
+					Status: ConnectionFail,
+					ErrMsg: err.Error(),
+				}
+				continue
+			}
+			s[id] = ConnectionStatus{
+				Status: ConnectionRunning,
+			}
+		} else {
+			s[id] = ConnectionStatus{
+				Status: ConnectionIntializing,
 			}
 		}
-		s[id] = status
 	}
 	return s
 }
@@ -371,13 +385,14 @@ const (
 var DefaultBackoffMaxElapsedDuration = 3 * time.Minute
 
 const (
-	ConnectionRunning = "running"
-	ConnectionFail    = "fail"
+	ConnectionRunning     = "running"
+	ConnectionIntializing = "initializing"
+	ConnectionFail        = "fail"
 )
 
 type ConnectionStatus struct {
-	Status string
-	ErrMsg string
+	Status string `json:"status"`
+	ErrMsg string `json:"errMsg,omitempty"`
 }
 
 func extractSelID(props map[string]interface{}) string {
@@ -419,6 +434,12 @@ func (cw *ConnWrapper) Wait() (modules.Connection, error) {
 	return cw.conn, cw.err
 }
 
+func (cw *ConnWrapper) IsInitialized() bool {
+	cw.cond.L.Lock()
+	defer cw.cond.L.Unlock()
+	return cw.initialized
+}
+
 func newConnWrapper(ctx api.StreamContext, meta *ConnectionMeta) *ConnWrapper {
 	cw := &ConnWrapper{
 		ID:   meta.ID,


Test patch:
diff --git a/internal/io/http/rest_sink_test.go b/internal/io/http/rest_sink_test.go
index 1191958f..c06387cd 100644
--- a/internal/io/http/rest_sink_test.go
+++ b/internal/io/http/rest_sink_test.go
@@ -16,9 +16,13 @@ package http
 
 import (
 	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"github.com/pingcap/failpoint"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
@@ -26,6 +30,215 @@ import (
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
+type request struct {
+	Method      string
+	Body        string
+	ContentType string
+}
+
+func TestRestSink_Apply(t *testing.T) {
+	tests := []struct {
+		name   string
+		config map[string]interface{}
+		data   []map[string]interface{}
+		result []request
+	}{
+		{
+			name: "1",
+			config: map[string]interface{}{
+				"method": "post",
+				//"url": "http://localhost/test",  //set dynamically to the test server
+				"sendSingle": true,
+			},
+			data: []map[string]interface{}{{
+				"ab": "hello1",
+			}, {
+				"ab": "hello2",
+			}},
+			result: []request{{
+				Method:      "POST",
+				Body:        `{"ab":"hello1"}`,
+				ContentType: "application/json",
+			}, {
+				Method:      "POST",
+				Body:        `{"ab":"hello2"}`,
+				ContentType: "application/json",
+			}},
+		}, {
+			name: "2",
+			config: map[string]interface{}{
+				"method": "post",
+				//"url": "http://localhost/test",  //set dynamically to the test server
+				"headers": map[string]any{
+					"Content-Type": "application/vnd.microsoft.servicebus.json",
+				},
+			},
+			data: []map[string]interface{}{{
+				"ab": "hello1",
+			}, {
+				"ab": "hello2",
+			}},
+			result: []request{{
+				Method:      "POST",
+				Body:        `[{"ab":"hello1"},{"ab":"hello2"}]`,
+				ContentType: "application/vnd.microsoft.servicebus.json",
+			}},
+		}, {
+			name: "3",
+			config: map[string]interface{}{
+				"method": "get",
+				//"url": "http://localhost/test",  //set dynamically to the test server
+			},
+			data: []map[string]interface{}{{
+				"ab": "hello1",
+			}, {
+				"ab": "hello2",
+			}},
+			result: []request{{
+				Method:      "GET",
+				ContentType: "",
+			}},
+		}, {
+			name: "4",
+			config: map[string]interface{}{
+				"method": "put",
+				//"url": "http://localhost/test",  //set dynamically to the test server
+				"bodyType": "text",
+			},
+			data: []map[string]interface{}{{
+				"ab": "hello1",
+			}, {
+				"ab": "hello2",
+			}},
+			result: []request{{
+				Method:      "PUT",
+				ContentType: "text/plain",
+				Body:        `[{"ab":"hello1"},{"ab":"hello2"}]`,
+			}},
+		}, {
+			name: "5",
+			config: map[string]interface{}{
+				"method": "post",
+				//"url": "http://localhost/test",  //set dynamically to the test server
+				"bodyType": "form",
+			},
+			data: []map[string]interface{}{{
+				"ab": "hello1",
+			}, {
+				"ab": "hello2",
+			}},
+			result: []request{{
+				Method:      "POST",
+				ContentType: "application/x-www-form-urlencoded;param=value",
+				Body:        `result=%5B%7B%22ab%22%3A%22hello1%22%7D%2C%7B%22ab%22%3A%22hello2%22%7D%5D`,
+			}},
+		}, {
+			name: "6",
+			config: map[string]interface{}{
+				"method": "post",
+				//"url": "http://localhost/test",  //set dynamically to the test server
+				"bodyType":   "form",
+				"sendSingle": true,
+			},
+			data: []map[string]interface{}{{
+				"ab": "hello1",
+			}, {
+				"ab": "hello2",
+			}},
+			result: []request{{
+				Method:      "POST",
+				ContentType: "application/x-www-form-urlencoded;param=value",
+				Body:        `ab=hello1`,
+			}, {
+				Method:      "POST",
+				ContentType: "application/x-www-form-urlencoded;param=value",
+				Body:        `ab=hello2`,
+			}},
+		}, {
+			name: "7",
+			config: map[string]interface{}{
+				"method": "post",
+				//"url": "http://localhost/test",  //set dynamically to the test server
+				"bodyType":   "json",
+				"sendSingle": true,
+				//"timeout":    float64(1000),
+			},
+			data: []map[string]interface{}{{
+				"ab": "hello1",
+			}, {
+				"ab": "hello2",
+			}},
+			result: []request{{
+				Method:      "POST",
+				Body:        `{"ab":"hello1"}`,
+				ContentType: "application/json",
+			}, {
+				Method:      "POST",
+				Body:        `{"ab":"hello2"}`,
+				ContentType: "application/json",
+			}},
+		},
+	}
+	ctx := mockContext.NewMockContext("testApply", "op")
+
+	var requests []request
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			fmt.Printf("Error reading body: %v", err)
+			http.Error(w, "can't read body", http.StatusBadRequest)
+			return
+		}
+
+		requests = append(requests, request{
+			Method:      r.Method,
+			Body:        string(body),
+			ContentType: r.Header.Get("Content-Type"),
+		})
+		ctx.GetLogger().Debugf(string(body))
+		fmt.Fprint(w, string(body))
+	}))
+	defer ts.Close()
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			requests = nil
+			ss, ok := tt.config["sendSingle"]
+			if !ok {
+				ss = false
+			}
+			s := &RestSink{}
+			tt.config["url"] = ts.URL
+			e := s.Provision(ctx, tt.config)
+			assert.NoError(t, e)
+			e = s.Connect(ctx)
+			assert.NoError(t, e)
+			if ss.(bool) {
+				for _, d := range tt.data {
+					e = s.Collect(ctx, &xsql.Tuple{
+						Message: d,
+					})
+					assert.NoError(t, e)
+				}
+			} else {
+				b := &xsql.WindowTuples{
+					Content: make([]xsql.Row, 0, len(tt.data)),
+				}
+				for _, d := range tt.data {
+					b.Content = append(b.Content, &xsql.Tuple{
+						Message: d,
+					})
+				}
+				e = s.CollectList(ctx, b)
+				assert.NoError(t, e)
+			}
+
+			err := s.Close(ctx)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.result, requests)
+		})
+	}
+}
+
 func TestRestSinkCollect(t *testing.T) {
 	server := createServer()
 	defer func() {
@@ -44,7 +257,7 @@ func TestRestSinkCollect(t *testing.T) {
 		},
 	}
 	require.NoError(t, s.Connect(ctx))
-	require.NoError(t, s.collect(ctx, data))
+	require.NoError(t, s.collect(ctx, data, data.ToMap()))
 	require.NoError(t, s.Close(ctx))
 }
 
@@ -65,7 +278,7 @@ func TestRestSinkRecoverErr(t *testing.T) {
 		"method": "get",
 	}))
 	require.NoError(t, sErr.Connect(ctx))
-	err := sErr.collect(ctx, data)
+	err := sErr.collect(ctx, data, data.ToMap())
 	require.Error(t, err)
 	require.False(t, errorx.IsIOError(err))
 	s := &RestSink{}
@@ -76,7 +289,7 @@ func TestRestSinkRecoverErr(t *testing.T) {
 	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/io/http/recoverAbleErr", "return(true)")
 	defer failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/io/http/recoverAbleErr")
 	require.NoError(t, s.Connect(ctx))
-	err = s.collect(ctx, data)
+	err = s.collect(ctx, data, data.ToMap())
 	require.Error(t, err)
 	require.True(t, errorx.IsIOError(err))
 }
diff --git a/internal/io/mqtt/client/client_test.go b/internal/io/mqtt/client/client_test.go
index 21741854..78de4a6c 100644
--- a/internal/io/mqtt/client/client_test.go
+++ b/internal/io/mqtt/client/client_test.go
@@ -16,11 +16,13 @@ package client
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestValidate(t *testing.T) {
@@ -59,3 +61,21 @@ func TestValidate(t *testing.T) {
 		})
 	}
 }
+
+func TestMqttClientPing(t *testing.T) {
+	url, cancel, err := testx.InitBroker("TestMqttClientPing")
+	require.NoError(t, err)
+	ctx := mockContext.NewMockContext("1", "2")
+	c, err := CreateClient(ctx, map[string]any{
+		"server":     url,
+		"datasource": "demo",
+	})
+	require.NoError(t, err)
+	// wait connection done
+	time.Sleep(100 * time.Millisecond)
+	require.NoError(t, c.Ping(ctx))
+	cancel()
+	// wait status done
+	time.Sleep(100 * time.Millisecond)
+	require.Error(t, c.Ping(ctx))
+}
diff --git a/internal/plugin/portable/runtime/plugin_ins_manager_test.go b/internal/plugin/portable/runtime/plugin_ins_manager_test.go
index 0fdf2d49..f449b416 100644
--- a/internal/plugin/portable/runtime/plugin_ins_manager_test.go
+++ b/internal/plugin/portable/runtime/plugin_ins_manager_test.go
@@ -197,7 +197,7 @@ func createMockClient(pluginName string) (mangos.Socket, error) {
 func TestPluginStatus(t *testing.T) {
 	p := NewPluginIns("mock", nil, nil)
 	require.Equal(t, PluginStatusInit, p.GetStatus().Status)
-	p.Status.StartRunning()
+	p.Status.StartRunning(13)
 	require.Equal(t, PluginStatusRunning, p.GetStatus().Status)
 	p.Status.StatusErr(errors.New("mock"))
 	require.Equal(t, PluginStatusErr, p.GetStatus().Status)
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 0c4c65f2..99797291 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -25,6 +25,7 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/gorilla/mux"
 	"github.com/pingcap/failpoint"
@@ -33,6 +34,7 @@ import (
 	"github.com/stretchr/testify/suite"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
 	"github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
@@ -1014,3 +1016,61 @@ func (suite *RestTestSuite) TestSinkHiddenPassword() {
 	m := r.Actions[0]["mqtt"].(map[string]interface{})
 	require.Equal(suite.T(), "12345", m["password"])
 }
+
+func (suite *RestTestSuite) TestWaitStopRule() {
+	ip := "127.0.0.1"
+	port := 10085
+	httpserver.InitGlobalServerManager(ip, port, nil)
+	connection.InitConnectionManager4Test()
+
+	// delete create
+	req, _ := http.NewRequest(http.MethodDelete, "http://localhost:8080/streams/demo221", bytes.NewBufferString("any"))
+	w := httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+
+	// delete rule
+	req, _ = http.NewRequest(http.MethodDelete, "http://localhost:8080/rules/rule221", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+
+	// create stream
+	buf := bytes.NewBuffer([]byte(`{"sql":"CREATE stream demo221() WITH (DATASOURCE=\"/data1\", TYPE=\"websocket\")"}`))
+	req, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/streams", buf)
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	require.Equal(suite.T(), http.StatusCreated, w.Code)
+
+	// create rule
+	ruleJson := `{"id": "rule221","sql": "select a,b from demo221","actions": [{"log": {}}]}`
+	buf = bytes.NewBuffer([]byte(ruleJson))
+	req, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/rules", buf)
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	require.Equal(suite.T(), http.StatusCreated, w.Code)
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/topo/node/mockTimeConsumingClose", "return(true)")
+	defer failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/topo/node/mockTimeConsumingClose")
+	now := time.Now()
+	// delete rule
+	req, _ = http.NewRequest(http.MethodDelete, "http://localhost:8080/rules/rule221", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	end := time.Now()
+	require.True(suite.T(), end.Sub(now) >= 300*time.Millisecond)
+
+	// create rule
+	buf = bytes.NewBuffer([]byte(ruleJson))
+	req, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/rules", buf)
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	require.Equal(suite.T(), http.StatusCreated, w.Code)
+
+	now = time.Now()
+	// stop rule
+	req, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/rules/rule221/stop", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	end = time.Now()
+	require.True(suite.T(), end.Sub(now) >= 300*time.Millisecond)
+	waitAllRuleStop()
+}
diff --git a/internal/trial/manager_test.go b/internal/trial/manager_test.go
index 71a1df12..64c48526 100644
--- a/internal/trial/manager_test.go
+++ b/internal/trial/manager_test.go
@@ -34,7 +34,7 @@ import (
 // Run two test rules in parallel. Rerun one of the rules
 func TestTrialRule(t *testing.T) {
 	ip := "127.0.0.1"
-	port := 10087
+	port := 10089
 	httpserver.InitGlobalServerManager(ip, port, nil)
 	connection.InitConnectionManager4Test()
 	conf.IsTesting = true
@@ -43,16 +43,16 @@ func TestTrialRule(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, store.SetupDefault(dataDir))
 	p := processor.NewStreamProcessor()
-	p.ExecStmt("DROP STREAM demo")
+	p.ExecStmt("DROP STREAM demo876")
 	// Test 1 wrong rule
-	mockDef1 := `{"id":"rule1","sql":"select * from demo","mockSource":{"demo":{"data":[{"name":"demo","value":1}],"interval":100,"loop":true}},"sinkProps":{"sendSingle":true}}`
+	mockDef1 := `{"id":"rule876","sql":"select * from demo876","mockSource":{"demo876":{"data":[{"name":"demo876","value":1}],"interval":100,"loop":true}},"sinkProps":{"sendSingle":true}}`
 	_, err = TrialManager.CreateRule(mockDef1)
 	require.Error(t, err)
-	require.Equal(t, "fail to run rule rule1: fail to get stream demo, please check if stream is created", err.Error())
+	require.Equal(t, "fail to run rule rule876: fail to get stream demo876, please check if stream is created", err.Error())
 
-	_, err = p.ExecStmt("CREATE STREAM demo () WITH (DATASOURCE=\"demo\", TYPE=\"simulator\", FORMAT=\"json\", KEY=\"ts\")")
+	_, err = p.ExecStmt("CREATE STREAM demo876 () WITH (DATASOURCE=\"demo876\", TYPE=\"simulator\", FORMAT=\"json\", KEY=\"ts\")")
 	require.NoError(t, err)
-	defer p.ExecStmt("DROP STREAM demo")
+	defer p.ExecStmt("DROP STREAM demo876")
 
 	// Test 2 valid rule with mock
 	testValidTrial(t, mockDef1)
@@ -69,9 +69,9 @@ func testValidTrial(t *testing.T, mockDef1 string) {
 	// Test 2 valid rule with mock
 	id, err := TrialManager.CreateRule(mockDef1)
 	require.NoError(t, err)
-	require.Equal(t, "rule1", id)
+	require.Equal(t, "rule876", id)
 	// Read from ws
-	u := url.URL{Scheme: "ws", Host: "localhost:10087", Path: "/test/rule1"}
+	u := url.URL{Scheme: "ws", Host: "localhost:10089", Path: "/test/rule876"}
 	c1, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
 	assert.NoError(t, err)
 	recvCh := make(chan []byte, 10)
@@ -93,21 +93,21 @@ func testValidTrial(t *testing.T, mockDef1 string) {
 		}
 	}()
 	time.Sleep(100 * time.Millisecond)
-	require.NoError(t, TrialManager.StartRule("rule1"))
-	require.Equal(t, []byte(`{"name":"demo","value":1}`), <-recvCh)
+	require.NoError(t, TrialManager.StartRule("rule876"))
+	require.Equal(t, []byte(`{"name":"demo876","value":1}`), <-recvCh)
 	c1.Close()
-	TrialManager.StopRule("rule1")
+	TrialManager.StopRule("rule876")
 	closeCh <- struct{}{}
 }
 
 func testRuntimeErrorTrial(t *testing.T) {
 	// Test 3 Runtime error rule
-	mockDefErr := `{"id":"ruleErr","sql":"select name + value from demo","mockSource":{"demo":{"data":[{"name":"demo","value":1}],"interval":100,"loop":true}},"sinkProps":{"sendSingle":true}}`
+	mockDefErr := `{"id":"ruleErr","sql":"select name + value from demo876","mockSource":{"demo876":{"data":[{"name":"demo876","value":1}],"interval":100,"loop":true}},"sinkProps":{"sendSingle":true}}`
 	id, err := TrialManager.CreateRule(mockDefErr)
 	require.NoError(t, err)
 	require.Equal(t, "ruleErr", id)
 	// Read from ws
-	u := url.URL{Scheme: "ws", Host: "localhost:10087", Path: "/test/ruleErr"}
+	u := url.URL{Scheme: "ws", Host: "localhost:10089", Path: "/test/ruleErr"}
 	c2, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
 	require.NoError(t, err)
 	recvCh := make(chan []byte, 10)
@@ -130,19 +130,19 @@ func testRuntimeErrorTrial(t *testing.T) {
 	}()
 	time.Sleep(10 * time.Millisecond)
 	require.NoError(t, TrialManager.StartRule(id))
-	require.Equal(t, `run Select error: expr: binaryExpr:{ demo.name + demo.value } meet error, err:invalid operation string(demo) + float64(1)`, string(<-recvCh))
+	require.Equal(t, `run Select error: expr: binaryExpr:{ demo876.name + demo876.value } meet error, err:invalid operation string(demo876) + float64(1)`, string(<-recvCh))
 	TrialManager.StopRule(id)
 	closeCh <- struct{}{}
 	c2.Close()
 }
 
 func testRealSourceTrial(t *testing.T) {
-	noMockDef := `{"id":"rule2","sql":"select * from demo","sinkProps":{"sendSingle":true}}`
+	noMockDef := `{"id":"rule8765","sql":"select * from demo876","sinkProps":{"sendSingle":true}}`
 	id, err := TrialManager.CreateRule(noMockDef)
-	assert.Equal(t, "rule2", id)
+	assert.Equal(t, "rule8765", id)
 	assert.NoError(t, err)
 
-	u := url.URL{Scheme: "ws", Host: "localhost:10087", Path: "/test/rule2"}
+	u := url.URL{Scheme: "ws", Host: "localhost:10089", Path: "/test/rule8765"}
 	c3, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
 	require.NoError(t, err)
 	recvCh := make(chan []byte, 10)
diff --git a/pkg/connection/connection_pool_test.go b/pkg/connection/connection_pool_test.go
index 9c63cc2b..aa7b129c 100644
--- a/pkg/connection/connection_pool_test.go
+++ b/pkg/connection/connection_pool_test.go
@@ -17,6 +17,7 @@ package connection
 import (
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/pingcap/failpoint"
 	"github.com/stretchr/testify/require"
@@ -120,6 +121,7 @@ func TestConnectionStatus(t *testing.T) {
 	conf.WriteCfgIntoKVStorage("connections", "mockErr", "a1", map[string]interface{}{})
 	conf.WriteCfgIntoKVStorage("connections", "mock", "a2", map[string]interface{}{})
 	require.NoError(t, ReloadConnection())
+	time.Sleep(100 * time.Millisecond)
 	ctx := context.Background()
 	allStatus := GetAllConnectionStatus(ctx)
 	s, ok := allStatus["a1"]
@@ -135,6 +137,44 @@ func TestConnectionStatus(t *testing.T) {
 	}, s)
 }
 
+func TestGetAllConnectionStatus(t *testing.T) {
+	require.NoError(t, InitConnectionManager4Test())
+	ctx := mockContext.NewMockContext("id", "2")
+	cw1, err := FetchConnection(ctx, "id1", "mock", nil)
+	require.NoError(t, err)
+	cw2, err := FetchConnection(ctx, "id2", "mockErr", nil)
+	require.NoError(t, err)
+	cw3, err := FetchConnection(ctx, "id3", "blockconn", nil)
+	require.NoError(t, err)
+	cw1.Wait()
+	cw2.Wait()
+	allStatus := GetAllConnectionStatus(ctx)
+	s, ok := allStatus["id2"]
+	require.True(t, ok)
+	require.Equal(t, ConnectionStatus{
+		Status: ConnectionFail,
+		ErrMsg: "mockErr",
+	}, s)
+	s, ok = allStatus["id1"]
+	require.True(t, ok)
+	require.Equal(t, ConnectionStatus{
+		Status: ConnectionRunning,
+	}, s)
+	s, ok = allStatus["id3"]
+	require.True(t, ok)
+	require.Equal(t, ConnectionStatus{
+		Status: ConnectionIntializing,
+	}, s)
+	blockCh <- struct{}{}
+	cw3.Wait()
+	allStatus = GetAllConnectionStatus(ctx)
+	s, ok = allStatus["id3"]
+	require.True(t, ok)
+	require.Equal(t, ConnectionStatus{
+		Status: ConnectionRunning,
+	}, s)
+}
+
 func TestNonStoredConnection(t *testing.T) {
 	require.NoError(t, InitConnectionManager4Test())
 	ctx := mockContext.NewMockContext("id", "2")
@@ -151,13 +191,9 @@ func TestNonStoredConnection(t *testing.T) {
 	require.False(t, IsConnectionExists("id1"))
 }
 
-var blockCh chan any
-
 func TestConnectionLock(t *testing.T) {
 	require.NoError(t, InitConnectionManager4Test())
 	ctx := mockContext.NewMockContext("id", "2")
-	modules.RegisterConnection("blockconn", CreateBlockConnection)
-	blockCh = make(chan any, 10)
 	wg := sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
@@ -170,7 +206,16 @@ func TestConnectionLock(t *testing.T) {
 	require.True(t, CheckConn("ccc1"))
 }
 
-type blockConnection struct{}
+var blockCh chan any
+
+func init() {
+	blockCh = make(chan any, 10)
+	modules.RegisterConnection("blockconn", CreateBlockConnection)
+}
+
+type blockConnection struct {
+	blochCh chan any
+}
 
 func (b blockConnection) Ping(ctx api.StreamContext) error {
 	return nil