Repository: lf-edge/ekuiper
Pull request #2978 (2024-07-02 17:20:48)

feat: support cpu profiling with rule

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2978?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 `17.64706%` with `28 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@ed7bd5b`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit) about missing BASE report.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2978?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 |
|---|---|---|
| [internal/server/server.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2978?src=pr&el=tree&filepath=internal%2Fserver%2Fserver.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3NlcnZlci5nbw==) | 9.09% | [20 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2978?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/topo/context/default.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2978?src=pr&el=tree&filepath=internal%2Ftopo%2Fcontext%2Fdefault.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9jb250ZXh0L2RlZmF1bHQuZ28=) | 37.50% | [5 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2978?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/pprof\_init.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2978?src=pr&el=tree&filepath=internal%2Fserver%2Fpprof_init.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3Bwcm9mX2luaXQuZ28=) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2978?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-v2    #2978   +/-   ##
============================================
  Coverage             ?   68.76%           
============================================
  Files                ?      329           
  Lines                ?    31702           
  Branches             ?        0           
============================================
  Hits                 ?    21797           
  Misses               ?     7977           
  Partials             ?     1928           
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2978?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/etc/kuiper.yaml b/etc/kuiper.yaml
index 7291ff56..5c94e02b 100644
--- a/etc/kuiper.yaml
+++ b/etc/kuiper.yaml
@@ -64,6 +64,8 @@ basic:
   gracefulShutdownTimeout: 3s
   connection:
     backoffMaxElapsedDuration: 3m
+  # enableResourceProfiling indicates whether to enable resource profiling for eKuiper. If it is enabled, the cpu usage of the rule will be recorded.
+  enableResourceProfiling: false
 
 # The default options for all rules. Each rule can override this setting by defining its own option
 rule:
diff --git a/go.mod b/go.mod
index 25f9687b..c866d8db 100644
--- a/go.mod
+++ b/go.mod
@@ -4,6 +4,7 @@ require (
 	github.com/Masterminds/sprig/v3 v3.2.3
 	github.com/PaesslerAG/gval v1.2.2
 	github.com/PaesslerAG/jsonpath v0.1.1
+	github.com/Rookiecom/cpuprofile v1.0.2
 	github.com/alicebob/miniredis/v2 v2.30.0
 	github.com/apple/foundationdb/bindings/go v0.0.0-20240315202255-8943393e84fc
 	github.com/benbjohnson/clock v1.3.5
@@ -18,6 +19,7 @@ require (
 	github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3
 	github.com/golang-jwt/jwt/v5 v5.2.1
 	github.com/golang/protobuf v1.5.4
+	github.com/google/pprof v0.0.0-20240528025155-186aa0362fba
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/handlers v1.5.2
 	github.com/gorilla/mux v1.8.1
@@ -87,7 +89,6 @@ require (
 	github.com/go-playground/validator/v10 v10.19.0 // indirect
 	github.com/go-redis/redis/v7 v7.4.1 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.4+incompatible // indirect
-	github.com/google/pprof v0.0.0-20240319011627-a57c5dfe54fd // indirect
 	github.com/gorilla/websocket v1.5.1 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
diff --git a/go.sum b/go.sum
index c7a6d088..7b733b57 100644
--- a/go.sum
+++ b/go.sum
@@ -21,6 +21,8 @@ github.com/PaesslerAG/gval v1.2.2/go.mod h1:XRFLwvmkTEdYziLdaCeCa5ImcGVrfQbeNUbV
 github.com/PaesslerAG/jsonpath v0.1.0/go.mod h1:4BzmtoM/PI8fPO4aQGIusjGxGir2BzcV0grWtFzq1Y8=
 github.com/PaesslerAG/jsonpath v0.1.1 h1:c1/AToHQMVsduPAa4Vh6xp2U0evy4t8SWp8imEsylIk=
 github.com/PaesslerAG/jsonpath v0.1.1/go.mod h1:lVboNxFGal/VwW6d9JzIy56bUsYAP6tH/x80vjnCseY=
+github.com/Rookiecom/cpuprofile v1.0.2 h1:wXGOmpSyCfoi7D0OB79srblNSw5Ccn0fTy4c537yWy4=
+github.com/Rookiecom/cpuprofile v1.0.2/go.mod h1:JEBcjNHceN6/g4ppueDEqWlbtbMGMyLy0YlT8QnKFbE=
 github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=
 github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
 github.com/VividCortex/gohistogram v1.0.0 h1:6+hBz+qvs0JOrrNhhmR7lFxo5sINxBCGXrdtl/UvroE=
@@ -221,8 +223,8 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/pprof v0.0.0-20230207041349-798e818bf904/go.mod h1:uglQLonpP8qtYCYyzA+8c/9qtqgA3qsXGYqCPKARAFg=
-github.com/google/pprof v0.0.0-20240319011627-a57c5dfe54fd h1:LjW4RcTwfcqOYGmD7UpFrn1gfBZ9mgu7QN5mSeFkCog=
-github.com/google/pprof v0.0.0-20240319011627-a57c5dfe54fd/go.mod h1:czg5+yv1E0ZGTi6S6vVK1mke0fV+FaUhNGcd6VRS9Ik=
+github.com/google/pprof v0.0.0-20240528025155-186aa0362fba h1:ql1qNgCyOB7iAEk8JTNM+zJrgIbnyCKX/wdlyPufP5g=
+github.com/google/pprof v0.0.0-20240528025155-186aa0362fba/go.mod h1:K1liHPHnj73Fdn/EKuT8nrFqBihUSKXoLYU0BuatOYo=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
diff --git a/internal/conf/conf.go b/internal/conf/conf.go
index 4065f807..7b24524c 100644
--- a/internal/conf/conf.go
+++ b/internal/conf/conf.go
@@ -195,6 +195,7 @@ type KuiperConf struct {
 		EnableOpenZiti          bool              `yaml:"enableOpenZiti"`
 		AesKey                  string            `yaml:"aesKey"`
 		GracefulShutdownTimeout cast.DurationConf `yaml:"gracefulShutdownTimeout"`
+		EnableResourceProfiling bool              `yaml:"enableResourceProfiling"`
 	}
 	Rule   def.RuleOption
 	Sink   *SinkConf
diff --git a/internal/io/mqtt/client/client.go b/internal/io/mqtt/client/client.go
index 420fa0f6..fee87f69 100644
--- a/internal/io/mqtt/client/client.go
+++ b/internal/io/mqtt/client/client.go
@@ -151,7 +151,7 @@ func CreateClient(ctx api.StreamContext, props map[string]any) (*Connection, err
 	opts.OnReconnecting = con.onReconnecting
 	opts.ConnectRetry = true
 	opts.ConnectRetryInterval = connection.DefaultInitialInterval
-	opts.MaxReconnectInterval = connection.DefaultBackoffMaxElapsedDuration
+	opts.MaxReconnectInterval = connection.DefaultMaxInterval
 
 	cli := pahoMqtt.NewClient(opts)
 	token := cli.Connect()
diff --git a/internal/server/pprof_init.go b/internal/server/pprof_init.go
index 6d7ab503..87139a33 100644
--- a/internal/server/pprof_init.go
+++ b/internal/server/pprof_init.go
@@ -21,6 +21,10 @@ import (
 	"net/http"
 	_ "net/http/pprof"
 	"os"
+
+	"github.com/Rookiecom/cpuprofile"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 )
 
 func init() {
@@ -33,6 +37,11 @@ type pprofComp struct {
 
 func (p pprofComp) serve() {
 	go func() {
+		if conf.Config.Basic.EnableResourceProfiling {
+			cpuprofile.WebProfile(":6060")
+			return
+		}
+
 		if err := http.ListenAndServe(":6060", nil); err != nil {
 			log.Fatal(err)
 		}
diff --git a/internal/server/server.go b/internal/server/server.go
index 34847546..e47ee266 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -28,6 +28,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/Rookiecom/cpuprofile"
 	"github.com/sirupsen/logrus"
 	"go.uber.org/automaxprocs/maxprocs"
 
@@ -139,6 +140,11 @@ func StartUp(Version string) {
 	createPaths()
 	conf.InitConf()
 
+	serverCtx, serverCancel := context.WithCancel(context.Background())
+	if conf.Config.Basic.EnableResourceProfiling {
+		startCPUProfiling(serverCtx)
+	}
+
 	undo, _ := maxprocs.Set(maxprocs.Logger(conf.Log.Infof))
 	defer undo()
 
@@ -185,7 +191,8 @@ func StartUp(Version string) {
 		panic(err)
 	}
 	connection.InitConnectionManager()
-	if err := connection.ReloadConnection(); err != nil {
+	// reload with 3s timeout
+	if err := connection.ReloadConnection(3 * time.Second); err != nil {
 		conf.Log.Warn(err)
 	}
 	meta.Bind()
@@ -213,8 +220,7 @@ func StartUp(Version string) {
 			}
 		}
 	}
-	exit := make(chan struct{})
-	go runScheduleRuleChecker(exit)
+	go runScheduleRuleChecker(serverCtx)
 
 	// Start rest service
 	srvRest := createRestServer(conf.Config.Basic.RestIp, conf.Config.Basic.RestPort, conf.Config.Basic.Authentication)
@@ -267,7 +273,7 @@ func StartUp(Version string) {
 		time.Sleep(time.Second)
 		conf.Log.Info("eKuiper stopped by Stop request")
 	}
-	exit <- struct{}{}
+	serverCancel()
 	// wait rule checker exit
 	time.Sleep(10 * time.Millisecond)
 
@@ -368,7 +374,7 @@ func resetAllStreams() error {
 	return nil
 }
 
-func runScheduleRuleCheckerByInterval(d time.Duration, exit <-chan struct{}) {
+func runScheduleRuleCheckerByInterval(d time.Duration, ctx context.Context) {
 	conf.Log.Infof("start patroling schedule rule state")
 	ticker := time.NewTicker(d)
 	defer func() {
@@ -377,7 +383,7 @@ func runScheduleRuleCheckerByInterval(d time.Duration, exit <-chan struct{}) {
 	}()
 	for {
 		select {
-		case <-exit:
+		case <-ctx.Done():
 			return
 		case <-ticker.C:
 			rs, err := getAllRulesWithState()
@@ -392,8 +398,8 @@ func runScheduleRuleCheckerByInterval(d time.Duration, exit <-chan struct{}) {
 	}
 }
 
-func runScheduleRuleChecker(exit <-chan struct{}) {
-	runScheduleRuleCheckerByInterval(time.Duration(conf.Config.Basic.RulePatrolInterval), exit)
+func runScheduleRuleChecker(ctx context.Context) {
+	runScheduleRuleCheckerByInterval(time.Duration(conf.Config.Basic.RulePatrolInterval), ctx)
 }
 
 type RuleStatusMetricsValue int
@@ -474,6 +480,26 @@ func handleScheduleRule(now time.Time, r *def.Rule, state string) scheduleRuleAc
 	return scheduleRuleActionDoNothing
 }
 
+func startCPUProfiling(ctx context.Context) error {
+	if err := cpuprofile.StartProfilerAndAggregater(ctx, time.Duration(1000)*time.Millisecond); err != nil {
+		return err
+	}
+	receiveChan := make(chan *cpuprofile.DataSetAggregate, 1024)
+	cpuprofile.RegisterTag("rule", receiveChan)
+	go func(ctx context.Context) {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case data := <-receiveChan:
+				// TODO: support query in future
+				conf.Log.Debugf("cpu profile data: %v", data)
+			}
+		}
+	}(ctx)
+	return nil
+}
+
 func waitAllRuleStop(ctx context.Context) error {
 	rules, _ := ruleProcessor.GetAllRules()
 	for _, r := range rules {
diff --git a/internal/topo/context/default.go b/internal/topo/context/default.go
index fdafba69..cb8afee8 100644
--- a/internal/topo/context/default.go
+++ b/internal/topo/context/default.go
@@ -19,6 +19,7 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"runtime/pprof"
 	"sync"
 	"text/template"
 	"time"
@@ -55,6 +56,18 @@ type DefaultContext struct {
 	opsWg *sync.WaitGroup
 }
 
+func RuleBackground(ruleName string) *DefaultContext {
+	if !conf.Config.Basic.EnableResourceProfiling {
+		return Background()
+	}
+	ctx := pprof.WithLabels(context.Background(), pprof.Labels("rule", ruleName))
+	pprof.SetGoroutineLabels(ctx)
+	c := &DefaultContext{
+		ctx: ctx,
+	}
+	return c
+}
+
 func Background() *DefaultContext {
 	c := &DefaultContext{
 		ctx: context.Background(),
@@ -62,6 +75,13 @@ func Background() *DefaultContext {
 	return c
 }
 
+func WithContext(ctx context.Context) *DefaultContext {
+	c := &DefaultContext{
+		ctx: ctx,
+	}
+	return c
+}
+
 func WithValue(parent *DefaultContext, key, val interface{}) *DefaultContext {
 	parent.ctx = context.WithValue(parent.ctx, key, val)
 	return parent
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index 8857ab75..019c7fd0 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -224,7 +224,7 @@ func (s *Topo) prepareContext() {
 				}
 			}
 		}
-		ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
+		ctx := kctx.WithValue(kctx.RuleBackground(s.name), kctx.LoggerKey, contextLogger)
 		ctx = kctx.WithValue(ctx, kctx.RuleStartKey, timex.GetNowInMilli())
 		ctx = kctx.WithWg(ctx, s.opsWg)
 		s.ctx, s.cancel = ctx.WithCancel()
diff --git a/pkg/connection/connection_pool.go b/pkg/connection/connection_pool.go
index 59c33688..56a073f3 100644
--- a/pkg/connection/connection_pool.go
+++ b/pkg/connection/connection_pool.go
@@ -15,6 +15,7 @@
 package connection
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strings"
@@ -26,7 +27,7 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	topoContext "github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
@@ -236,6 +237,11 @@ func createNamedConnection(ctx api.StreamContext, meta *ConnectionMeta) (modules
 		return nil, fmt.Errorf("unknown connection type")
 	}
 	err = backoff.Retry(func() error {
+		select {
+		case <-ctx.Done():
+			return backoff.Permanent(errors.New("timeout"))
+		default:
+		}
 		conn, err = connRegister(ctx, meta.Props)
 		failpoint.Inject("createConnectionErr", func() {
 			if mockErr {
@@ -299,11 +305,19 @@ func InitConnectionManager() {
 	DefaultBackoffMaxElapsedDuration = time.Duration(conf.Config.Connection.BackoffMaxElapsedDuration)
 }
 
-func ReloadConnection() error {
+func ReloadConnection(timeout time.Duration) error {
 	cfgs, err := conf.GetCfgFromKVStorage("connections", "", "")
 	if err != nil {
 		return err
 	}
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer func() {
+		cancel()
+	}()
+	failpoint.Inject("reloadTimeout", func() {
+		time.Sleep(100 * time.Millisecond)
+	})
+
 	for key, props := range cfgs {
 		names := strings.Split(key, ".")
 		if len(names) != 3 {
@@ -316,7 +330,7 @@ func ReloadConnection() error {
 			Typ:   typ,
 			Props: props,
 		}
-		conn, err := createNamedConnection(context.Background(), meta)
+		conn, err := createNamedConnection(topoContext.WithContext(ctx), meta)
 		if err != nil {
 			conf.Log.Warnf("initialize connection:%v failed, err:%v", id, err)
 			globalConnectionManager.failConnection[id] = err.Error()


Test patch:
diff --git a/internal/server/server_test.go b/internal/server/server_test.go
index b199b214..d6608465 100644
--- a/internal/server/server_test.go
+++ b/internal/server/server_test.go
@@ -15,6 +15,7 @@
 package server
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
@@ -86,10 +87,10 @@ func TestHandleScheduleRule(t *testing.T) {
 }
 
 func TestRunScheduleRuleChecker(t *testing.T) {
-	exit := make(chan struct{})
-	go runScheduleRuleCheckerByInterval(3*time.Second, exit)
+	ctx, cancel := context.WithCancel(context.Background())
+	go runScheduleRuleCheckerByInterval(3*time.Second, ctx)
 	time.Sleep(1 * time.Second)
-	exit <- struct{}{}
+	cancel()
 }
 
 func TestHandleScheduleRuleState(t *testing.T) {
@@ -121,3 +122,13 @@ func TestHandleScheduleRuleState(t *testing.T) {
 	require.NoError(t, handleScheduleRuleState(now, r, rule.RuleStarted))
 	require.NoError(t, handleScheduleRuleState(now, r, rule.RuleWait))
 }
+
+func TestStartCPUProfiling(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	err := startCPUProfiling(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(5 * time.Second)
+	cancel()
+}
diff --git a/internal/topo/context/default_test.go b/internal/topo/context/default_test.go
index b9d37667..40390756 100644
--- a/internal/topo/context/default_test.go
+++ b/internal/topo/context/default_test.go
@@ -15,13 +15,17 @@
 package context
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"os"
 	"path"
 	"reflect"
+	"runtime/pprof"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
@@ -228,3 +232,11 @@ func TestParseTemplate(t *testing.T) {
 		}
 	}
 }
+
+func TestRuleBackground(t *testing.T) {
+	conf.InitConf()
+	conf.Config.Basic.EnableResourceProfiling = true
+	c := RuleBackground("test")
+	ctx := pprof.WithLabels(context.Background(), pprof.Labels("rule", "test"))
+	assert.Equal(t, c.ctx, ctx)
+}
diff --git a/pkg/connection/connection_pool_test.go b/pkg/connection/connection_pool_test.go
index b4c199f5..341573e5 100644
--- a/pkg/connection/connection_pool_test.go
+++ b/pkg/connection/connection_pool_test.go
@@ -16,6 +16,7 @@ package connection
 
 import (
 	"testing"
+	"time"
 
 	"github.com/pingcap/failpoint"
 	"github.com/stretchr/testify/require"
@@ -114,7 +115,7 @@ func TestConnectionStatus(t *testing.T) {
 	require.NoError(t, InitConnectionManager4Test())
 	conf.WriteCfgIntoKVStorage("connections", "mockErr", "a1", map[string]interface{}{})
 	conf.WriteCfgIntoKVStorage("connections", "mock", "a2", map[string]interface{}{})
-	require.NoError(t, ReloadConnection())
+	require.NoError(t, ReloadConnection(999*time.Second))
 	ctx := context.Background()
 	allStatus := GetAllConnectionStatus(ctx)
 	s, ok := allStatus["a1"]
@@ -130,6 +131,16 @@ func TestConnectionStatus(t *testing.T) {
 	}, s)
 }
 
+func TestReloadConnectionErr(t *testing.T) {
+	require.NoError(t, InitConnectionManager4Test())
+	conf.WriteCfgIntoKVStorage("connections", "mockErr", "a1", map[string]interface{}{})
+	conf.WriteCfgIntoKVStorage("connections", "mock", "a2", map[string]interface{}{})
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/pkg/connection/reloadTimeout", "return(true)")
+	require.NoError(t, ReloadConnection(time.Microsecond))
+	require.True(t, len(globalConnectionManager.failConnection) > 0)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/pkg/connection/reloadTimeout")
+}
+
 func TestNonStoredConnection(t *testing.T) {
 	require.NoError(t, InitConnectionManager4Test())
 	ctx := mockContext.NewMockContext("id", "2")