Repository: lf-edge/ekuiper
Pull request #2877 (2024-05-27 07:03:38)

refactor(converter): add context to provider func

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2877?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 `69.23077%` with `4 lines` in your changes are missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@e5d6e68`). [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/2877?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/converter/converter.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2877?src=pr&el=tree&filepath=internal%2Fconverter%2Fconverter.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvY29udmVydGVyL2NvbnZlcnRlci5nbw==) | 60.00% | [0 Missing and 2 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2877?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/converter/ext\_converter.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2877?src=pr&el=tree&filepath=internal%2Fconverter%2Fext_converter.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvY29udmVydGVyL2V4dF9jb252ZXJ0ZXIuZ28=) | 0.00% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2877?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/converter/ext\_mock.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2877?src=pr&el=tree&filepath=internal%2Fconverter%2Fext_mock.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvY29udmVydGVyL2V4dF9tb2NrLmdv) | 0.00% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2877?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    #2877   +/-   ##
============================================
  Coverage             ?   68.79%           
============================================
  Files                ?      276           
  Lines                ?    26300           
  Branches             ?        0           
============================================
  Hits                 ?    18091           
  Misses               ?     6585           
  Partials             ?     1624           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2877?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/conf/yaml_config_storage.go b/internal/conf/yaml_config_storage.go
index 5411cac9..9de1eb2a 100644
--- a/internal/conf/yaml_config_storage.go
+++ b/internal/conf/yaml_config_storage.go
@@ -175,3 +175,23 @@ func (s *sqlKVStore) GetByPrefix(prefix string) (map[string]map[string]interface
 	}
 	return r, nil
 }
+
+// WriteCfgIntoKVStorage only used for unit test
+func WriteCfgIntoKVStorage(typ string, plugin string, confKey string, confData map[string]interface{}) error {
+	key := buildKey(typ, plugin, confKey)
+	return saveCfgKeyToKV(key, confData)
+}
+
+// GotCfgFromKVStorage only used for unit test
+func GotCfgFromKVStorage(typ string, plugin string, confKey string) (map[string]interface{}, error) {
+	key := buildKey(typ, plugin, confKey)
+	kvStorage, err := getKVStorage()
+	if err != nil {
+		return nil, err
+	}
+	d, err := kvStorage.GetByPrefix(key)
+	if err != nil {
+		return nil, err
+	}
+	return d[key], nil
+}
diff --git a/internal/converter/converter.go b/internal/converter/converter.go
index b7eff524..cf81a55f 100644
--- a/internal/converter/converter.go
+++ b/internal/converter/converter.go
@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/converter/binary"
 	"github.com/lf-edge/ekuiper/v2/internal/converter/delimited"
 	"github.com/lf-edge/ekuiper/v2/internal/converter/json"
@@ -28,21 +29,21 @@ import (
 )
 
 func init() {
-	modules.RegisterConverter(message.FormatJson, func(_ string, _ string, _ string, schema map[string]*ast.JsonStreamField) (message.Converter, error) {
+	modules.RegisterConverter(message.FormatJson, func(_ api.StreamContext, _ string, _ string, _ string, schema map[string]*ast.JsonStreamField) (message.Converter, error) {
 		if schema == nil {
 			return json.GetConverter()
 		}
 		return json.NewFastJsonConverter(schema), nil
 	})
-	modules.RegisterConverter(message.FormatBinary, func(_ string, _ string, _ string, _ map[string]*ast.JsonStreamField) (message.Converter, error) {
+	modules.RegisterConverter(message.FormatBinary, func(_ api.StreamContext, _ string, _ string, _ string, _ map[string]*ast.JsonStreamField) (message.Converter, error) {
 		return binary.GetConverter()
 	})
-	modules.RegisterConverter(message.FormatDelimited, func(_ string, _ string, delimiter string, _ map[string]*ast.JsonStreamField) (message.Converter, error) {
+	modules.RegisterConverter(message.FormatDelimited, func(_ api.StreamContext, _ string, _ string, delimiter string, _ map[string]*ast.JsonStreamField) (message.Converter, error) {
 		return delimited.NewConverter(delimiter)
 	})
 }
 
-func GetOrCreateConverter(options *ast.Options) (c message.Converter, err error) {
+func GetOrCreateConverter(ctx api.StreamContext, options *ast.Options) (c message.Converter, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
@@ -68,7 +69,7 @@ func GetOrCreateConverter(options *ast.Options) (c message.Converter, err error)
 		schema = nil
 	}
 	if c, ok := modules.Converters[t]; ok {
-		return c(schemaFile, schemaName, options.DELIMITER, schema)
+		return c(ctx, schemaFile, schemaName, options.DELIMITER, schema)
 	}
 	return nil, fmt.Errorf("format type %s not supported", t)
 }
diff --git a/internal/converter/ext_converter.go b/internal/converter/ext_converter.go
index eba3ab06..5176804f 100644
--- a/internal/converter/ext_converter.go
+++ b/internal/converter/ext_converter.go
@@ -17,6 +17,7 @@
 package converter
 
 import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/converter/protobuf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/schema"
@@ -26,7 +27,7 @@ import (
 )
 
 func init() {
-	modules.RegisterConverter(message.FormatProtobuf, func(schemaFileName string, schemaMessageName string, _ string, _ map[string]*ast.JsonStreamField) (message.Converter, error) {
+	modules.RegisterConverter(message.FormatProtobuf, func(_ api.StreamContext, schemaFileName string, schemaMessageName string, _ string, _ map[string]*ast.JsonStreamField) (message.Converter, error) {
 		ffs, err := schema.GetSchemaFile(def.PROTOBUF, schemaFileName)
 		if err != nil {
 			return nil, err
diff --git a/internal/converter/ext_mock.go b/internal/converter/ext_mock.go
index 993dbae1..0f90b75e 100644
--- a/internal/converter/ext_mock.go
+++ b/internal/converter/ext_mock.go
@@ -28,7 +28,7 @@ import (
 )
 
 func init() {
-	modules.RegisterConverter("mock", func(_ string, _ string, _ string, _ map[string]*ast.JsonStreamField) (message.Converter, error) {
+	modules.RegisterConverter("mock", func(_ api.StreamContext, _ string, _ string, _ string, _ map[string]*ast.JsonStreamField) (message.Converter, error) {
 		return &MockConverter{}, nil
 	})
 }
diff --git a/internal/io/connection/connection_pool.go b/internal/io/connection/connection_pool.go
new file mode 100644
index 00000000..e829f880
--- /dev/null
+++ b/internal/io/connection/connection_pool.go
@@ -0,0 +1,258 @@
+// Copyright 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package connection
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/kv"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+var isTest bool
+
+func GetAllConnectionsID() []string {
+	globalConnectionManager.RLock()
+	defer globalConnectionManager.RUnlock()
+	ids := make([]string, 0)
+	for key := range globalConnectionManager.connectionPool {
+		ids = append(ids, key)
+	}
+	return ids
+}
+
+func PingConnection(ctx api.StreamContext, id string) error {
+	conn, err := GetNameConnection(id)
+	if err != nil {
+		return err
+	}
+	return conn.Ping(ctx)
+}
+
+func GetNameConnection(selId string) (modules.Connection, error) {
+	if selId == "" {
+		return nil, fmt.Errorf("connection id should be defined")
+	}
+	globalConnectionManager.RLock()
+	defer globalConnectionManager.RUnlock()
+	meta, ok := globalConnectionManager.connectionPool[selId]
+	if !ok {
+		return nil, fmt.Errorf("connection %s not existed", selId)
+	}
+	return meta.conn, nil
+}
+
+func CreateNamedConnection(ctx api.StreamContext, id, typ string, props map[string]any) (modules.Connection, error) {
+	if id == "" || typ == "" {
+		return nil, fmt.Errorf("connection id and type should be defined")
+	}
+	globalConnectionManager.Lock()
+	defer globalConnectionManager.Unlock()
+	_, ok := globalConnectionManager.connectionPool[id]
+	if ok {
+		return nil, fmt.Errorf("connection %v already been created", id)
+	}
+	meta := ConnectionMeta{
+		ID:    id,
+		Typ:   typ,
+		Props: props,
+	}
+	if !isTest {
+		b, err := json.Marshal(meta)
+		if err != nil {
+			return nil, err
+		}
+		if err := globalConnectionManager.store.Set(id, string(b)); err != nil {
+			return nil, err
+		}
+	}
+	conn, err := createNamedConnection(ctx, meta)
+	if err != nil {
+		return nil, err
+	}
+	meta.conn = conn
+	globalConnectionManager.connectionPool[id] = meta
+	return conn, nil
+}
+
+func CreateNonStoredConnection(ctx api.StreamContext, id, typ string, props map[string]any) (modules.Connection, error) {
+	if id == "" || typ == "" {
+		return nil, fmt.Errorf("connection id and type should be defined")
+	}
+	globalConnectionManager.Lock()
+	defer globalConnectionManager.Unlock()
+	_, ok := globalConnectionManager.connectionPool[id]
+	if ok {
+		return nil, fmt.Errorf("connection %v already been created", id)
+	}
+	meta := ConnectionMeta{
+		ID:    id,
+		Typ:   typ,
+		Props: props,
+	}
+	conn, err := createNamedConnection(ctx, meta)
+	if err != nil {
+		return nil, err
+	}
+	meta.conn = conn
+	globalConnectionManager.connectionPool[id] = meta
+	return conn, nil
+}
+
+func DropNonStoredConnection(ctx api.StreamContext, selId string) error {
+	if selId == "" {
+		return fmt.Errorf("connection id should be defined")
+	}
+	globalConnectionManager.Lock()
+	defer globalConnectionManager.Unlock()
+	meta, ok := globalConnectionManager.connectionPool[selId]
+	if !ok {
+		return nil
+	}
+	conn := meta.conn
+	conn.Close(ctx)
+	delete(globalConnectionManager.connectionPool, selId)
+	return nil
+}
+
+func createNamedConnection(ctx api.StreamContext, meta ConnectionMeta) (modules.Connection, error) {
+	var conn modules.Connection
+	var err error
+	connRegister, ok := modules.ConnectionRegister[strings.ToLower(meta.Typ)]
+	if !ok {
+		return nil, fmt.Errorf("unknown connection type")
+	}
+	conn, err = connRegister(ctx, meta.ID, meta.Props)
+	if err != nil {
+		return nil, err
+	}
+	return conn, nil
+}
+
+func DropNameConnection(ctx api.StreamContext, selId string) error {
+	if selId == "" {
+		return fmt.Errorf("connection id should be defined")
+	}
+	globalConnectionManager.Lock()
+	defer globalConnectionManager.Unlock()
+	meta, ok := globalConnectionManager.connectionPool[selId]
+	if !ok {
+		return nil
+	}
+	conn := meta.conn
+	if conn.Ref(ctx) > 0 {
+		return fmt.Errorf("connection %s can't be dropped due to reference", selId)
+	}
+	if !isTest {
+		err := globalConnectionManager.store.Delete(selId)
+		if err != nil {
+			return fmt.Errorf("drop connection %s failed, err:%v", selId, err)
+		}
+	}
+	conn.Close(ctx)
+	delete(globalConnectionManager.connectionPool, selId)
+	return nil
+}
+
+var globalConnectionManager *ConnectionManager
+
+func InitConnectionManagerInTest() {
+	isTest = true
+	InitConnectionManager()
+}
+
+func InitConnectionManager() error {
+	globalConnectionManager = &ConnectionManager{
+		connectionPool: make(map[string]ConnectionMeta),
+	}
+	if !isTest {
+		globalConnectionManager.store, _ = store.GetKV("connectionMeta")
+		kvs, _ := globalConnectionManager.store.All()
+		for connectionID, raw := range kvs {
+			meta := ConnectionMeta{}
+			err := json.Unmarshal([]byte(raw), &meta)
+			if err != nil {
+				return fmt.Errorf("initialize connection:%v failed, err:%v", connectionID, err)
+			}
+			conn, err := createNamedConnection(context.Background(), meta)
+			if err != nil {
+				return fmt.Errorf("initialize connection:%v failed, err:%v", connectionID, err)
+			}
+			meta.conn = conn
+			globalConnectionManager.connectionPool[connectionID] = meta
+		}
+	}
+	return nil
+}
+
+type ConnectionManager struct {
+	sync.RWMutex
+	store          kv.KeyValue
+	connectionPool map[string]ConnectionMeta
+}
+
+type ConnectionMeta struct {
+	ID    string             `json:"id"`
+	Typ   string             `json:"typ"`
+	Props map[string]any     `json:"props"`
+	conn  modules.Connection `json:"-"`
+}
+
+type mockConnection struct {
+	id  string
+	ref int
+}
+
+func (m *mockConnection) Ping(ctx api.StreamContext) error {
+	return nil
+}
+
+func (m *mockConnection) Close(ctx api.StreamContext) {
+	return
+}
+
+func (m *mockConnection) Attach(ctx api.StreamContext) {
+	m.ref++
+	return
+}
+
+func (m *mockConnection) DetachSub(ctx api.StreamContext, props map[string]any) {
+	m.ref--
+	return
+}
+
+func (m *mockConnection) DetachPub(ctx api.StreamContext, props map[string]any) {
+	m.ref--
+	return
+}
+
+func (m *mockConnection) Ref(ctx api.StreamContext) int {
+	return m.ref
+}
+
+func CreateMockConnection(ctx api.StreamContext, id string, props map[string]any) (modules.Connection, error) {
+	m := &mockConnection{id: id, ref: 0}
+	return m, nil
+}
+
+func init() {
+	modules.ConnectionRegister["mock"] = CreateMockConnection
+}
diff --git a/internal/io/mqtt/connection.go b/internal/io/mqtt/client/client.go
similarity index 78%
rename from internal/io/mqtt/connection.go
rename to internal/io/mqtt/client/client.go
index 95132e1f..de87a527 100644
--- a/internal/io/mqtt/connection.go
+++ b/internal/io/mqtt/client/client.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package mqtt
+package client
 
 import (
 	"crypto/tls"
@@ -25,12 +25,16 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cert"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
+func init() {
+	modules.ConnectionRegister["mqtt"] = CreateConnection
+}
+
 type Connection struct {
 	pahoMqtt.Client
 	selId     string
@@ -90,27 +94,27 @@ func (conn *Connection) onReconnecting(_ pahoMqtt.Client, _ *pahoMqtt.ClientOpti
 }
 
 // Do not call this directly. Call connection pool Attach method to get the connection
-func (conn *Connection) attach() {
+func (conn *Connection) Attach(ctx api.StreamContext) {
 	conn.refCount.Add(1)
 }
 
+func (conn *Connection) Ref(ctx api.StreamContext) int {
+	return int(conn.refCount.Load())
+}
+
 // Do not call this directly. Call connection pool Detach method to release the connection
-func (conn *Connection) detachSub(topic string) bool {
+func (conn *Connection) DetachSub(ctx api.StreamContext, props map[string]any) {
+	conn.refCount.Add(-1)
+	topic, err := getTopicFromProps(props)
+	if err != nil {
+		return
+	}
 	delete(conn.subscriptions, topic)
 	conn.Client.Unsubscribe(topic)
-	if conn.refCount.Add(-1) == 0 {
-		go conn.Close()
-		return true
-	}
-	return false
 }
 
-func (conn *Connection) detachPub() bool {
-	if conn.refCount.Add(-1) == 0 {
-		go conn.Close()
-		return true
-	}
-	return false
+func (conn *Connection) DetachPub(ctx api.StreamContext, props map[string]any) {
+	conn.refCount.Add(-1)
 }
 
 func (conn *Connection) Subscribe(topic string, info *SubscriptionInfo) error {
@@ -119,11 +123,11 @@ func (conn *Connection) Subscribe(topic string, info *SubscriptionInfo) error {
 	return handleToken(token)
 }
 
-func (conn *Connection) Close() {
+func (conn *Connection) Close(ctx api.StreamContext) {
 	conn.Client.Disconnect(1)
 }
 
-func (conn *Connection) Ping() error {
+func (conn *Connection) Ping(ctx api.StreamContext) error {
 	if conn.Client.IsConnected() {
 		return nil
 	} else {
@@ -131,22 +135,13 @@ func (conn *Connection) Ping() error {
 	}
 }
 
+func CreateConnection(ctx api.StreamContext, selId string, props map[string]any) (modules.Connection, error) {
+	return CreateClient(ctx, selId, props)
+}
+
 // CreateClient creates a new mqtt client. It is anonymous and does not require a name.
 func CreateClient(ctx api.StreamContext, selId string, props map[string]any) (*Connection, error) {
-	if selId != "" {
-		selectCfg := &conf.ConSelector{
-			ConnSelectorStr: selId,
-		}
-		if err := selectCfg.Init(); err != nil {
-			return nil, err
-		}
-		cf, err := selectCfg.ReadCfgFromYaml()
-		if err != nil {
-			return nil, err
-		}
-		props = cf
-	}
-	c, err := validateConfig(props)
+	c, err := ValidateConfig(props)
 	if err != nil {
 		return nil, err
 	}
@@ -180,7 +175,9 @@ func CreateClient(ctx api.StreamContext, selId string, props map[string]any) (*C
 	}
 	ctx.GetLogger().Infof("new mqtt client created")
 	con.Client = cli
-	con.attach()
+	if len(selId) > 0 {
+		con.Attach(ctx)
+	}
 	return con, nil
 }
 
@@ -193,7 +190,7 @@ func handleToken(token pahoMqtt.Token) error {
 	return nil
 }
 
-func validateConfig(props map[string]any) (*ConnectionConfig, error) {
+func ValidateConfig(props map[string]any) (*ConnectionConfig, error) {
 	c := &ConnectionConfig{PVersion: "3.1.1"}
 	err := cast.MapToStruct(props, c)
 	if err != nil {
@@ -224,6 +221,22 @@ func validateConfig(props map[string]any) (*ConnectionConfig, error) {
 	return c, nil
 }
 
-func (conn *Connection) GetClientId() string {
-	return conn.selId
+func CreateAnonymousConnection(ctx api.StreamContext, props map[string]any) (*Connection, error) {
+	cli, err := CreateClient(ctx, "", props)
+	if err != nil {
+		return nil, err
+	}
+	return cli, nil
+}
+
+const (
+	dataSourceProp = "datasource"
+)
+
+func getTopicFromProps(props map[string]any) (string, error) {
+	v, ok := props[dataSourceProp]
+	if ok {
+		return v.(string), nil
+	}
+	return "", fmt.Errorf("topic or datasource not defined")
 }
diff --git a/internal/io/mqtt/connection_pool.go b/internal/io/mqtt/connection_pool.go
deleted file mode 100644
index e63e1258..00000000
--- a/internal/io/mqtt/connection_pool.go
+++ /dev/null
@@ -1,59 +0,0 @@
-// Copyright 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.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package mqtt
-
-import (
-	"sync"
-
-	"github.com/lf-edge/ekuiper/contract/v2/api"
-)
-
-var (
-	connectionPool = make(map[string]*Connection)
-	lock           sync.RWMutex
-)
-
-func GetConnection(ctx api.StreamContext, selId string, props map[string]any) (*Connection, error) {
-	if selId == "" {
-		return CreateClient(ctx, "", props)
-	}
-	lock.Lock()
-	defer lock.Unlock()
-	if conn, ok := connectionPool[selId]; ok {
-		conn.attach()
-		return conn, nil
-	} else {
-		cli, err := CreateClient(ctx, selId, props)
-		if err != nil {
-			return nil, err
-		}
-		connectionPool[selId] = cli
-		return cli, nil
-	}
-}
-
-func DetachConnection(conn *Connection, selId string, subscribedTopic string) {
-	var closed bool
-	if subscribedTopic != "" {
-		closed = conn.detachSub(subscribedTopic)
-	} else {
-		closed = conn.detachPub()
-	}
-	lock.Lock()
-	defer lock.Unlock()
-	if _, ok := connectionPool[selId]; closed && ok {
-		delete(connectionPool, selId)
-	}
-}
diff --git a/internal/io/mqtt/sink.go b/internal/io/mqtt/sink.go
index 791afe4f..0384464a 100644
--- a/internal/io/mqtt/sink.go
+++ b/internal/io/mqtt/sink.go
@@ -19,6 +19,8 @@ import (
 	"strings"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
+	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 )
 
@@ -33,11 +35,11 @@ type AdConf struct {
 type Sink struct {
 	adconf *AdConf
 	config map[string]interface{}
-	cli    *Connection
+	cli    *client.Connection
 }
 
 func (ms *Sink) Provision(_ api.StreamContext, ps map[string]any) error {
-	_, err := validateConfig(ps)
+	_, err := client.ValidateConfig(ps)
 	if err != nil {
 		return err
 	}
@@ -62,7 +64,26 @@ func (ms *Sink) Provision(_ api.StreamContext, ps map[string]any) error {
 
 func (ms *Sink) Connect(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Connecting to mqtt server")
-	cli, err := GetConnection(ctx, ms.adconf.SelId, ms.config)
+	var cli *client.Connection
+	var err error
+	if len(ms.adconf.SelId) > 0 {
+		conn, err := connection.GetNameConnection(ms.adconf.SelId)
+		if err != nil {
+			return err
+		}
+		c, ok := conn.(*client.Connection)
+		if !ok {
+			return fmt.Errorf("connection %s should be mqtt connection", ms.adconf.SelId)
+		}
+		cli = c
+	} else {
+		id := fmt.Sprintf("%s-%s-%s-mqtt-sink", ctx.GetRuleId(), ctx.GetOpId(), ms.adconf.Tpc)
+		conn, err := connection.CreateNonStoredConnection(ctx, id, "mqtt", ms.config)
+		if err != nil {
+			return err
+		}
+		cli = conn.(*client.Connection)
+	}
 	ms.cli = cli
 	return err
 }
@@ -90,7 +111,12 @@ func (ms *Sink) Collect(ctx api.StreamContext, item api.RawTuple) error {
 func (ms *Sink) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Info("Closing mqtt sink connector")
 	if ms.cli != nil {
-		DetachConnection(ms.cli, ms.adconf.SelId, "")
+		if len(ms.adconf.SelId) < 1 {
+			id := fmt.Sprintf("%s-%s-%s-mqtt-sink", ctx.GetRuleId(), ctx.GetOpId(), ms.adconf.Tpc)
+			connection.DropNonStoredConnection(ctx, id)
+		} else {
+			ms.cli.DetachPub(ctx, nil)
+		}
 	}
 	return nil
 }
diff --git a/internal/io/mqtt/source.go b/internal/io/mqtt/source.go
index 8c132bef..c446a7ce 100644
--- a/internal/io/mqtt/source.go
+++ b/internal/io/mqtt/source.go
@@ -20,6 +20,8 @@ import (
 	pahoMqtt "github.com/eclipse/paho.mqtt.golang"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
+	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
@@ -32,7 +34,7 @@ type SourceConnector struct {
 	cfg   *Conf
 	props map[string]any
 
-	cli *Connection
+	cli *client.Connection
 }
 
 type Conf struct {
@@ -50,7 +52,7 @@ func (ms *SourceConnector) Provision(ctx api.StreamContext, props map[string]any
 	if cfg.Topic == "" {
 		return fmt.Errorf("topic is required")
 	}
-	_, err = validateConfig(props)
+	_, err = client.ValidateConfig(props)
 	if err != nil {
 		return err
 	}
@@ -60,18 +62,37 @@ func (ms *SourceConnector) Provision(ctx api.StreamContext, props map[string]any
 	return nil
 }
 
-func (ms *SourceConnector) Ping(props map[string]interface{}) error {
-	cli, err := CreateClient(context.Background(), "", ms.props)
+func (ms *SourceConnector) Ping(ctx api.StreamContext, props map[string]interface{}) error {
+	cli, err := client.CreateAnonymousConnection(context.Background(), props)
 	if err != nil {
 		return err
 	}
-	defer cli.Close()
-	return cli.Ping()
+	defer cli.Close(ctx)
+	return cli.Ping(ctx)
 }
 
 func (ms *SourceConnector) Connect(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Connecting to mqtt server")
-	cli, err := GetConnection(ctx, ms.cfg.SelId, ms.props)
+	var cli *client.Connection
+	var err error
+	if len(ms.cfg.SelId) > 0 {
+		conn, err := connection.GetNameConnection(ms.cfg.SelId)
+		if err != nil {
+			return err
+		}
+		c, ok := conn.(*client.Connection)
+		if !ok {
+			return fmt.Errorf("connection %s should be mqtt connection", ms.cfg.SelId)
+		}
+		cli = c
+	} else {
+		id := fmt.Sprintf("%s-%s-%s-mqtt-source", ctx.GetRuleId(), ctx.GetOpId(), ms.tpc)
+		conn, err := connection.CreateNonStoredConnection(ctx, id, "mqtt", ms.props)
+		if err != nil {
+			return err
+		}
+		cli = conn.(*client.Connection)
+	}
 	ms.cli = cli
 	return err
 }
@@ -79,7 +100,7 @@ func (ms *SourceConnector) Connect(ctx api.StreamContext) error {
 // Subscribe is a one time only operation for source. It connects to the mqtt broker and subscribe to the topic
 // Run open before subscribe
 func (ms *SourceConnector) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error {
-	return ms.cli.Subscribe(ms.tpc, &SubscriptionInfo{
+	return ms.cli.Subscribe(ms.tpc, &client.SubscriptionInfo{
 		Qos: byte(ms.cfg.Qos),
 		Handler: func(client pahoMqtt.Client, message pahoMqtt.Message) {
 			ms.onMessage(ctx, message, ingest)
@@ -105,7 +126,12 @@ func (ms *SourceConnector) onMessage(ctx api.StreamContext, msg pahoMqtt.Message
 func (ms *SourceConnector) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Closing mqtt source connector to topic %s.", ms.tpc)
 	if ms.cli != nil {
-		DetachConnection(ms.cli, ms.cfg.SelId, ms.tpc)
+		if len(ms.cfg.SelId) < 1 {
+			id := fmt.Sprintf("%s-%s-%s-mqtt-source", ctx.GetRuleId(), ctx.GetOpId(), ms.tpc)
+			connection.DropNonStoredConnection(ctx, id)
+		} else {
+			ms.cli.DetachSub(ctx, ms.props)
+		}
 	}
 	return nil
 }
diff --git a/internal/server/bump/bump_manager.go b/internal/server/bump/bump_manager.go
new file mode 100644
index 00000000..2c6a0039
--- /dev/null
+++ b/internal/server/bump/bump_manager.go
@@ -0,0 +1,156 @@
+// Copyright 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bump
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path"
+	"path/filepath"
+	"strings"
+
+	"github.com/pingcap/failpoint"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/filex"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/pkg/kv"
+)
+
+const currentVersion = 1
+
+var GlobalBumpManager *BumpManager
+
+type BumpManager struct {
+	Version int
+	store   kv.KeyValue
+}
+
+func InitBumpManager() error {
+	s, err := store.GetKV("$$eKuiperMeta")
+	failpoint.Inject("initManagerError", func() {
+		err = errors.New("initManagerError")
+	})
+	if err != nil {
+		return fmt.Errorf("init bump manager failed, err:%v", err)
+	}
+	b := &BumpManager{
+		store: s,
+	}
+	GlobalBumpManager = b
+	GlobalBumpManager.Version, err = loadVersionFromStorage()
+	return err
+}
+
+func loadVersionFromStorage() (int, error) {
+	var ver int
+	got, err := GlobalBumpManager.store.Get("version", &ver)
+	failpoint.Inject("loadVersionError", func() {
+		err = errors.New("loadVersionError")
+	})
+	if err != nil {
+		return 0, fmt.Errorf("init bump manager failed, err:%v", err)
+	}
+	if got {
+		return ver, nil
+	}
+	return 0, nil
+}
+
+func BumpToCurrentVersion(dataDir string) error {
+	for i := GlobalBumpManager.Version; i <= currentVersion; i++ {
+		switch i {
+		case 1:
+			if err := bumpFrom0To1(dataDir); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func bumpFrom0To1(dir string) error {
+	if err := migrateDataIntoStorage(dir, "sources"); err != nil {
+		return err
+	}
+	if err := migrateDataIntoStorage(dir, "connections"); err != nil {
+		return err
+	}
+	if err := migrateDataIntoStorage(dir, "sinks"); err != nil {
+		return err
+	}
+	if err := storeGlobalVersion(1); err != nil {
+		return err
+	}
+	GlobalBumpManager.Version = 1
+	return nil
+}
+
+func migrateDataIntoStorage(dataDir, confType string) error {
+	dir := path.Join(dataDir, confType)
+	files, err := os.ReadDir(dir)
+	failpoint.Inject("migrateReadError", func(val failpoint.Value) {
+		x := ""
+		switch val.(int) {
+		case 1:
+			x = "sources"
+		case 2:
+			x = "sinks"
+		case 3:
+			x = "connections"
+		}
+		if x == confType {
+			err = errors.New("migrateReadError")
+		}
+	})
+	if err != nil {
+		return err
+	}
+	for _, file := range files {
+		fname := file.Name()
+		if !strings.HasSuffix(fname, ".yaml") {
+			continue
+		}
+		pluginTyp := strings.TrimSuffix(fname, ".yaml")
+		filePath := filepath.Join(dir, fname)
+		confKeys := make(map[string]map[string]interface{})
+		err = filex.ReadYamlUnmarshal(filePath, &confKeys)
+		failpoint.Inject("migrateUnmarshalErr", func() {
+			err = errors.New("migrateUnmarshalErr")
+		})
+		if err != nil {
+			return err
+		}
+		for confKey, confData := range confKeys {
+			err = conf.WriteCfgIntoKVStorage(confType, pluginTyp, confKey, confData)
+			failpoint.Inject("migrateWriteErr", func() {
+				err = errors.New("migrateWriteErr")
+			})
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func storeGlobalVersion(ver int) error {
+	err := GlobalBumpManager.store.Set("version", ver)
+	failpoint.Inject("storeVersionErr", func() {
+		err = errors.New("storeVersionErr")
+	})
+	return err
+}
diff --git a/internal/server/connection.go b/internal/server/connection.go
new file mode 100644
index 00000000..ee60d7a2
--- /dev/null
+++ b/internal/server/connection.go
@@ -0,0 +1,89 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+
+	"github.com/gorilla/mux"
+
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+)
+
+type ConnectionRequest struct {
+	ID    string                 `json:"id"`
+	Typ   string                 `json:"typ"`
+	Props map[string]interface{} `json:"props"`
+}
+
+func connectionsHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	switch r.Method {
+	case http.MethodPost:
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			handleError(w, err, "Invalid body", logger)
+			return
+		}
+		req := &ConnectionRequest{}
+		if err := json.Unmarshal(body, req); err != nil {
+			handleError(w, err, "Invalid body", logger)
+			return
+		}
+		_, err = connection.CreateNamedConnection(context.Background(), req.ID, req.Typ, req.Props)
+		if err != nil {
+			handleError(w, err, "create connection failed", logger)
+			return
+		}
+		w.WriteHeader(http.StatusCreated)
+		w.Write([]byte("success"))
+	case http.MethodGet:
+		ids := connection.GetAllConnectionsID()
+		w.WriteHeader(http.StatusOK)
+		jsonResponse(ids, w, logger)
+	}
+}
+
+type ConnectionResponse struct {
+	ID  string `json:"id"`
+	Err string `json:"err"`
+}
+
+func connectionHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	id := mux.Vars(r)["id"]
+	switch r.Method {
+	case http.MethodGet:
+		res := &ConnectionResponse{
+			ID: id,
+		}
+		err := connection.PingConnection(context.Background(), id)
+		if err != nil {
+			res.Err = err.Error()
+		}
+		w.WriteHeader(http.StatusOK)
+		jsonResponse(res, w, logger)
+	case http.MethodDelete:
+		if err := connection.DropNameConnection(context.Background(), id); err != nil {
+			handleError(w, err, "drop connection failed", logger)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte("success"))
+	}
+}
diff --git a/internal/server/rest.go b/internal/server/rest.go
index f505508f..6267e803 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -175,6 +175,8 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 	r.HandleFunc("/data/export", configurationExportHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/data/import", configurationImportHandler).Methods(http.MethodPost)
 	r.HandleFunc("/data/import/status", configurationStatusHandler).Methods(http.MethodGet)
+	r.HandleFunc("/connections", connectionsHandler).Methods(http.MethodGet, http.MethodPost)
+	r.HandleFunc("/connection/{id}", connectionHandler).Methods(http.MethodGet, http.MethodDelete)
 	// r.HandleFunc("/connection/websocket", connectionHandler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 	// Register extended routes
 	for k, v := range components {
diff --git a/internal/server/server.go b/internal/server/server.go
index bff7b80a..cfaed7bd 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -35,6 +35,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/binder/io"
 	"github.com/lf-edge/ekuiper/v2/internal/binder/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/keyedstate"
 	meta2 "github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
@@ -42,6 +43,7 @@ import (
 	"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/processor"
+	"github.com/lf-edge/ekuiper/v2/internal/server/bump"
 	"github.com/lf-edge/ekuiper/v2/internal/server/promMetrics"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
@@ -147,9 +149,21 @@ func StartUp(Version string) {
 	if err != nil {
 		panic(err)
 	}
+	if err := bump.InitBumpManager(); err != nil {
+		panic(err)
+	}
+	dataDir, _ := conf.GetDataLoc()
+	if err := bump.BumpToCurrentVersion(dataDir); err != nil {
+		panic(err)
+	}
+
 	keyedstate.InitKeyedStateKV()
 
 	meta2.InitYamlConfigManager()
+	if err := connection.InitConnectionManager(); err != nil {
+		panic(err)
+	}
+
 	ruleProcessor = processor.NewRuleProcessor()
 	streamProcessor = processor.NewStreamProcessor()
 	rulesetProcessor = processor.NewRulesetProcessor(ruleProcessor, streamProcessor)
diff --git a/internal/topo/node/decode_op.go b/internal/topo/node/decode_op.go
index 0b3944b9..34a0ba4f 100644
--- a/internal/topo/node/decode_op.go
+++ b/internal/topo/node/decode_op.go
@@ -67,7 +67,7 @@ func (o *DecodeOp) DetachSchema(ctx api.StreamContext, ruleId string) {
 	}
 }
 
-func NewDecodeOp(name, StreamName string, ruleId string, rOpt *def.RuleOption, options *ast.Options, isWildcard, isSchemaless bool, schema map[string]*ast.JsonStreamField, props map[string]any) (*DecodeOp, error) {
+func NewDecodeOp(ctx api.StreamContext, name, StreamName string, ruleId string, rOpt *def.RuleOption, options *ast.Options, isWildcard, isSchemaless bool, schema map[string]*ast.JsonStreamField, props map[string]any) (*DecodeOp, error) {
 	options.Schema = nil
 	options.IsWildCard = isWildcard
 	options.IsSchemaLess = isSchemaless
@@ -76,7 +76,7 @@ func NewDecodeOp(name, StreamName string, ruleId string, rOpt *def.RuleOption, o
 		options.StreamName = StreamName
 	}
 	options.RuleID = ruleId
-	converterTool, err := converter.GetOrCreateConverter(options)
+	converterTool, err := converter.GetOrCreateConverter(ctx, options)
 	if err != nil {
 		msg := fmt.Sprintf("cannot get converter from format %s, schemaId %s: %v", options.FORMAT, options.SCHEMAID, err)
 		return nil, fmt.Errorf(msg)
diff --git a/internal/topo/node/encode_op.go b/internal/topo/node/encode_op.go
index d2dac42e..13d170f4 100644
--- a/internal/topo/node/encode_op.go
+++ b/internal/topo/node/encode_op.go
@@ -35,8 +35,8 @@ type EncodeOp struct {
 	converter message.Converter
 }
 
-func NewEncodeOp(name string, rOpt *def.RuleOption, sc *SinkConf) (*EncodeOp, error) {
-	c, err := converter.GetOrCreateConverter(&ast.Options{FORMAT: sc.Format, SCHEMAID: sc.SchemaId, DELIMITER: sc.Delimiter})
+func NewEncodeOp(ctx api.StreamContext, name string, rOpt *def.RuleOption, sc *SinkConf) (*EncodeOp, error) {
+	c, err := converter.GetOrCreateConverter(ctx, &ast.Options{FORMAT: sc.Format, SCHEMAID: sc.SchemaId, DELIMITER: sc.Delimiter})
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/topo/planner/planner_sink.go b/internal/topo/planner/planner_sink.go
index d74a5782..add7dbcb 100644
--- a/internal/topo/planner/planner_sink.go
+++ b/internal/topo/planner/planner_sink.go
@@ -152,7 +152,7 @@ func splitSink(tp *topo.Topo, inputs []node.Emitter, s api.Sink, sinkName string
 	newInputs = []node.Emitter{transformOp}
 	// Encode will convert the result to []byte
 	if _, ok := s.(api.BytesCollector); ok {
-		encodeOp, err := node.NewEncodeOp(fmt.Sprintf("%s_%d_encode", sinkName, index), options, sc)
+		encodeOp, err := node.NewEncodeOp(tp.GetContext(), fmt.Sprintf("%s_%d_encode", sinkName, index), options, sc)
 		if err != nil {
 			return nil, err
 		}
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index 798722f2..8f28f85a 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -131,7 +131,7 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 
 	if info.NeedDecode {
 		// Create the decode node
-		decodeNode, err := node.NewDecodeOp(fmt.Sprintf("%d_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields, props)
+		decodeNode, err := node.NewDecodeOp(ctx, fmt.Sprintf("%d_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields, props)
 		if err != nil {
 			return nil, nil, 0, err
 		}
diff --git a/pkg/message/artifacts.go b/pkg/message/artifacts.go
index 43c0c261..37685dbf 100644
--- a/pkg/message/artifacts.go
+++ b/pkg/message/artifacts.go
@@ -38,7 +38,7 @@ type Converter interface {
 
 // ConverterProvider The format, schema information are passed in by stream options
 // The columns information is defined in the source side, like file source
-type ConverterProvider func(schemaFileName string, SchemaMessageName string, delimiter string, logicalSchema map[string]*ast.JsonStreamField) (Converter, error)
+type ConverterProvider func(ctx api.StreamContext, schemaFileName string, SchemaMessageName string, delimiter string, logicalSchema map[string]*ast.JsonStreamField) (Converter, error)
 
 type SchemaResetAbleConverter interface {
 	ResetSchema(schema map[string]*ast.JsonStreamField)
diff --git a/pkg/modules/connection.go b/pkg/modules/connection.go
new file mode 100644
index 00000000..feab1007
--- /dev/null
+++ b/pkg/modules/connection.go
@@ -0,0 +1,34 @@
+// Copyright 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package modules
+
+import "github.com/lf-edge/ekuiper/contract/v2/api"
+
+type Connection interface {
+	Ping(ctx api.StreamContext) error
+	Close(ctx api.StreamContext)
+	Attach(ctx api.StreamContext)
+	DetachSub(ctx api.StreamContext, props map[string]any)
+	DetachPub(ctx api.StreamContext, props map[string]any)
+	Ref(ctx api.StreamContext) int
+}
+
+type RegisterConnection func(ctx api.StreamContext, id string, props map[string]any) (Connection, error)
+
+var ConnectionRegister map[string]RegisterConnection
+
+func init() {
+	ConnectionRegister = map[string]RegisterConnection{}
+}


Test patch:
diff --git a/internal/io/connection/connection_pool_test.go b/internal/io/connection/connection_pool_test.go
new file mode 100644
index 00000000..dd0f4f9d
--- /dev/null
+++ b/internal/io/connection/connection_pool_test.go
@@ -0,0 +1,78 @@
+// Copyright 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package connection
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+)
+
+func TestConnection(t *testing.T) {
+	InitConnectionManagerInTest()
+	ctx := context.Background()
+	conn, err := CreateNamedConnection(ctx, "id1", "mock", nil)
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+	require.NoError(t, conn.Ping(ctx))
+	_, err = CreateNamedConnection(ctx, "id1", "mock", nil)
+	require.Error(t, err)
+	require.Equal(t, 0, conn.Ref(ctx))
+	conn.Attach(ctx)
+	require.Equal(t, 1, conn.Ref(ctx))
+	conn.Attach(ctx)
+	require.Equal(t, 2, conn.Ref(ctx))
+	conn.DetachPub(ctx, nil)
+	require.Equal(t, 1, conn.Ref(ctx))
+	err = DropNameConnection(ctx, "id1")
+	require.Error(t, err)
+	conn2, err := GetNameConnection("id1")
+	require.NoError(t, err)
+	require.NotNil(t, conn2)
+	conn.DetachSub(ctx, nil)
+	require.Equal(t, 0, conn.Ref(ctx))
+	err = DropNameConnection(ctx, "id1")
+	require.NoError(t, err)
+	err = DropNameConnection(ctx, "id1")
+	require.NoError(t, err)
+	conn3, err := GetNameConnection("id1")
+	require.Error(t, err)
+	require.Nil(t, conn3)
+
+	_, err = CreateNamedConnection(ctx, "", "mock", nil)
+	require.Error(t, err)
+	err = DropNameConnection(ctx, "")
+	require.Error(t, err)
+	_, err = CreateNamedConnection(ctx, "12", "unknown", nil)
+	require.Error(t, err)
+	_, err = GetNameConnection("")
+	require.Error(t, err)
+	err = PingConnection(ctx, "")
+	require.Error(t, err)
+	_, err = CreateNonStoredConnection(ctx, "", "mock", nil)
+	require.Error(t, err)
+
+	conn4, err := CreateNonStoredConnection(ctx, "id2", "mock", nil)
+	require.NoError(t, err)
+	require.NotNil(t, conn4)
+	_, err = CreateNonStoredConnection(ctx, "id2", "mock", nil)
+	require.Error(t, err)
+	err = DropNonStoredConnection(ctx, "")
+	require.Error(t, err)
+	err = DropNonStoredConnection(ctx, "nonexists")
+	require.NoError(t, err)
+}
diff --git a/internal/io/mqtt/client/client_test.go b/internal/io/mqtt/client/client_test.go
new file mode 100644
index 00000000..b4e180bf
--- /dev/null
+++ b/internal/io/mqtt/client/client_test.go
@@ -0,0 +1,55 @@
+// Copyright 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package client
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// NOTICE!!! Need to run a MQTT broker in localhost:1883 for this test or change the url to your broker
+const url = "tcp://127.0.0.1:1883"
+
+func TestValidate(t *testing.T) {
+	tests := []struct {
+		name  string
+		props map[string]any
+		err   string
+	}{
+		{
+			name: "No server",
+			props: map[string]any{
+				"server": "",
+			},
+			err: "missing server property",
+		},
+		{
+			name: "invalid protocol",
+			props: map[string]any{
+				"server":          url,
+				"protocolVersion": "5.0",
+			},
+			err: "unsupported protocol version 5.0",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := ValidateConfig(tt.props)
+			assert.Error(t, err)
+			assert.Equal(t, tt.err, err.Error())
+		})
+	}
+}
diff --git a/internal/io/mqtt/connection_test.go b/internal/io/mqtt/connection_test.go
deleted file mode 100644
index 42704f30..00000000
--- a/internal/io/mqtt/connection_test.go
+++ /dev/null
@@ -1,94 +0,0 @@
-// Copyright 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.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package mqtt
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-
-	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
-)
-
-func TestConnectionLC(t *testing.T) {
-	propsShared := map[string]any{
-		"connectionSelector": "mqtt.localConnection",
-	}
-	propsNormal := map[string]any{
-		"server": url,
-	}
-	propsInvalid := map[string]any{
-		"server": "abc",
-	}
-	ctx := mockContext.NewMockContext("test", "op")
-	connShared1, err := GetConnection(ctx, "mqtt.localConnection", propsShared)
-	assert.NoError(t, err)
-	assert.Equal(t, 1, len(connectionPool))
-	_, err = GetConnection(ctx, "", propsNormal)
-	assert.NoError(t, err)
-	assert.Equal(t, 1, len(connectionPool))
-	connShared2, err := GetConnection(ctx, "mqtt.localConnection", propsShared)
-	assert.NoError(t, err)
-	assert.Equal(t, 1, len(connectionPool))
-	assert.Equal(t, connShared1, connShared2)
-	_, err = GetConnection(ctx, "", propsInvalid)
-	assert.Error(t, err)
-	assert.Equal(t, err.Error(), "found error when connecting for abc: network Error : dial tcp: address abc: missing port in address")
-	assert.Equal(t, 1, len(connectionPool))
-
-	c := connShared1.refCount.Load()
-	assert.Equal(t, int32(2), c)
-
-	err = connShared1.Ping()
-	assert.NoError(t, err)
-
-	// Test subscribe in the connector test.
-
-	DetachConnection(connShared1, "mqtt.localConnection", "")
-	assert.Equal(t, 1, len(connectionPool))
-	DetachConnection(connShared2, "mqtt.localConnection", "")
-	assert.Equal(t, 0, len(connectionPool))
-}
-
-func TestValidate(t *testing.T) {
-	tests := []struct {
-		name  string
-		props map[string]any
-		err   string
-	}{
-		{
-			name: "No server",
-			props: map[string]any{
-				"server": "",
-			},
-			err: "missing server property",
-		},
-		{
-			name: "invalid protocol",
-			props: map[string]any{
-				"server":          url,
-				"protocolVersion": "5.0",
-			},
-			err: "unsupported protocol version 5.0",
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			_, err := validateConfig(tt.props)
-			assert.Error(t, err)
-			assert.Equal(t, tt.err, err.Error())
-		})
-	}
-}
diff --git a/internal/io/mqtt/sink_test.go b/internal/io/mqtt/sink_test.go
index 903efd3e..5ce5af52 100644
--- a/internal/io/mqtt/sink_test.go
+++ b/internal/io/mqtt/sink_test.go
@@ -21,10 +21,12 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestSinkConfigure(t *testing.T) {
+	connection.InitConnectionManagerInTest()
 	tests := []struct {
 		name           string
 		input          map[string]interface{}
@@ -135,6 +137,7 @@ func TestSinkConfigure(t *testing.T) {
 }
 
 func TestValidateMQTTSinkConf(t *testing.T) {
+	connection.InitConnectionManagerInTest()
 	testcases := []struct {
 		topic       string
 		expectError bool
diff --git a/internal/io/mqtt/source_sink_test.go b/internal/io/mqtt/source_sink_test.go
index 7a203db8..a7d4aeaf 100644
--- a/internal/io/mqtt/source_sink_test.go
+++ b/internal/io/mqtt/source_sink_test.go
@@ -20,12 +20,14 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mockclock"
 	"github.com/lf-edge/ekuiper/v2/pkg/mock"
 	"github.com/lf-edge/ekuiper/v2/pkg/model"
 )
 
 func TestSourceSink(t *testing.T) {
+	connection.InitConnectionManagerInTest()
 	sc := GetSource().(api.BytesSource)
 	sk := GetSink().(api.BytesCollector)
 	mc := mockclock.GetMockClock()
@@ -57,10 +59,14 @@ func TestSourceSink(t *testing.T) {
 	mock.TestSourceConnector(t, sc, map[string]any{
 		"server":     url,
 		"datasource": "demo",
+		"qos":        0,
+		"topic":      "demo",
 	}, result, func() {
 		err := mock.RunBytesSinkCollect(sk, data, map[string]any{
-			"server": url,
-			"topic":  "demo",
+			"server":   url,
+			"topic":    "demo",
+			"qos":      0,
+			"retained": false,
 		})
 		assert.NoError(t, err)
 	})
diff --git a/internal/io/mqtt/source_test.go b/internal/io/mqtt/source_test.go
index 2bc05f24..6b74691e 100644
--- a/internal/io/mqtt/source_test.go
+++ b/internal/io/mqtt/source_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
@@ -33,6 +34,7 @@ func init() {
 }
 
 func TestProvision(t *testing.T) {
+	connection.InitConnectionManagerInTest()
 	tests := []struct {
 		name  string
 		props map[string]any
diff --git a/internal/server/bump/bump_test.go b/internal/server/bump/bump_test.go
new file mode 100644
index 00000000..88603b23
--- /dev/null
+++ b/internal/server/bump/bump_test.go
@@ -0,0 +1,137 @@
+// Copyright 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bump
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/pingcap/failpoint"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/yaml.v3"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+)
+
+func TestBumpVersion(t *testing.T) {
+	conf.InitConf()
+	data, err := conf.GetDataLoc()
+	if err != nil {
+		t.Error(err)
+	}
+	err = store.SetupDefault(data)
+	if err != nil {
+		t.Error(err)
+	}
+	require.NoError(t, InitBumpManager())
+	GlobalBumpManager.Version = 0
+	testBumpVersion(t)
+}
+
+func prepareBumpVersion(t *testing.T, dir string) {
+	prepareBumpVersion1Data(t, dir, "sources")
+	prepareBumpVersion1Data(t, dir, "sinks")
+	prepareBumpVersion1Data(t, dir, "connections")
+}
+
+func testBumpVersion(t *testing.T) {
+	dir := os.TempDir()
+	prepareBumpVersion(t, dir)
+	require.NoError(t, BumpToCurrentVersion(dir))
+	require.Equal(t, 1, GlobalBumpManager.Version)
+	v, err := loadVersionFromStorage()
+	require.NoError(t, err)
+	require.Equal(t, 1, v)
+	assertBumpVersion1Data(t, "sources")
+	assertBumpVersion1Data(t, "sinks")
+	assertBumpVersion1Data(t, "connections")
+}
+
+func assertBumpVersion1Data(t *testing.T, typ string) {
+	d, err := conf.GotCfgFromKVStorage(typ, "mqtt", "conf1")
+	require.NoError(t, err)
+	require.Equal(t, map[string]interface{}{
+		"p1": 1,
+	}, d)
+	d, err = conf.GotCfgFromKVStorage(typ, "mqtt", "conf2")
+	require.NoError(t, err)
+	require.Equal(t, map[string]interface{}{
+		"p2": 2,
+	}, d)
+}
+
+func prepareBumpVersion1Data(t *testing.T, dir, typ string) {
+	sourceDir := filepath.Join(dir, typ)
+	require.NoError(t, os.MkdirAll(sourceDir, 0o755))
+	file, err := os.Create(filepath.Join(sourceDir, "mqtt.yaml"))
+	require.NoError(t, err)
+	defer file.Close()
+	m := make(map[string]map[string]interface{})
+	m["conf1"] = map[string]interface{}{
+		"p1": 1,
+	}
+	m["conf2"] = map[string]interface{}{
+		"p2": 2,
+	}
+	d, err := yaml.Marshal(m)
+	require.NoError(t, err)
+	_, err = file.Write(d)
+	require.NoError(t, err)
+}
+
+func TestBumpManager0To1(t *testing.T) {
+	conf.InitConf()
+	data, err := conf.GetDataLoc()
+	if err != nil {
+		t.Error(err)
+	}
+	err = store.SetupDefault(data)
+	if err != nil {
+		t.Error(err)
+	}
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/bump/initManagerError", "return(true)")
+	require.Error(t, InitBumpManager())
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/server/bump/initManagerError")
+	require.NoError(t, InitBumpManager())
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/bump/loadVersionError", "return(true)")
+	_, err = loadVersionFromStorage()
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/server/bump/loadVersionError")
+
+	dir := os.TempDir()
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/bump/migrateReadError", "return(1)")
+	require.Error(t, bumpFrom0To1(dir))
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/bump/migrateReadError", "return(2)")
+	require.Error(t, bumpFrom0To1(dir))
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/bump/migrateReadError", "return(3)")
+	require.Error(t, bumpFrom0To1(dir))
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/server/bump/migrateReadError")
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/bump/storeVersionErr", "return(true)")
+	require.Error(t, bumpFrom0To1(dir))
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/server/bump/storeVersionErr")
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/bump/migrateUnmarshalErr", "return(true)")
+	require.Error(t, bumpFrom0To1(dir))
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/server/bump/migrateUnmarshalErr")
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/server/bump/migrateWriteErr", "return(true)")
+	require.Error(t, bumpFrom0To1(dir))
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/server/bump/migrateWriteErr")
+}
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 9e2b4119..df6239a3 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -32,6 +32,7 @@ import (
 	"github.com/stretchr/testify/suite"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"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"
@@ -91,6 +92,8 @@ func (suite *RestTestSuite) SetupTest() {
 	r.HandleFunc("/data/export", configurationExportHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/data/import", configurationImportHandler).Methods(http.MethodPost)
 	r.HandleFunc("/data/import/status", configurationStatusHandler).Methods(http.MethodGet)
+	r.HandleFunc("/connections", connectionsHandler).Methods(http.MethodGet, http.MethodPost)
+	r.HandleFunc("/connection/{id}", connectionHandler).Methods(http.MethodGet, http.MethodDelete)
 	// r.HandleFunc("/connection/websocket", connectionHandler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 	r.HandleFunc("/metadata/sinks/{name}/confKeys/{confKey}", sinkConfKeyHandler).Methods(http.MethodDelete, http.MethodPut)
 	suite.r = r
@@ -802,3 +805,32 @@ func (suite *RestTestSuite) TestCreateDuplicateRule() {
 	returnVal, _ = io.ReadAll(w2.Result().Body)
 	require.Equal(suite.T(), `{"error":1000,"message":"rule test12345 already exists"}`+"\n", string(returnVal))
 }
+
+func (suite *RestTestSuite) TestConnection() {
+	connection.InitConnectionManagerInTest()
+	buf1 := bytes.NewBuffer([]byte(`{"id":"id1","typ":"mock","props":{}}`))
+	req1, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/connections", buf1)
+	w1 := httptest.NewRecorder()
+	suite.r.ServeHTTP(w1, req1)
+	require.Equal(suite.T(), http.StatusCreated, w1.Code)
+
+	req, _ := http.NewRequest(http.MethodGet, "http://localhost:8080/connections", bytes.NewBufferString("any"))
+	w := httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	assert.Equal(suite.T(), http.StatusOK, w.Code)
+	returnVal, _ := io.ReadAll(w.Result().Body)
+	require.Equal(suite.T(), `["id1"]`, string(returnVal))
+
+	req, _ = http.NewRequest(http.MethodGet, "http://localhost:8080/connection/id1", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	assert.Equal(suite.T(), http.StatusOK, w.Code)
+
+	returnVal, _ = io.ReadAll(w.Result().Body)
+	require.Equal(suite.T(), `{"id":"id1","err":""}`, string(returnVal))
+
+	req, _ = http.NewRequest(http.MethodDelete, "http://localhost:8080/connection/id1", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	assert.Equal(suite.T(), http.StatusOK, w.Code)
+}
diff --git a/internal/topo/node/decode_op_test.go b/internal/topo/node/decode_op_test.go
index 2480b134..dbb9d048 100644
--- a/internal/topo/node/decode_op_test.go
+++ b/internal/topo/node/decode_op_test.go
@@ -37,9 +37,10 @@ func TestJSON(t *testing.T) {
 		{"single", 1},
 		{"multi", 10},
 	}
+	ctx := mockContext.NewMockContext("test", "Test")
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			op, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{
+			op, err := NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{
 				"sendInterval": "10ms",
 			})
 			assert.NoError(t, err)
@@ -92,12 +93,12 @@ func TestJSON(t *testing.T) {
 // Concurrency 10 - BenchmarkThrougput-16           1000000000               0.1553 ns/op
 // This is useful when a node is much slower
 func BenchmarkThrougput(b *testing.B) {
-	op, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: 10, Debug: true}, &ast.Options{FORMAT: "mock"}, false, true, nil, nil)
+	ctx := mockContext.NewMockContext("test1", "decode_test")
+	op, err := NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: 10, Debug: true}, &ast.Options{FORMAT: "mock"}, false, true, nil, nil)
 	assert.NoError(b, err)
 	out := make(chan any, 100)
 	err = op.AddOutput(out, "test")
 	assert.NoError(b, err)
-	ctx := mockContext.NewMockContext("test1", "decode_test")
 	errCh := make(chan error)
 	op.Exec(ctx, errCh)
 	go func() {
@@ -127,9 +128,10 @@ func TestJSONWithSchema(t *testing.T) {
 			},
 		}},
 	}
+	ctx := mockContext.NewMockContext("test1", "decode_test")
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			op, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json", SHARED: true}, false, false, map[string]*ast.JsonStreamField{
+			op, err := NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json", SHARED: true}, false, false, map[string]*ast.JsonStreamField{
 				"a": {
 					Type: "bigint",
 				},
@@ -223,13 +225,14 @@ func TestJSONWithSchema(t *testing.T) {
 }
 
 func TestValidate(t *testing.T) {
-	_, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "cann"}, false, true, nil, nil)
+	ctx := mockContext.NewMockContext("test1", "decode_test")
+	_, err := NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "cann"}, false, true, nil, nil)
 	assert.Error(t, err)
 	assert.Equal(t, "cannot get converter from format cann, schemaId : format type cann not supported", err.Error())
-	_, err = NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "none"})
+	_, err = NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "none"})
 	assert.Error(t, err)
 	assert.EqualError(t, err, "1 error(s) decoding:\n\n* error decoding 'sendInterval': time: invalid duration \"none\"")
-	do, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "12s"})
+	do, err := NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "12s"})
 	assert.NoError(t, err)
 	assert.Equal(t, 12*time.Second, do.sendInterval)
 }
diff --git a/internal/topo/node/encode_op_test.go b/internal/topo/node/encode_op_test.go
index 370d93a0..2b5f75a8 100644
--- a/internal/topo/node/encode_op_test.go
+++ b/internal/topo/node/encode_op_test.go
@@ -82,12 +82,12 @@ func TestEncodeJSON(t *testing.T) {
 			out: &xsql.RawTuple{Rawdata: []byte(`[{"age":20,"name":"joe"},{"age":21,"name":"tom"}]`), Props: map[string]string{"{{.a}}": "1"}},
 		},
 	}
-	op, err := NewEncodeOp("test", &def.RuleOption{BufferLength: 10, SendError: true}, &SinkConf{Format: "json"})
+	ctx := mockContext.NewMockContext("test1", "encode_test")
+	op, err := NewEncodeOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &SinkConf{Format: "json"})
 	assert.NoError(t, err)
 	out := make(chan any, 100)
 	err = op.AddOutput(out, "test")
 	assert.NoError(t, err)
-	ctx := mockContext.NewMockContext("test1", "transform_test")
 	errCh := make(chan error)
 	op.Exec(ctx, errCh)
 	for _, tt := range tests {
@@ -100,7 +100,8 @@ func TestEncodeJSON(t *testing.T) {
 }
 
 func TestEncodeValidate(t *testing.T) {
-	_, err := NewEncodeOp("test", &def.RuleOption{BufferLength: 10, SendError: true}, &SinkConf{Format: "cann"})
+	ctx := mockContext.NewMockContext("test1", "encode_test")
+	_, err := NewEncodeOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &SinkConf{Format: "cann"})
 	assert.Error(t, err)
 	assert.Equal(t, "format type cann not supported", err.Error())
 }
diff --git a/internal/topo/operator/preprocessor_test.go b/internal/topo/operator/preprocessor_test.go
index dbe2f69a..314676fd 100644
--- a/internal/topo/operator/preprocessor_test.go
+++ b/internal/topo/operator/preprocessor_test.go
@@ -1056,7 +1056,7 @@ func TestPreprocessorForBinary(t *testing.T) {
 		pp := &Preprocessor{checkSchema: true}
 		pp.streamFields = tt.stmt.StreamFields.ToJsonSchema()
 		format := message.FormatJson
-		ccc, _ := converter.GetOrCreateConverter(&ast.Options{FORMAT: format})
+		ccc, _ := converter.GetOrCreateConverter(ctx, &ast.Options{FORMAT: format})
 		if dm, e := ccc.Decode(ctx, tt.data); e != nil {
 			log.Fatal(e)
 			return
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index 6ead3104..7c39f787 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -4785,11 +4785,11 @@ func TestTransformSourceNode(t *testing.T) {
 	props := nodeConf.GetSourceConf("mqtt", &ast.Options{TYPE: "mqtt", DATASOURCE: "topic1"})
 	srcNode, err := node.NewSourceNode(tp.GetContext(), "test", &mqtt.SourceConnector{}, props, &def.RuleOption{SendError: false})
 	assert.NoError(t, err)
-	decodeNode, err := node.NewDecodeOp("2_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
+	decodeNode, err := node.NewDecodeOp(tp.GetContext(), "2_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
 	assert.NoError(t, err)
 	decomNode, err := node.NewDecompressOp("2_decompressor", &def.RuleOption{SendError: false}, "gzip")
 	assert.NoError(t, err)
-	decodeNode2, err := node.NewDecodeOp("3_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
+	decodeNode2, err := node.NewDecodeOp(tp.GetContext(), "3_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
 	assert.NoError(t, err)
 	props2 := nodeConf.GetSourceConf("mqtt", &ast.Options{TYPE: "mqtt", CONF_KEY: "testCom", DATASOURCE: "topic2"})
 	srcNode2, err := node.NewSourceNode(tp.GetContext(), "test", &mqtt.SourceConnector{}, props2, &def.RuleOption{SendError: false})