Repository: lf-edge/ekuiper
Pull request #3059 (2024-07-25 10:13:20)

feat: add initialize connection stauts

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3059?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `84.00000%` with `4 lines` in your changes missing coverage. Please review.
> Project coverage is 67.84%. 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 [(`05e4004`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/05e400436c351355af54ecb03edf17a65826ba13?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3059?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [pkg/connection/connection\_pool.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3059?src=pr&el=tree&filepath=pkg%2Fconnection%2Fconnection_pool.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-cGtnL2Nvbm5lY3Rpb24vY29ubmVjdGlvbl9wb29sLmdv) | 84.00% | [4 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3059?src=pr&el=tree&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    #3059      +/-   ##
==========================================
+ Coverage   67.84%   67.84%   +0.01%     
==========================================
  Files         390      390              
  Lines       37135    37148      +13     
==========================================
+ Hits        25191    25202      +11     
- Misses       9667     9668       +1     
- Partials     2277     2278       +1     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3059?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/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/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