Repository: lf-edge/ekuiper
Pull request #2819 (2024-04-28 06:16:11)

feat(sink): support file sink rolling hook


Patch:
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index a25634e7..7e1bebb6 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -24,7 +24,7 @@ import (
 )
 
 func init() {
-	modules.RegisterSource("mqtt", func() api.Source { return &mqtt.SourceConnector{} })
+	modules.RegisterSource("mqtt", mqtt.GetSource)
 	// modules.RegisterSource("httppull", func() api.Source { return &http.PullSource{} })
 	// modules.RegisterSource("httppush", func() api.Source { return &http.PushSource{} })
 	// modules.RegisterSource("file", func() api.Source { return &file.FileSource{} })
@@ -35,7 +35,7 @@ func init() {
 
 	modules.RegisterSink("log", sink.NewLogSink)
 	modules.RegisterSink("logToMemory", sink.NewLogSinkToMemory)
-	modules.RegisterSink("mqtt", func() api.Sink { return &mqtt.MQTTSink{} })
+	modules.RegisterSink("mqtt", mqtt.GetSink)
 	// modules.RegisterSink("rest", func() api.Sink { return &http.RestSink{} })
 	modules.RegisterSink("nop", func() api.Sink { return &sink.NopSink{} })
 	modules.RegisterSink("memory", func() api.Sink { return memory.GetSink() })
diff --git a/internal/converter/converter.go b/internal/converter/converter.go
index c900a420..02eccb4d 100644
--- a/internal/converter/converter.go
+++ b/internal/converter/converter.go
@@ -55,7 +55,10 @@ func GetOrCreateConverter(options *ast.Options) (c message.Converter, err error)
 		if options.RuleID == "" || options.StreamName == "" {
 			return json.GetConverter()
 		}
-		return json.NewFastJsonConverter(options.RuleID, options.StreamName, options.Schema, options.IsWildCard, options.IsSchemaLess), nil
+		if options.IsWildCard {
+			return json.NewFastJsonConverter(nil), nil
+		}
+		return json.NewFastJsonConverter(options.Schema), nil
 	}
 
 	schemaFile := ""
diff --git a/internal/converter/json/converter.go b/internal/converter/json/converter.go
index e1f9ef94..82114233 100644
--- a/internal/converter/json/converter.go
+++ b/internal/converter/json/converter.go
@@ -22,7 +22,6 @@ import (
 	"github.com/valyala/fastjson"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/converter/merge"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
@@ -57,92 +56,20 @@ func (c *Converter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 
 type FastJsonConverter struct {
 	sync.RWMutex
-	isSchemaLess bool
-	// ruleID -> schema
-	schemaMap map[string]map[string]*ast.JsonStreamField
-	// ruleID -> StreamName
-	streamMap map[string]string
-	schema    map[string]*ast.JsonStreamField
-	// ruleID -> wildcard
-	wildcardMap map[string]struct{}
+	schema map[string]*ast.JsonStreamField
 }
 
-func NewFastJsonConverter(ruleID, streamName string, schema map[string]*ast.JsonStreamField, isWildcard, isSchemaLess bool) *FastJsonConverter {
+func NewFastJsonConverter(schema map[string]*ast.JsonStreamField) *FastJsonConverter {
 	f := &FastJsonConverter{
-		schemaMap:    make(map[string]map[string]*ast.JsonStreamField),
-		schema:       schema,
-		wildcardMap:  make(map[string]struct{}),
-		isSchemaLess: isSchemaLess,
-		streamMap:    map[string]string{},
+		schema: schema,
 	}
-	f.schemaMap[ruleID] = schema
-	f.streamMap[ruleID] = streamName
-	if isWildcard {
-		f.wildcardMap[ruleID] = struct{}{}
-	}
-	f.storeSchema()
 	return f
 }
 
-func (c *FastJsonConverter) MergeSchema(ruleID, dataSource string, newSchema map[string]*ast.JsonStreamField, isWildcard bool) error {
-	c.Lock()
-	defer c.Unlock()
-	_, ok := c.schemaMap[ruleID]
-	if ok {
-		return nil
-	}
-	c.schemaMap[ruleID] = newSchema
-	c.streamMap[ruleID] = dataSource
-	if isWildcard {
-		c.wildcardMap[ruleID] = struct{}{}
-	} else {
-		mergedSchema, err := mergeSchema(c.schema, newSchema)
-		if err != nil {
-			return err
-		}
-		c.schema = mergedSchema
-	}
-	c.storeSchema()
-	return nil
-}
-
-func (c *FastJsonConverter) DetachSchema(ruleID string) error {
-	var err error
+func (c *FastJsonConverter) ResetSchema(schema map[string]*ast.JsonStreamField) {
 	c.Lock()
 	defer c.Unlock()
-	_, ok := c.schemaMap[ruleID]
-	if ok {
-		merge.RemoveRuleSchema(ruleID)
-		delete(c.streamMap, ruleID)
-		delete(c.wildcardMap, ruleID)
-		delete(c.schemaMap, ruleID)
-		newSchema := make(map[string]*ast.JsonStreamField)
-		for _, schema := range c.schemaMap {
-			newSchema, err = mergeSchema(newSchema, schema)
-			if err != nil {
-				return err
-			}
-		}
-		c.schema = newSchema
-		c.storeSchema()
-	}
-	return nil
-}
-
-func (c *FastJsonConverter) storeSchema() {
-	if len(c.wildcardMap) > 0 {
-		for ruleID := range c.schemaMap {
-			merge.AddRuleSchema(ruleID, c.streamMap[ruleID], nil, true)
-		}
-		return
-	}
-	for ruleID := range c.schemaMap {
-		merge.AddRuleSchema(ruleID, c.streamMap[ruleID], c.schema, false)
-	}
-}
-
-func mergeSchema(originSchema, newSchema map[string]*ast.JsonStreamField) (map[string]*ast.JsonStreamField, error) {
-	return merge.MergeSchema(originSchema, newSchema)
+	c.schema = schema
 }
 
 func (c *FastJsonConverter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
@@ -157,7 +84,7 @@ func (c *FastJsonConverter) Decode(ctx api.StreamContext, b []byte) (m any, err
 	}()
 	c.RLock()
 	defer c.RUnlock()
-	if len(c.wildcardMap) > 0 {
+	if c.schema == nil {
 		return converter.Decode(ctx, b)
 	}
 	return c.decodeWithSchema(b, c.schema)
@@ -203,9 +130,6 @@ func (f *FastJsonConverter) decodeWithSchema(b []byte, schema map[string]*ast.Js
 }
 
 func (f *FastJsonConverter) decodeArray(array []*fastjson.Value, field *ast.JsonStreamField) ([]interface{}, error) {
-	if !f.isSchemaLess && field == nil {
-		return nil, nil
-	}
 	vs := make([]interface{}, len(array))
 	for i, item := range array {
 		switch item.Type() {
@@ -289,54 +213,50 @@ func (f *FastJsonConverter) decodeObject(obj *fastjson.Object, schema map[string
 		case fastjson.TypeNull:
 			m[key] = nil
 		case fastjson.TypeObject:
-			add, valid := f.checkSchema(key, "struct", schema)
+			valid := f.checkSchema(key, "struct", schema)
 			if !valid {
 				err = fmt.Errorf("%v has wrong type:%v, expect:%v", key, v.Type().String(), getType(schema[key]))
 				return
 			}
-			if add {
-				childObj, err2 := v.Object()
-				if err2 != nil {
-					err = err2
-					return
-				}
-				var props map[string]*ast.JsonStreamField
-				if schema != nil && schema[key] != nil {
-					props = schema[key].Properties
-				}
-				childMap, err2 := f.decodeObject(childObj, props)
-				if err2 != nil {
-					err = err2
-					return
-				}
-				if childMap != nil {
-					m[key] = childMap
-				}
+			childObj, err2 := v.Object()
+			if err2 != nil {
+				err = err2
+				return
+			}
+			var props map[string]*ast.JsonStreamField
+			if schema != nil && schema[key] != nil {
+				props = schema[key].Properties
+			}
+			childMap, err2 := f.decodeObject(childObj, props)
+			if err2 != nil {
+				err = err2
+				return
+			}
+			if childMap != nil {
+				m[key] = childMap
 			}
 		case fastjson.TypeArray:
-			add, valid := f.checkSchema(key, "array", schema)
+			valid := f.checkSchema(key, "array", schema)
 			if !valid {
 				err = fmt.Errorf("%v has wrong type:%v, expect:%v", key, v.Type().String(), getType(schema[key]))
 				return
 			}
-			if add {
-				childArray, err2 := v.Array()
-				if err2 != nil {
-					err = err2
-					return
-				}
-				var items *ast.JsonStreamField
-				if schema != nil && schema[key] != nil {
-					items = schema[key].Items
-				}
-				subList, err2 := f.decodeArray(childArray, items)
-				if err2 != nil {
-					err = err2
-					return
-				}
-				if subList != nil {
-					m[key] = subList
-				}
+			childArray, err2 := v.Array()
+			if err2 != nil {
+				err = err2
+				return
+			}
+			var items *ast.JsonStreamField
+			if schema != nil && schema[key] != nil {
+				items = schema[key].Items
+			}
+			subList, err2 := f.decodeArray(childArray, items)
+			if err2 != nil {
+				err = err2
+				return
+			}
+			if subList != nil {
+				m[key] = subList
 			}
 		case fastjson.TypeString:
 			if schema != nil {
@@ -391,110 +311,92 @@ func (f *FastJsonConverter) decodeObject(obj *fastjson.Object, schema map[string
 	return m, nil
 }
 
-func (f *FastJsonConverter) checkSchema(key, typ string, schema map[string]*ast.JsonStreamField) (add bool, valid bool) {
-	if f.isSchemaLess {
-		_, ok := schema[key]
-		return ok, true
+func (f *FastJsonConverter) checkSchema(key, typ string, schema map[string]*ast.JsonStreamField) (valid bool) {
+	v := schema[key]
+	if v == nil {
+		return true
 	}
-	if !f.isSchemaLess && schema[key] != nil && schema[key].Type == typ {
-		return true, true
-	}
-	return false, false
+	return v.Type == typ
 }
 
 func (f *FastJsonConverter) extractNumberValue(name string, v *fastjson.Value, field *ast.JsonStreamField) (interface{}, error) {
-	if f.isSchemaLess && field == nil {
+	if field == nil {
 		f64, err := v.Float64()
 		if err != nil {
 			return nil, err
 		}
 		return f64, nil
 	}
-	if !f.isSchemaLess {
-		if field == nil {
-			return nil, nil
+	switch {
+	case field.Type == "float", field.Type == "datetime":
+		f64, err := v.Float64()
+		if err != nil {
+			return nil, err
 		}
-		switch {
-		case field.Type == "float", field.Type == "datetime":
-			f64, err := v.Float64()
-			if err != nil {
-				return nil, err
-			}
-			return f64, nil
-		case field.Type == "bigint":
-			i64, err := v.Int64()
-			if err != nil {
-				return nil, err
-			}
-			return i64, nil
-		case field.Type == "string":
-			f64, err := v.Float64()
-			if err != nil {
-				return nil, err
-			}
-			return cast.ToStringAlways(f64), nil
-		case field.Type == "boolean":
-			bv, err := getBooleanFromValue(v)
-			if err != nil {
-				return nil, err
-			}
-			return bv, nil
+		return f64, nil
+	case field.Type == "bigint":
+		i64, err := v.Int64()
+		if err != nil {
+			return nil, err
+		}
+		return i64, nil
+	case field.Type == "string":
+		f64, err := v.Float64()
+		if err != nil {
+			return nil, err
+		}
+		return cast.ToStringAlways(f64), nil
+	case field.Type == "boolean":
+		bv, err := getBooleanFromValue(v)
+		if err != nil {
+			return nil, err
 		}
+		return bv, nil
 	}
 	return nil, fmt.Errorf("%v has wrong type:%v, expect:%v", name, fastjson.TypeNumber.String(), getType(field))
 }
 
 func (f *FastJsonConverter) extractStringValue(name string, v *fastjson.Value, field *ast.JsonStreamField) (interface{}, error) {
-	if f.isSchemaLess && field == nil {
+	if field == nil {
 		bs, err := v.StringBytes()
 		if err != nil {
 			return nil, err
 		}
 		return string(bs), nil
 	}
-	if !f.isSchemaLess {
-		if field == nil {
-			return nil, nil
+	switch {
+	case field.Type == "string", field.Type == "datetime":
+		bs, err := v.StringBytes()
+		if err != nil {
+			return nil, err
 		}
-		switch {
-		case field.Type == "string", field.Type == "datetime":
-			bs, err := v.StringBytes()
-			if err != nil {
-				return nil, err
-			}
-			return string(bs), nil
-		case field.Type == "bytea":
-			s, err := v.StringBytes()
-			if err != nil {
-				return nil, err
-			}
-			return cast.ToByteA(string(s), cast.CONVERT_ALL)
-		case field.Type == "boolean":
-			return getBooleanFromValue(v)
+		return string(bs), nil
+	case field.Type == "bytea":
+		s, err := v.StringBytes()
+		if err != nil {
+			return nil, err
 		}
+		return cast.ToByteA(string(s), cast.CONVERT_ALL)
+	case field.Type == "boolean":
+		return getBooleanFromValue(v)
 	}
 	return nil, fmt.Errorf("%v has wrong type:%v, expect:%v", name, fastjson.TypeString.String(), getType(field))
 }
 
 func (f *FastJsonConverter) extractBooleanFromValue(name string, v *fastjson.Value, field *ast.JsonStreamField) (interface{}, error) {
-	if f.isSchemaLess && field == nil {
+	if field == nil {
 		s, err := v.Bool()
 		if err != nil {
 			return nil, err
 		}
 		return s, nil
 	}
-	if !f.isSchemaLess {
-		if field == nil {
-			return nil, nil
-		}
-		if field.Type == "boolean" {
-			s, err := v.Bool()
-			if err != nil {
-				return nil, err
-			}
-			return s, nil
+	if field.Type == "boolean" {
+		s, err := v.Bool()
+		if err != nil {
+			return nil, err
 		}
+		return s, nil
 	}
 	return nil, fmt.Errorf("%v has wrong type:%v, expect:%v", name, v.Type().String(), getType(field))
 }
@@ -533,11 +435,3 @@ func getType(t *ast.JsonStreamField) string {
 		return t.Type
 	}
 }
-
-func getType2(t *ast.JsonStreamField) string {
-	if t == nil {
-		return "null"
-	} else {
-		return t.Type
-	}
-}
diff --git a/internal/converter/schema/schema.go b/internal/converter/schema/schema.go
new file mode 100644
index 00000000..4fba89d1
--- /dev/null
+++ b/internal/converter/schema/schema.go
@@ -0,0 +1,119 @@
+// 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 schema
+
+import (
+	"sync"
+
+	"github.com/lf-edge/ekuiper/v2/internal/converter/merge"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+type SchemaLayer struct {
+	sync.RWMutex
+	schema map[string]*ast.JsonStreamField
+	// ruleID -> schema
+	schemaMap map[string]map[string]*ast.JsonStreamField
+	// ruleID -> StreamName
+	streamMap map[string]string
+	// ruleID -> wildcard
+	wildcardMap map[string]struct{}
+}
+
+func NewSchemaLayer(ruleID, streamName string, schema map[string]*ast.JsonStreamField, isWildcard bool) *SchemaLayer {
+	s := &SchemaLayer{
+		schemaMap:   make(map[string]map[string]*ast.JsonStreamField),
+		streamMap:   make(map[string]string),
+		wildcardMap: make(map[string]struct{}),
+	}
+	s.schemaMap[ruleID] = schema
+	s.streamMap[ruleID] = streamName
+	if isWildcard {
+		s.wildcardMap[ruleID] = struct{}{}
+	}
+	s.schema = schema
+	s.storeSchema()
+	return s
+}
+
+func (s *SchemaLayer) storeSchema() {
+	if len(s.wildcardMap) > 0 {
+		for ruleID := range s.schemaMap {
+			merge.AddRuleSchema(ruleID, s.streamMap[ruleID], nil, true)
+		}
+		return
+	}
+	for ruleID := range s.schemaMap {
+		merge.AddRuleSchema(ruleID, s.streamMap[ruleID], s.schema, false)
+	}
+}
+
+func (s *SchemaLayer) MergeSchema(ruleID, dataSource string, newSchema map[string]*ast.JsonStreamField, isWildcard bool) error {
+	s.Lock()
+	defer s.Unlock()
+	_, ok := s.schemaMap[ruleID]
+	if ok {
+		return nil
+	}
+	s.schemaMap[ruleID] = newSchema
+	s.streamMap[ruleID] = dataSource
+	if isWildcard {
+		s.wildcardMap[ruleID] = struct{}{}
+	} else {
+		mergedSchema, err := mergeSchema(s.schema, newSchema)
+		if err != nil {
+			return err
+		}
+		s.schema = mergedSchema
+	}
+	s.storeSchema()
+	return nil
+}
+
+func (s *SchemaLayer) DetachSchema(ruleID string) error {
+	var err error
+	s.Lock()
+	defer s.Unlock()
+	_, ok := s.schemaMap[ruleID]
+	if ok {
+		merge.RemoveRuleSchema(ruleID)
+		delete(s.streamMap, ruleID)
+		delete(s.wildcardMap, ruleID)
+		delete(s.schemaMap, ruleID)
+		newSchema := make(map[string]*ast.JsonStreamField)
+		for _, schema := range s.schemaMap {
+			newSchema, err = mergeSchema(newSchema, schema)
+			if err != nil {
+				return err
+			}
+		}
+		s.schema = newSchema
+		s.storeSchema()
+	}
+	return nil
+}
+
+func (s *SchemaLayer) GetSchema() map[string]*ast.JsonStreamField {
+	s.RLock()
+	defer s.RUnlock()
+	if len(s.wildcardMap) > 0 {
+		return nil
+	}
+	return s.schema
+}
+
+func mergeSchema(originSchema, newSchema map[string]*ast.JsonStreamField) (map[string]*ast.JsonStreamField, error) {
+	return merge.MergeSchema(originSchema, newSchema)
+}
diff --git a/internal/io/file/sink.go b/internal/io/file/sink.go
index 61cefe9f..2d4cb207 100644
--- a/internal/io/file/sink.go
+++ b/internal/io/file/sink.go
@@ -27,29 +27,33 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 	"github.com/lf-edge/ekuiper/v2/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 type sinkConf struct {
-	RollingInterval    int64    `json:"rollingInterval"`
-	RollingCount       int      `json:"rollingCount"`
-	RollingNamePattern string   `json:"rollingNamePattern"` // where to add the timestamp to the file name
-	CheckInterval      int64    `json:"checkInterval"`
-	Path               string   `json:"path"` // support dynamic property, when rolling, make sure the path is updated
-	FileType           FileType `json:"fileType"`
-	HasHeader          bool     `json:"hasHeader"`
-	Delimiter          string   `json:"delimiter"`
-	Format             string   `json:"format"` // only use for validation; transformation is done in sink_node
-	Compression        string   `json:"compression"`
-	Encryption         string   `json:"encryption"`
-	Fields             []string `json:"fields"` // only use for extracting header for csv; transformation is done in sink_node
+	RollingInterval    int64          `json:"rollingInterval"`
+	RollingCount       int            `json:"rollingCount"`
+	RollingNamePattern string         `json:"rollingNamePattern"` // where to add the timestamp to the file name
+	RollingHook        string         `json:"rollingHook"`
+	RollingHookProps   map[string]any `json:"rollingHookProps"`
+	CheckInterval      int64          `json:"checkInterval"`
+	Path               string         `json:"path"` // support dynamic property, when rolling, make sure the path is updated
+	FileType           FileType       `json:"fileType"`
+	HasHeader          bool           `json:"hasHeader"`
+	Delimiter          string         `json:"delimiter"`
+	Format             string         `json:"format"` // only use for validation; transformation is done in sink_node
+	Compression        string         `json:"compression"`
+	Encryption         string         `json:"encryption"`
+	Fields             []string       `json:"fields"` // only use for extracting header for csv; transformation is done in sink_node
 }
 
 type fileSink struct {
 	c *sinkConf
 
-	mux sync.Mutex
-	fws map[string]*fileWriter
+	mux      sync.Mutex
+	fws      map[string]*fileWriter
+	rollHook modules.RollHook
 }
 
 func (m *fileSink) Provision(ctx api.StreamContext, props map[string]interface{}) error {
@@ -97,7 +101,17 @@ func (m *fileSink) Provision(ctx api.StreamContext, props map[string]interface{}
 	if _, ok := compressionTypes[c.Compression]; !ok && c.Compression != "" {
 		return fmt.Errorf("compression must be one of gzip, zstd")
 	}
-
+	if c.RollingHook != "" {
+		h, ok := modules.GetFileRollHook(c.RollingHook)
+		if !ok {
+			return fmt.Errorf("rolling hook %s is not registered", c.RollingHook)
+		}
+		err := h.Provision(ctx, c.RollingHookProps)
+		if err != nil {
+			return err
+		}
+		m.rollHook = h
+	}
 	m.c = c
 	m.fws = make(map[string]*fileWriter)
 	return nil
@@ -116,15 +130,11 @@ func (m *fileSink) Connect(ctx api.StreamContext) error {
 					m.mux.Lock()
 					for k, v := range m.fws {
 						if now.Sub(v.Start) > time.Duration(m.c.RollingInterval)*time.Millisecond {
-							ctx.GetLogger().Debugf("rolling file %s", k)
-							err := v.Close(ctx)
-							// TODO how to inform this error to the rule
+							err := m.roll(ctx, k, v)
+							// TODO how to deal with this error
 							if err != nil {
 								ctx.GetLogger().Errorf("file sink fails to close file %s with error %s.", k, err)
 							}
-							delete(m.fws, k)
-							// The file will be created when the next item comes
-							v.Written = false
 						}
 					}
 					m.mux.Unlock()
@@ -166,16 +176,9 @@ func (m *fileSink) Collect(ctx api.StreamContext, item []byte) error {
 	if m.c.RollingCount > 0 {
 		fw.Count++
 		if fw.Count >= m.c.RollingCount {
-			e = fw.Close(ctx)
-			if e != nil {
-				return e
-			}
-			delete(m.fws, fn)
-			fw.Count = 0
-			fw.Written = false
+			return m.roll(ctx, fn, fw)
 		}
 	}
-
 	return nil
 }
 
@@ -183,14 +186,40 @@ func (m *fileSink) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Closing file sink")
 	var errs []error
 	for k, v := range m.fws {
-		if e := v.Close(ctx); e != nil {
+		e := m.roll(ctx, k, v)
+		if e != nil {
 			ctx.GetLogger().Errorf("failed to close file %s: %v", k, e)
 			errs = append(errs, e)
 		}
 	}
+	if m.rollHook != nil {
+		e := m.rollHook.Close(ctx)
+		if e != nil {
+			errs = append(errs, e)
+		}
+	}
 	return errors.Join(errs...)
 }
 
+func (m *fileSink) roll(ctx api.StreamContext, k string, v *fileWriter) error {
+	ctx.GetLogger().Debugf("rolling file %s", k)
+	err := v.Close(ctx)
+	if err != nil {
+		return err
+	} else {
+		if m.rollHook != nil {
+			err = m.rollHook.RollDone(ctx, v.File.Name())
+			if err != nil {
+				return err
+			}
+		}
+	}
+	delete(m.fws, k)
+	// The file will be created when the next item comes
+	v.Written = false
+	return nil
+}
+
 // GetFws returns the file writer for the given file name, if the file writer does not exist, it will create one
 // The item is used to get the csv header if needed
 func (m *fileSink) GetFws(ctx api.StreamContext, fn string, item interface{}) (*fileWriter, error) {
diff --git a/internal/io/mqtt/connection.go b/internal/io/mqtt/connection.go
index 1d4cf864..3e6e0cad 100644
--- a/internal/io/mqtt/connection.go
+++ b/internal/io/mqtt/connection.go
@@ -87,7 +87,7 @@ func (conn *Connection) attach() {
 }
 
 // Do not call this directly. Call connection pool Detach method to release the connection
-func (conn *Connection) detach(topic string) bool {
+func (conn *Connection) detachSub(topic string) bool {
 	delete(conn.subscriptions, topic)
 	conn.Client.Unsubscribe(topic)
 	if conn.refCount.Add(-1) == 0 {
@@ -97,6 +97,14 @@ func (conn *Connection) detach(topic string) bool {
 	return false
 }
 
+func (conn *Connection) detachPub() bool {
+	if conn.refCount.Add(-1) == 0 {
+		go conn.Close()
+		return true
+	}
+	return false
+}
+
 func (conn *Connection) Subscribe(topic string, info *SubscriptionInfo) error {
 	conn.subscriptions[topic] = info
 	token := conn.Client.Subscribe(topic, info.Qos, info.Handler)
diff --git a/internal/io/mqtt/connection_pool.go b/internal/io/mqtt/connection_pool.go
index acd233a4..cc944a9c 100644
--- a/internal/io/mqtt/connection_pool.go
+++ b/internal/io/mqtt/connection_pool.go
@@ -53,11 +53,16 @@ func GetConnection(ctx api.StreamContext, props map[string]any) (*Connection, er
 	}
 }
 
-func DetachConnection(clientId string, topic string) {
+func DetachConnection(clientId string, subscribedTopic string) {
 	lock.Lock()
 	defer lock.Unlock()
 	if conn, ok := connectionPool[clientId]; ok {
-		closed := conn.detach(topic)
+		var closed bool
+		if subscribedTopic != "" {
+			closed = conn.detachSub(subscribedTopic)
+		} else {
+			closed = conn.detachPub()
+		}
 		if closed {
 			delete(connectionPool, clientId)
 		}
diff --git a/internal/io/mqtt/sink.go b/internal/io/mqtt/sink.go
index cf8d92e7..805a00ba 100644
--- a/internal/io/mqtt/sink.go
+++ b/internal/io/mqtt/sink.go
@@ -19,9 +19,6 @@ import (
 	"strings"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/connection/clients"
-	mqttClient "github.com/lf-edge/ekuiper/v2/internal/topo/connection/clients/mqtt"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 )
@@ -35,20 +32,22 @@ type AdConf struct {
 	ResendTopic string `json:"resendDestination"`
 }
 
-type MQTTSink struct {
-	adconf     *AdConf
-	config     map[string]interface{}
-	cli        io.MessageClient
-	sendParams map[string]any
+type Sink struct {
+	adconf *AdConf
+	config map[string]interface{}
+	cli    *Connection
 }
 
-func (ms *MQTTSink) Provision(_ api.StreamContext, ps map[string]any) error {
+func (ms *Sink) Provision(_ api.StreamContext, ps map[string]any) error {
+	_, err := validateConfig(ps)
+	if err != nil {
+		return err
+	}
 	adconf := &AdConf{}
-	err := cast.MapToStruct(ps, adconf)
+	err = cast.MapToStruct(ps, adconf)
 	if err != nil {
 		return err
 	}
-
 	if adconf.Tpc == "" {
 		return fmt.Errorf("mqtt sink is missing property topic")
 	}
@@ -63,24 +62,14 @@ func (ms *MQTTSink) Provision(_ api.StreamContext, ps map[string]any) error {
 		adconf.ResendTopic = adconf.Tpc
 	}
 	ms.adconf = adconf
-	ms.sendParams = map[string]any{
-		"qos":      adconf.Qos,
-		"retained": adconf.Retained,
-	}
-	mc := &mqttClient.MQTTClient{}
-	return mc.CfgValidate(ms.config)
+	return nil
 }
 
-func (ms *MQTTSink) Connect(ctx api.StreamContext) error {
-	log := ctx.GetLogger()
-	cli, err := clients.GetClient("mqtt", ms.config)
-	if err != nil {
-		log.Errorf("found error when get mqtt client config %v, error %s", ms.config, err.Error())
-		return err
-	}
+func (ms *Sink) Connect(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Connecting to mqtt server")
+	cli, err := GetConnection(ctx, ms.config)
 	ms.cli = cli
-
-	return nil
+	return err
 }
 
 func validateMQTTSinkTopic(topic string) error {
@@ -90,25 +79,27 @@ func validateMQTTSinkTopic(topic string) error {
 	return nil
 }
 
-func (ms *MQTTSink) Collect(ctx api.StreamContext, item []byte) error {
-	tpc, err := ctx.ParseTemplate(ms.adconf.Tpc, item)
+func (ms *Sink) Collect(ctx api.StreamContext, item []byte) error {
+	tpc := ms.adconf.Tpc
+	token := ms.cli.Publish(tpc, ms.adconf.Qos, ms.adconf.Retained, item)
+	err := handleToken(token)
 	if err != nil {
-		return err
-	}
-
-	if err := ms.cli.Publish(ctx, tpc, item, ms.sendParams); err != nil {
-		return errorx.NewIOErr(err.Error())
+		return errorx.NewIOErr(fmt.Sprintf("found error when publishing to topic %s: %s", ms.adconf.Tpc, err))
 	}
 	return nil
 }
 
-func (ms *MQTTSink) Close(ctx api.StreamContext) error {
-	logger := ctx.GetLogger()
-	logger.Infof("Closing mqtt sink")
+func (ms *Sink) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Info("Closing mqtt sink connector")
 	if ms.cli != nil {
-		clients.ReleaseClient(ctx, ms.cli)
+		DetachConnection(ms.cli.GetClientId(), "")
+		ms.cli = nil
 	}
 	return nil
 }
 
-var _ api.BytesCollector = &MQTTSink{}
+func GetSink() api.Sink {
+	return &Sink{}
+}
+
+var _ api.BytesCollector = &Sink{}
diff --git a/internal/io/mqtt/source.go b/internal/io/mqtt/source.go
index 5a53f1b9..7c364f3b 100644
--- a/internal/io/mqtt/source.go
+++ b/internal/io/mqtt/source.go
@@ -21,7 +21,6 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/node/metric"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
@@ -33,8 +32,7 @@ type SourceConnector struct {
 	cfg   *Conf
 	props map[string]any
 
-	cli   *Connection
-	stats metric.StatManager
+	cli *Connection
 }
 
 type Conf struct {
@@ -42,10 +40,6 @@ type Conf struct {
 	Qos   int    `json:"qos"`
 }
 
-func (ms *SourceConnector) SetupStats(stats metric.StatManager) {
-	ms.stats = stats
-}
-
 func (ms *SourceConnector) Provision(ctx api.StreamContext, props map[string]any) error {
 	cfg := &Conf{}
 	err := cast.MapToStruct(props, cfg)
@@ -116,6 +110,8 @@ func (ms *SourceConnector) Close(ctx api.StreamContext) error {
 	return nil
 }
 
-func GetSource() api.BytesSource {
+func GetSource() api.Source {
 	return &SourceConnector{}
 }
+
+var _ api.BytesSource = &SourceConnector{}
diff --git a/internal/server/server.go b/internal/server/server.go
index 537b9f69..9ff66261 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -42,7 +42,6 @@ import (
 	"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/promMetrics"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/connection/factory"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
@@ -135,7 +134,6 @@ func StartUp(Version string) {
 	startTimeStamp = time.Now().Unix()
 	createPaths()
 	conf.InitConf()
-	factory.InitClientsFactory()
 
 	undo, _ := maxprocs.Set(maxprocs.Logger(conf.Log.Infof))
 	defer undo()
diff --git a/internal/topo/connection/clients/client_registy.go b/internal/topo/connection/clients/client_registy.go
deleted file mode 100644
index cb95ba67..00000000
--- a/internal/topo/connection/clients/client_registy.go
+++ /dev/null
@@ -1,130 +0,0 @@
-// Copyright 2022 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 clients
-
-import (
-	"fmt"
-	"strings"
-	"sync"
-
-	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io"
-)
-
-type clientRegistry struct {
-	Lock                sync.Mutex
-	supportedClientType []string
-	clientFactory       map[string]ClientFactoryFunc
-	shareClientStore    map[string]ClientWrapper
-}
-
-var gClientRegistry = clientRegistry{
-	clientFactory:       make(map[string]ClientFactoryFunc),
-	Lock:                sync.Mutex{},
-	supportedClientType: make([]string, 0),
-	shareClientStore:    make(map[string]ClientWrapper),
-}
-
-func RegisterClientFactory(clientType string, creatorFunc ClientFactoryFunc) {
-	gClientRegistry.Lock.Lock()
-	gClientRegistry.clientFactory[clientType] = creatorFunc
-	gClientRegistry.supportedClientType = append(gClientRegistry.supportedClientType, clientType)
-	gClientRegistry.Lock.Unlock()
-}
-
-func getConnectionSelector(props map[string]interface{}) (ConnectionSelector string, err error) {
-	for key, v := range props {
-		if strings.EqualFold(key, "connectionSelector") {
-			if conVal, ok := v.(string); ok {
-				return strings.ToLower(conVal), nil
-			} else {
-				return "", fmt.Errorf("connectionSelector value: %v is not string", v)
-			}
-		}
-	}
-	return "", nil
-}
-
-func GetClient(connectionType string, props map[string]interface{}) (io.MessageClient, error) {
-	gClientRegistry.Lock.Lock()
-	defer gClientRegistry.Lock.Unlock()
-
-	connectSelector, err := getConnectionSelector(props)
-	if err != nil {
-		return nil, err
-	}
-	if connectSelector != "" {
-		if cliWpr, found := gClientRegistry.shareClientStore[connectSelector]; found {
-			cliWpr.AddRef()
-			return cliWpr, nil
-		}
-	}
-
-	clientCreator, ok := gClientRegistry.clientFactory[connectionType]
-	if !ok {
-		conf.Log.Errorf("can not find clientCreator for connection type : %s. only support %s", connectionType, gClientRegistry.supportedClientType)
-		return nil, fmt.Errorf("can not find clientCreator for connection type : %s. only support %s", connectionType, gClientRegistry.supportedClientType)
-	}
-
-	if connectSelector != "" {
-		selectCfg := &conf.ConSelector{
-			ConnSelectorStr: connectSelector,
-		}
-		if err := selectCfg.Init(); err != nil {
-			return nil, err
-		}
-		cf, err := selectCfg.ReadCfgFromYaml()
-		if err != nil {
-			return nil, err
-		}
-		cliWpr, err := clientCreator(cf)
-		if err != nil {
-			conf.Log.Errorf("can not create client for connection selector : %s have error %s", connectSelector, err)
-			return nil, err
-		}
-		cliWpr.SetConnectionSelector(connectSelector)
-		conf.Log.Infof("Init client wrapper for client type %s and connection selector %s", connectionType, connectSelector)
-		gClientRegistry.shareClientStore[connectSelector] = cliWpr
-		return cliWpr, nil
-	} else {
-		cliWpr, err := clientCreator(props)
-		if err != nil {
-			conf.Log.Errorf("can not create client for cfg : %v have error %s", conf.Printable(props), err)
-			return nil, err
-		}
-		conf.Log.Infof("Init client wrapper for client type %s", connectionType)
-		return cliWpr, nil
-	}
-}
-
-func ReleaseClient(ctx api.StreamContext, cli io.MessageClient) {
-	var log api.Logger
-	if ctx != nil {
-		log = ctx.GetLogger()
-	} else {
-		log = conf.Log
-	}
-	wrapper := cli.(ClientWrapper)
-	sel := wrapper.GetConnectionSelector()
-	ok := wrapper.Release(ctx)
-
-	if sel != "" && ok {
-		log.Infof("remove mqtt client wrapper for connection selector %s", sel)
-		gClientRegistry.Lock.Lock()
-		delete(gClientRegistry.shareClientStore, sel)
-		gClientRegistry.Lock.Unlock()
-	}
-}
diff --git a/internal/topo/connection/clients/mqtt/mqtt.go b/internal/topo/connection/clients/mqtt/mqtt.go
deleted file mode 100644
index 5d6933f5..00000000
--- a/internal/topo/connection/clients/mqtt/mqtt.go
+++ /dev/null
@@ -1,170 +0,0 @@
-// Copyright 2022-2023 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 (
-	"crypto/tls"
-	"fmt"
-	"strings"
-	"time"
-
-	MQTT "github.com/eclipse/paho.mqtt.golang"
-	"github.com/google/uuid"
-
-	"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"
-)
-
-type MQTTConnectionConfig struct {
-	Server   string `json:"server"`
-	PVersion string `json:"protocolVersion"`
-	ClientId string `json:"clientid"`
-	Uname    string `json:"username"`
-	Password string `json:"password"`
-}
-
-type MQTTClient struct {
-	srv      string
-	clientid string
-	pVersion uint
-	uName    string
-	password string
-	tls      *tls.Config
-
-	conn MQTT.Client
-}
-
-func (ms *MQTTClient) CfgValidate(props map[string]interface{}) error {
-	cfg := MQTTConnectionConfig{}
-
-	err := cast.MapToStruct(props, &cfg)
-	if err != nil {
-		return fmt.Errorf("failed to get config, the error is %s", err)
-	}
-
-	if cfg.Server != "" {
-		ms.srv = cfg.Server
-	} else {
-		return fmt.Errorf("missing server property")
-	}
-
-	if cfg.ClientId == "" {
-		if newUUID, err := uuid.NewUUID(); err != nil {
-			return fmt.Errorf("failed to get uuid, the error is %s", err)
-		} else {
-			ms.clientid = newUUID.String()
-		}
-	} else {
-		ms.clientid = cfg.ClientId
-	}
-	// Default to MQTT 3.1.1 or NanoMQ cannot connect
-	ms.pVersion = 4
-	if cfg.PVersion == "3.1" {
-		ms.pVersion = 3
-	}
-	tlsConfig, err := cert.GenTLSConfig(props, "mqtt")
-	if err != nil {
-		return err
-	}
-	conf.Log.Infof("Connect MQTT broker %s with TLS configs", ms.srv)
-	ms.tls = tlsConfig
-	if err := ms.checkMQTTServer(); err != nil {
-		return err
-	}
-	ms.uName = cfg.Uname
-	ms.password = strings.Trim(cfg.Password, " ")
-	return nil
-}
-
-func (ms *MQTTClient) checkMQTTServer() error {
-	if ms.tls != nil && !ms.tls.InsecureSkipVerify {
-		if !strings.HasPrefix(strings.ToLower(ms.srv), "ssl://") {
-			return fmt.Errorf("mqtt server should start with ssl:// when tls enabled")
-		}
-	}
-	return nil
-}
-
-func (ms *MQTTClient) Connect(connHandler MQTT.OnConnectHandler, lostHandler MQTT.ConnectionLostHandler) error {
-	if conf.Config.Basic.Debug {
-		MQTT.DEBUG = conf.Log
-		MQTT.ERROR = conf.Log
-	}
-	opts := MQTT.NewClientOptions().AddBroker(ms.srv).SetProtocolVersion(4)
-
-	opts = opts.SetTLSConfig(ms.tls)
-
-	if ms.uName != "" {
-		opts = opts.SetUsername(ms.uName)
-	}
-	if ms.password != "" {
-		opts = opts.SetPassword(ms.password)
-	}
-	opts = opts.SetClientID(ms.clientid)
-	opts = opts.SetAutoReconnect(true)
-	opts.OnConnect = connHandler
-	opts.OnConnectionLost = lostHandler
-	opts.OnReconnecting = func(MQTT.Client, *MQTT.ClientOptions) {
-		conf.Log.Infof("Reconnecting to mqtt broker %s client id %s", ms.srv, ms.clientid)
-	}
-
-	c := MQTT.NewClient(opts)
-	token := c.Connect()
-	err := handleToken(token)
-	if err != nil {
-		conf.Log.Errorf("The connection to mqtt broker %s failed: %s", ms.srv, err)
-		return fmt.Errorf("found error when connecting for %s: %s", ms.srv, err)
-	}
-	conf.Log.Infof("The connection to mqtt broker is established successfully for %s.", ms.srv)
-	ms.conn = c
-	return nil
-}
-
-func (ms *MQTTClient) Subscribe(topic string, qos byte, handler MQTT.MessageHandler) error {
-	token := ms.conn.Subscribe(topic, qos, handler)
-	err := handleToken(token)
-	if err != nil {
-		return fmt.Errorf("found error when subscribing to %s of topic %s: %s", ms.srv, topic, err)
-	}
-	return nil
-}
-
-func (ms *MQTTClient) Publish(topic string, qos byte, retained bool, message []byte) error {
-	token := ms.conn.Publish(topic, qos, retained, message)
-	err := handleToken(token)
-	if err != nil {
-		return fmt.Errorf("found error when publishing to %s of topic %s: %s", ms.srv, topic, err)
-	}
-	return nil
-}
-
-func handleToken(token MQTT.Token) error {
-	if !token.WaitTimeout(5 * time.Second) {
-		return errorx.NewIOErr("timeout")
-	} else if token.Error() != nil {
-		return errorx.NewIOErr(token.Error().Error())
-	}
-	return nil
-}
-
-func (ms *MQTTClient) Disconnect() error {
-	conf.Log.Infof("Closing the connection to mqtt broker for %s", ms.srv)
-	if ms.conn != nil && ms.conn.IsConnected() {
-		ms.conn.Disconnect(5000)
-	}
-	return nil
-}
diff --git a/internal/topo/connection/clients/mqtt/mqtt_wrapper.go b/internal/topo/connection/clients/mqtt/mqtt_wrapper.go
deleted file mode 100644
index a35c95d3..00000000
--- a/internal/topo/connection/clients/mqtt/mqtt_wrapper.go
+++ /dev/null
@@ -1,309 +0,0 @@
-// Copyright 2022-2023 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 (
-	"errors"
-	"fmt"
-	"strings"
-	"sync"
-
-	pahoMqtt "github.com/eclipse/paho.mqtt.golang"
-
-	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/connection/clients"
-	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
-)
-
-type mqttSubscriptionInfo struct {
-	topic        string
-	qos          byte
-	topicHandler pahoMqtt.MessageHandler
-
-	topicConsumers []*clients.ConsumerInfo
-}
-
-type mqttClientWrapper struct {
-	cli *MQTTClient
-
-	subLock sync.RWMutex
-	// topic: subscriber
-	// multiple go routine can sub same topic
-	topicSubscriptions map[string]*mqttSubscriptionInfo
-
-	// consumerId: SubscribedTopics
-	subscribers map[string]clients.SubscribedTopics
-
-	conSelector string
-
-	connected bool
-
-	refLock sync.RWMutex
-	refCnt  uint64
-}
-
-func NewMqttClientWrapper(props map[string]interface{}) (clients.ClientWrapper, error) {
-	if props == nil {
-		conf.Log.Warnf("props is nill for mqtt client wrapper")
-	}
-	client := &MQTTClient{}
-	err := client.CfgValidate(props)
-	if err != nil {
-		return nil, err
-	}
-	cliWpr := &mqttClientWrapper{
-		cli:                client,
-		subLock:            sync.RWMutex{},
-		topicSubscriptions: make(map[string]*mqttSubscriptionInfo),
-		subscribers:        make(map[string]clients.SubscribedTopics),
-		refCnt:             1,
-		// set default connected to true, because we only return the client when it is create
-		connected: true,
-	}
-	err = client.Connect(cliWpr.onConnectHandler, cliWpr.onConnectLost)
-	if err != nil {
-		return nil, err
-	}
-	conf.Log.Infof("new mqtt client created")
-	return cliWpr, nil
-}
-
-func (mc *mqttClientWrapper) onConnectHandler(_ pahoMqtt.Client) {
-	// activeSubscriptions will be empty on the first connection.
-	// On a re-connect is when the subscriptions must be re-created.
-	conf.Log.Infof("The connection to mqtt broker %s client id %s established", mc.cli.srv, mc.cli.clientid)
-	mc.subLock.Lock()
-	defer mc.subLock.Unlock()
-	mc.connected = true
-	for topic, subscription := range mc.topicSubscriptions {
-		token := mc.cli.conn.Subscribe(topic, subscription.qos, subscription.topicHandler)
-		if token.Error() != nil {
-			for _, con := range subscription.topicConsumers {
-				select {
-				case con.SubErrors <- token.Error():
-					break
-				default:
-					conf.Log.Warnf("consumer SubErrors channel full for request id %s", con.ConsumerId)
-				}
-			}
-		}
-	}
-}
-
-func (mc *mqttClientWrapper) onConnectLost(_ pahoMqtt.Client, err error) {
-	mc.subLock.Lock()
-	defer mc.subLock.Unlock()
-	mc.connected = false
-	e := fmt.Errorf("The connection to mqtt broker %s client id %s disconnected with error: %s ", mc.cli.srv, mc.cli.clientid, err.Error())
-	conf.Log.Warnf(e.Error())
-	for _, sub := range mc.topicSubscriptions {
-		if sub != nil {
-			// broadcast errors to all consumers
-			for _, consumer := range sub.topicConsumers {
-				select {
-				case consumer.SubErrors <- e:
-					break
-				default:
-					conf.Log.Warnf("consumer chan full for request id %s", consumer.ConsumerId)
-				}
-			}
-		}
-	}
-}
-
-func (mc *mqttClientWrapper) newMessageHandler(sub *mqttSubscriptionInfo) pahoMqtt.MessageHandler {
-	return func(client pahoMqtt.Client, message pahoMqtt.Message) {
-		if sub != nil {
-			// broadcast to all consumers
-			for _, consumer := range sub.topicConsumers {
-				select {
-				case consumer.ConsumerChan <- message:
-					break
-				default:
-					conf.Log.Warnf("consumer chan full for request id %s", consumer.ConsumerId)
-				}
-			}
-		}
-	}
-}
-
-func (mc *mqttClientWrapper) Ping() error {
-	if mc.cli.conn.IsConnected() {
-		return nil
-	} else {
-		return errors.New("mqtt ping failed")
-	}
-}
-
-func (mc *mqttClientWrapper) Publish(_ api.StreamContext, topic string, message []byte, params map[string]interface{}) error {
-	err := mc.checkConn()
-	if err != nil {
-		return err
-	}
-	var Qos byte = 0
-	if pq, ok := params["qos"]; ok {
-		if v, ok := pq.(byte); ok {
-			Qos = v
-		}
-	}
-	retained := false
-	if pk, ok := params["retained"]; ok {
-		if v, ok := pk.(bool); ok {
-			retained = v
-		}
-	}
-
-	err = mc.cli.Publish(topic, Qos, retained, message)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (mc *mqttClientWrapper) checkConn() error {
-	mc.subLock.RLock()
-	defer mc.subLock.RUnlock()
-	if !mc.connected {
-		return errorx.NewIOErr("mqtt client is not connected")
-	}
-	return nil
-}
-
-func (mc *mqttClientWrapper) Subscribe(c api.StreamContext, subChan []io.TopicChannel, messageErrors chan error, params map[string]interface{}) error {
-	log := c.GetLogger()
-
-	mc.subLock.Lock()
-	defer mc.subLock.Unlock()
-
-	subId := fmt.Sprintf("%s_%s_%d", c.GetRuleId(), c.GetOpId(), c.GetInstanceId())
-	if _, ok := mc.subscribers[subId]; ok {
-		return fmt.Errorf("already have subscription %s", subId)
-	}
-
-	var Qos byte = 0
-	if pq, ok := params["qos"]; ok {
-		if v, ok := pq.(byte); ok {
-			Qos = v
-		}
-	}
-
-	subTopics := clients.SubscribedTopics{
-		Topics: make([]string, 0),
-	}
-
-	for _, tpChan := range subChan {
-		tpc := tpChan.Topic
-		subTopics.Topics = append(subTopics.Topics, tpc)
-		sub, found := mc.topicSubscriptions[tpc]
-		if found {
-			sub.topicConsumers = append(sub.topicConsumers, &clients.ConsumerInfo{
-				ConsumerId:   subId,
-				ConsumerChan: tpChan.Messages,
-				SubErrors:    messageErrors,
-			})
-			log.Infof("subscription for topic %s already exists, reqId is %s, total subs %d", tpc, subId, len(sub.topicConsumers))
-		} else {
-			sub := &mqttSubscriptionInfo{
-				topic: tpc,
-				qos:   Qos,
-				topicConsumers: []*clients.ConsumerInfo{
-					{
-						ConsumerId:   subId,
-						ConsumerChan: tpChan.Messages,
-						SubErrors:    messageErrors,
-					},
-				},
-			}
-			sub.topicHandler = mc.newMessageHandler(sub)
-			log.Infof("new subscription for topic %s, reqId is %s", tpc, subId)
-			token := mc.cli.conn.Subscribe(tpc, Qos, sub.topicHandler)
-			if token.Error() != nil {
-				return token.Error()
-			}
-			mc.topicSubscriptions[tpc] = sub
-		}
-	}
-	mc.subscribers[subId] = subTopics
-
-	return nil
-}
-
-func (mc *mqttClientWrapper) unsubscribe(c api.StreamContext) {
-	log := c.GetLogger()
-	mc.subLock.Lock()
-	defer mc.subLock.Unlock()
-
-	subId := fmt.Sprintf("%s_%s_%d", c.GetRuleId(), c.GetOpId(), c.GetInstanceId())
-	subTopics, found := mc.subscribers[subId]
-	if !found {
-		log.Errorf("not found subscription id %s", subId)
-		return
-	}
-
-	for _, tpc := range subTopics.Topics {
-		if sub, found := mc.topicSubscriptions[tpc]; found {
-			for index, consumer := range sub.topicConsumers {
-				if strings.EqualFold(subId, consumer.ConsumerId) {
-					sub.topicConsumers = append(sub.topicConsumers[:index], sub.topicConsumers[index+1:]...)
-					log.Infof("unsubscription topic %s for reqId %s, total subs %d", tpc, subId, len(sub.topicConsumers))
-				}
-			}
-			if 0 == len(sub.topicConsumers) {
-				delete(mc.topicSubscriptions, tpc)
-				log.Infof("delete subscription for topic %s", tpc)
-				mc.cli.conn.Unsubscribe(tpc)
-			}
-		}
-	}
-	delete(mc.subscribers, subId)
-}
-
-func (mc *mqttClientWrapper) Release(c api.StreamContext) bool {
-	mc.unsubscribe(c)
-	return mc.deRef(c)
-}
-
-func (mc *mqttClientWrapper) SetConnectionSelector(conSelector string) {
-	mc.conSelector = conSelector
-}
-
-func (mc *mqttClientWrapper) GetConnectionSelector() string {
-	return mc.conSelector
-}
-
-func (mc *mqttClientWrapper) AddRef() {
-	mc.refLock.Lock()
-	defer mc.refLock.Unlock()
-	mc.refCnt = mc.refCnt + 1
-	conf.Log.Infof("mqtt client wrapper add refence for connection selector %s total refcount %d", mc.conSelector, mc.refCnt)
-}
-
-func (mc *mqttClientWrapper) deRef(c api.StreamContext) bool {
-	log := c.GetLogger()
-	mc.refLock.Lock()
-	defer mc.refLock.Unlock()
-
-	mc.refCnt = mc.refCnt - 1
-	log.Infof("mqtt client wrapper reference count %d", mc.refCnt)
-	if mc.refCnt == 0 {
-		_ = mc.cli.Disconnect()
-		return true
-	} else {
-		return false
-	}
-}
diff --git a/internal/topo/connection/clients/type.go b/internal/topo/connection/clients/type.go
deleted file mode 100644
index 23357066..00000000
--- a/internal/topo/connection/clients/type.go
+++ /dev/null
@@ -1,42 +0,0 @@
-// Copyright 2022 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 clients
-
-import (
-	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io"
-)
-
-type ConsumerInfo struct {
-	ConsumerId   string
-	ConsumerChan chan<- interface{}
-	SubErrors    chan error
-}
-
-type SubscribedTopics struct {
-	Topics []string
-}
-
-type ClientFactoryFunc func(props map[string]interface{}) (ClientWrapper, error)
-
-type ClientWrapper interface {
-	Subscribe(c api.StreamContext, subChan []io.TopicChannel, messageErrors chan error, params map[string]interface{}) error
-	Release(c api.StreamContext) bool
-	Publish(c api.StreamContext, topic string, message []byte, params map[string]interface{}) error
-	SetConnectionSelector(conSelector string)
-	GetConnectionSelector() string
-	AddRef()
-	Ping() error
-}
diff --git a/internal/topo/connection/factory/apply.go b/internal/topo/connection/factory/apply.go
deleted file mode 100644
index 2abf586a..00000000
--- a/internal/topo/connection/factory/apply.go
+++ /dev/null
@@ -1,27 +0,0 @@
-// Copyright 2022 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 factory
-
-import (
-	"github.com/lf-edge/ekuiper/v2/internal/topo/connection/clients"
-)
-
-var clientsFactory = make(map[string]clients.ClientFactoryFunc)
-
-func InitClientsFactory() {
-	for k, v := range clientsFactory {
-		clients.RegisterClientFactory(k, v)
-	}
-}
diff --git a/internal/topo/connection/factory/builtin.go b/internal/topo/connection/factory/builtin.go
deleted file mode 100644
index a0251137..00000000
--- a/internal/topo/connection/factory/builtin.go
+++ /dev/null
@@ -1,26 +0,0 @@
-// Copyright 2022 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 factory
-
-import (
-	"github.com/lf-edge/ekuiper/v2/internal/topo/connection/clients"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/connection/clients/mqtt"
-)
-
-func init() {
-	clientsFactory["mqtt"] = func(props map[string]interface{}) (clients.ClientWrapper, error) {
-		return mqtt.NewMqttClientWrapper(props)
-	}
-}
diff --git a/internal/topo/node/decode_op.go b/internal/topo/node/decode_op.go
index 0a9b05c0..cbab54f2 100644
--- a/internal/topo/node/decode_op.go
+++ b/internal/topo/node/decode_op.go
@@ -20,6 +20,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/converter"
+	schemaLayer "github.com/lf-edge/ekuiper/v2/internal/converter/schema"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
@@ -30,22 +31,27 @@ import (
 type DecodeOp struct {
 	*defaultSinkNode
 	converter message.Converter
+	sLayer    *schemaLayer.SchemaLayer
 }
 
 func (o *DecodeOp) AttachSchema(ctx api.StreamContext, dataSource string, schema map[string]*ast.JsonStreamField, isWildcard bool) {
 	ctx.GetLogger().Infof("attach schema to shared stream")
-	if fastDecoder, ok := o.converter.(message.SchemaMergeAbleConverter); ok {
-		if err := fastDecoder.MergeSchema(ctx.GetRuleId(), dataSource, schema, isWildcard); err != nil {
+	if fastDecoder, ok := o.converter.(message.SchemaResetAbleConverter); ok {
+		if err := o.sLayer.MergeSchema(ctx.GetRuleId(), dataSource, schema, isWildcard); err != nil {
 			ctx.GetLogger().Warnf("merge schema to shared stream failed, err: %v", err)
+		} else {
+			fastDecoder.ResetSchema(o.sLayer.GetSchema())
 		}
 	}
 }
 
 func (o *DecodeOp) DetachSchema(ruleId string) {
 	conf.Log.Infof("detach schema for shared stream rule %v", ruleId)
-	if fastDecoder, ok := o.converter.(message.SchemaMergeAbleConverter); ok {
-		if err := fastDecoder.DetachSchema(ruleId); err != nil {
+	if fastDecoder, ok := o.converter.(message.SchemaResetAbleConverter); ok {
+		if err := o.sLayer.DetachSchema(ruleId); err != nil {
 			conf.Log.Warnf("detach schema for shared stream rule %v failed, err:%v", ruleId, err)
+		} else {
+			fastDecoder.ResetSchema(o.sLayer.GetSchema())
 		}
 	}
 }
@@ -67,6 +73,7 @@ func NewDecodeOp(name, StreamName string, ruleId string, rOpt *def.RuleOption, o
 	return &DecodeOp{
 		defaultSinkNode: newDefaultSinkNode(name, rOpt),
 		converter:       converterTool,
+		sLayer:          schemaLayer.NewSchemaLayer(ruleId, StreamName, schema, isWildcard),
 	}, nil
 }
 
diff --git a/pkg/message/artifacts.go b/pkg/message/artifacts.go
index a8b90e54..bb9a127f 100644
--- a/pkg/message/artifacts.go
+++ b/pkg/message/artifacts.go
@@ -40,9 +40,8 @@ type Converter interface {
 // The columns information is defined in the source side, like file source
 type ConverterProvider func(schemaFileName string, SchemaMessageName string, delimiter string) (Converter, error)
 
-type SchemaMergeAbleConverter interface {
-	MergeSchema(key, datasource string, newSchema map[string]*ast.JsonStreamField, isWildcard bool) error
-	DetachSchema(key string) error
+type SchemaResetAbleConverter interface {
+	ResetSchema(schema map[string]*ast.JsonStreamField)
 }
 
 type ColumnSetter interface {
diff --git a/pkg/modules/file_roll_hook.go b/pkg/modules/file_roll_hook.go
new file mode 100644
index 00000000..1546a71f
--- /dev/null
+++ b/pkg/modules/file_roll_hook.go
@@ -0,0 +1,38 @@
+// 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 RollHook interface {
+	Provision(ctx api.StreamContext, props map[string]any) error
+	RollDone(ctx api.StreamContext, filePath string) error
+	api.Closable
+}
+
+type RollHookProvider func() RollHook
+
+var fileRollHooks = map[string]RollHookProvider{}
+
+func RegisterFileRollHook(name string, provider RollHookProvider) {
+	fileRollHooks[name] = provider
+}
+
+func GetFileRollHook(name string) (RollHook, bool) {
+	if p, ok := fileRollHooks[name]; ok {
+		return p(), ok
+	}
+	return nil, false
+}


Test patch:
diff --git a/internal/converter/json/convert_bench_test.go b/internal/converter/json/convert_bench_test.go
index 732a1171..934717bd 100644
--- a/internal/converter/json/convert_bench_test.go
+++ b/internal/converter/json/convert_bench_test.go
@@ -107,7 +107,7 @@ func benchmarkByFiles(filePath string, b *testing.B, schema map[string]*ast.Json
 		b.Fatalf(err.Error())
 	}
 	if schema != nil {
-		f := NewFastJsonConverter("", "", schema, false, false)
+		f := NewFastJsonConverter(schema)
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
 			f.Decode(ctx, payload)
diff --git a/internal/converter/json/converter_test.go b/internal/converter/json/converter_test.go
index 62014b37..5964d375 100644
--- a/internal/converter/json/converter_test.go
+++ b/internal/converter/json/converter_test.go
@@ -17,7 +17,6 @@ package json
 import (
 	"encoding/base64"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"os"
 	"path"
@@ -26,7 +25,6 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"github.com/lf-edge/ekuiper/v2/internal/converter/merge"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
@@ -218,7 +216,7 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
-		f := NewFastJsonConverter("", "", tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema)
 		v, err := f.Decode(ctx, tc.payload)
 		require.NoError(t, err)
 		require.Equal(t, v, tc.require)
@@ -229,7 +227,7 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter("", "", tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -389,7 +387,7 @@ func TestFastJsonConverterWithSchemaError(t *testing.T) {
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
-		f := NewFastJsonConverter("", "", tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema)
 		_, err := f.Decode(ctx, tc.payload)
 		require.Error(t, err)
 		require.Equal(t, err.Error(), tc.err.Error())
@@ -400,7 +398,7 @@ func TestFastJsonEncode(t *testing.T) {
 	a := make(map[string]int)
 	a["a"] = 1
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter("", "", nil, false, false)
+	f := NewFastJsonConverter(nil)
 	v, err := f.Encode(ctx, a)
 	require.NoError(t, err)
 	require.Equal(t, v, []byte(`{"a":1}`))
@@ -433,7 +431,7 @@ func TestArrayWithArray(t *testing.T) {
 		},
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter("", "", schema, false, false)
+	f := NewFastJsonConverter(schema)
 	v, err := f.Decode(ctx, payload)
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -614,7 +612,7 @@ func TestTypeNull(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter("", "", tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -624,7 +622,7 @@ func TestTypeNull(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter("", "", tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -641,7 +639,7 @@ func TestConvertBytea(t *testing.T) {
 		},
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter("", "", schema, false, false)
+	f := NewFastJsonConverter(schema)
 	v, err := f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -657,7 +655,7 @@ func TestConvertBytea(t *testing.T) {
 			},
 		},
 	}
-	f = NewFastJsonConverter("", "", schema, false, false)
+	f = NewFastJsonConverter(schema)
 	v, err = f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -665,213 +663,11 @@ func TestConvertBytea(t *testing.T) {
 	})
 }
 
-func TestMergeSchema(t *testing.T) {
-	testcases := []struct {
-		originSchema map[string]*ast.JsonStreamField
-		newSchema    map[string]*ast.JsonStreamField
-		resultSchema map[string]*ast.JsonStreamField
-		err          error
-	}{
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": nil,
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"b": nil,
-			},
-			resultSchema: map[string]*ast.JsonStreamField{
-				"a": nil,
-				"b": nil,
-			},
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": nil,
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"a": nil,
-			},
-			resultSchema: map[string]*ast.JsonStreamField{
-				"a": nil,
-			},
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "array",
-					Items: &ast.JsonStreamField{
-						Type: "struct",
-						Properties: map[string]*ast.JsonStreamField{
-							"b": {
-								Type: "bigint",
-							},
-						},
-					},
-				},
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "array",
-					Items: &ast.JsonStreamField{
-						Type: "struct",
-						Properties: map[string]*ast.JsonStreamField{
-							"b": {
-								Type: "string",
-							},
-						},
-					},
-				},
-			},
-			err: errors.New("column field type b between current[bigint] and new[string] are not equal"),
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "array",
-					Items: &ast.JsonStreamField{
-						Type: "bigint",
-					},
-				},
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "array",
-					Items: &ast.JsonStreamField{
-						Type: "string",
-					},
-				},
-			},
-			err: errors.New("array column field type a between current[bigint] and new[string] are not equal"),
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"b": {
-							Type: "bigint",
-						},
-					},
-				},
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"b": {
-							Type: "string",
-						},
-					},
-				},
-			},
-			err: errors.New("column field type b between current[bigint] and new[string] are not equal"),
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "bigint",
-				},
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"b": {
-					Type: "string",
-				},
-			},
-			resultSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "bigint",
-				},
-				"b": {
-					Type: "string",
-				},
-			},
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"b": {
-							Type: "bigint",
-						},
-					},
-				},
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"c": {
-							Type: "string",
-						},
-					},
-				},
-			},
-			resultSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"b": {
-							Type: "bigint",
-						},
-						"c": {
-							Type: "string",
-						},
-					},
-				},
-			},
-		},
-	}
-	for _, tc := range testcases {
-		f := NewFastJsonConverter("1", "", tc.originSchema, false, false)
-		err := f.MergeSchema("2", "", tc.newSchema, false)
-		if tc.err == nil {
-			require.NoError(t, err)
-			require.Equal(t, tc.resultSchema, f.schema)
-		} else {
-			require.Equal(t, tc.err, err)
-		}
-	}
-}
-
-func TestMergeWildcardSchema(t *testing.T) {
-	originSchema := map[string]*ast.JsonStreamField{
-		"a": {
-			Type: "bigint",
-		},
-	}
-	f := NewFastJsonConverter("1", "", originSchema, false, false)
-	require.NoError(t, f.MergeSchema("2", "", nil, true))
-	newSchema := map[string]*ast.JsonStreamField{
-		"b": {
-			Type: "bigint",
-		},
-	}
-	require.NoError(t, f.MergeSchema("3", "", newSchema, false))
-	data := map[string]interface{}{
-		"a": float64(1),
-		"b": float64(2),
-		"c": float64(3),
-	}
-	ctx := mockContext.NewMockContext("test", "op1")
-	bs, _ := json.Marshal(data)
-	d, err := f.Decode(ctx, bs)
-	require.NoError(t, err)
-	require.Equal(t, data, d)
-	require.NoError(t, f.DetachSchema("2"))
-	d, err = f.Decode(ctx, bs)
-	require.NoError(t, err)
-	require.Equal(t, map[string]interface{}{
-		"a": int64(1),
-		"b": int64(2),
-	}, d)
-}
-
 func TestSchemaless(t *testing.T) {
 	originSchema := map[string]*ast.JsonStreamField{
 		"a": nil,
 	}
-	f := NewFastJsonConverter("1", "", originSchema, false, true)
+	f := NewFastJsonConverter(originSchema)
 	testcases := []struct {
 		data   map[string]interface{}
 		expect map[string]interface{}
@@ -926,44 +722,10 @@ func TestJsonError(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, errorx.CovnerterErr, errWithCode.Code())
 	// fastjson
-	c := NewFastJsonConverter("", "", nil, false, true)
+	c := NewFastJsonConverter(nil)
 	_, err = c.Decode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok = err.(errorx.ErrorWithCode)
 	require.True(t, ok)
 	require.Equal(t, errorx.CovnerterErr, errWithCode.Code())
 }
-
-func TestAttachDetachSchema(t *testing.T) {
-	f := NewFastJsonConverter("rule1", "demo", nil, true, true)
-	err := f.MergeSchema("rule2", "demo", map[string]*ast.JsonStreamField{
-		"a": nil,
-	}, false)
-	require.NoError(t, err)
-	r := merge.GetRuleSchema("rule1")
-	er := merge.RuleSchemaResponse{
-		Schema: map[string]map[string]*ast.JsonStreamField{
-			"demo": nil,
-		},
-		Wildcard: map[string]bool{
-			"demo": true,
-		},
-	}
-	require.Equal(t, r, er)
-	r = merge.GetRuleSchema("rule2")
-	require.Equal(t, r, er)
-	// detach rule
-	require.NoError(t, f.DetachSchema("rule1"))
-	r = merge.GetRuleSchema("rule2")
-	er = merge.RuleSchemaResponse{
-		Schema: map[string]map[string]*ast.JsonStreamField{
-			"demo": {
-				"a": nil,
-			},
-		},
-		Wildcard: map[string]bool{
-			"demo": false,
-		},
-	}
-	require.Equal(t, r, er)
-}
diff --git a/internal/converter/schema/schema_test.go b/internal/converter/schema/schema_test.go
new file mode 100644
index 00000000..ecc2181f
--- /dev/null
+++ b/internal/converter/schema/schema_test.go
@@ -0,0 +1,245 @@
+// 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 schema
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/converter/merge"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+func TestMergeSchema(t *testing.T) {
+	testcases := []struct {
+		originSchema map[string]*ast.JsonStreamField
+		newSchema    map[string]*ast.JsonStreamField
+		resultSchema map[string]*ast.JsonStreamField
+		err          error
+	}{
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": nil,
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"b": nil,
+			},
+			resultSchema: map[string]*ast.JsonStreamField{
+				"a": nil,
+				"b": nil,
+			},
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": nil,
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"a": nil,
+			},
+			resultSchema: map[string]*ast.JsonStreamField{
+				"a": nil,
+			},
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "array",
+					Items: &ast.JsonStreamField{
+						Type: "struct",
+						Properties: map[string]*ast.JsonStreamField{
+							"b": {
+								Type: "bigint",
+							},
+						},
+					},
+				},
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "array",
+					Items: &ast.JsonStreamField{
+						Type: "struct",
+						Properties: map[string]*ast.JsonStreamField{
+							"b": {
+								Type: "string",
+							},
+						},
+					},
+				},
+			},
+			err: errors.New("column field type b between current[bigint] and new[string] are not equal"),
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "array",
+					Items: &ast.JsonStreamField{
+						Type: "bigint",
+					},
+				},
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "array",
+					Items: &ast.JsonStreamField{
+						Type: "string",
+					},
+				},
+			},
+			err: errors.New("array column field type a between current[bigint] and new[string] are not equal"),
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "struct",
+					Properties: map[string]*ast.JsonStreamField{
+						"b": {
+							Type: "bigint",
+						},
+					},
+				},
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "struct",
+					Properties: map[string]*ast.JsonStreamField{
+						"b": {
+							Type: "string",
+						},
+					},
+				},
+			},
+			err: errors.New("column field type b between current[bigint] and new[string] are not equal"),
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "bigint",
+				},
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "string",
+				},
+			},
+			resultSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "bigint",
+				},
+				"b": {
+					Type: "string",
+				},
+			},
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "struct",
+					Properties: map[string]*ast.JsonStreamField{
+						"b": {
+							Type: "bigint",
+						},
+					},
+				},
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "struct",
+					Properties: map[string]*ast.JsonStreamField{
+						"c": {
+							Type: "string",
+						},
+					},
+				},
+			},
+			resultSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "struct",
+					Properties: map[string]*ast.JsonStreamField{
+						"b": {
+							Type: "bigint",
+						},
+						"c": {
+							Type: "string",
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, tc := range testcases {
+		sLayer := NewSchemaLayer("r1", "s1", tc.originSchema, false)
+		err := sLayer.MergeSchema("r2", "s1", tc.newSchema, false)
+		if tc.err == nil {
+			require.NoError(t, err)
+		} else {
+			require.Equal(t, tc.err, err)
+		}
+	}
+}
+
+func TestMergeWildcardSchema(t *testing.T) {
+	originSchema := map[string]*ast.JsonStreamField{
+		"a": {
+			Type: "bigint",
+		},
+	}
+	f := NewSchemaLayer("1", "", originSchema, false)
+	require.NoError(t, f.MergeSchema("2", "", nil, true))
+	newSchema := map[string]*ast.JsonStreamField{
+		"b": {
+			Type: "bigint",
+		},
+	}
+	require.NoError(t, f.MergeSchema("3", "", newSchema, false))
+	s := f.GetSchema()
+	require.Nil(t, s)
+}
+
+func TestAttachDetachSchema(t *testing.T) {
+	f := NewSchemaLayer("rule1", "demo", nil, true)
+	err := f.MergeSchema("rule2", "demo", map[string]*ast.JsonStreamField{
+		"a": nil,
+	}, false)
+	require.NoError(t, err)
+	r := merge.GetRuleSchema("rule1")
+	er := merge.RuleSchemaResponse{
+		Schema: map[string]map[string]*ast.JsonStreamField{
+			"demo": nil,
+		},
+		Wildcard: map[string]bool{
+			"demo": true,
+		},
+	}
+	require.Equal(t, r, er)
+	r = merge.GetRuleSchema("rule2")
+	require.Equal(t, r, er)
+	// detach rule
+	require.NoError(t, f.DetachSchema("rule1"))
+	r = merge.GetRuleSchema("rule2")
+	er = merge.RuleSchemaResponse{
+		Schema: map[string]map[string]*ast.JsonStreamField{
+			"demo": {
+				"a": nil,
+			},
+		},
+		Wildcard: map[string]bool{
+			"demo": false,
+		},
+	}
+	require.Equal(t, r, er)
+}
diff --git a/internal/io/file/sink_rolling_hook_test.go b/internal/io/file/sink_rolling_hook_test.go
new file mode 100644
index 00000000..4b5889e0
--- /dev/null
+++ b/internal/io/file/sink_rolling_hook_test.go
@@ -0,0 +1,238 @@
+// 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 file
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mockclock"
+	"github.com/lf-edge/ekuiper/v2/pkg/message"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+func TestRegister(t *testing.T) {
+	modules.RegisterFileRollHook("mock", func() modules.RollHook {
+		return &MockRollingHook{}
+	})
+	_, ok := modules.GetFileRollHook("mock")
+	assert.True(t, ok)
+	_, ok = modules.GetFileRollHook("none")
+	assert.False(t, ok)
+}
+
+func TestProvision(t *testing.T) {
+	modules.RegisterFileRollHook("mock", func() modules.RollHook {
+		return &MockRollingHook{}
+	})
+	tests := []struct {
+		name string
+		c    *sinkConf
+		err  string
+		p    map[string]interface{}
+	}{
+		{
+			name: "new props",
+			c: &sinkConf{
+				CheckInterval:      500,
+				Path:               "test",
+				FileType:           CSV_TYPE,
+				Format:             message.FormatDelimited,
+				Delimiter:          ",",
+				RollingCount:       1000000,
+				RollingNamePattern: "none",
+				RollingHook:        "mock",
+				RollingHookProps: map[string]any{
+					"custom": "test",
+				},
+			},
+			p: map[string]interface{}{
+				"checkInterval":      500,
+				"path":               "test",
+				"fileType":           "csv",
+				"format":             message.FormatDelimited,
+				"rollingNamePattern": "none",
+				"rollingHook":        "mock",
+				"rollingHookProps": map[string]any{
+					"custom": "test",
+				},
+			},
+		},
+		{ // invalid rolling hook
+			name: "invalid rolling hook",
+			p: map[string]interface{}{
+				"rollingInterval": 500,
+				"rollingCount":    0,
+				"rollingHook":     "invalid",
+			},
+			err: "rolling hook invalid is not registered",
+		},
+		{
+			name: "invalid rolling props",
+			err:  "missing props custom",
+			p: map[string]interface{}{
+				"rollingInterval": 500,
+				"rollingCount":    0,
+				"rollingHook":     "mock",
+				"rollingHookProps": map[string]any{
+					"delay": 25,
+				},
+			},
+		},
+	}
+	ctx := mockContext.NewMockContext("test1", "test")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			m := &fileSink{}
+			err := m.Provision(ctx, tt.p)
+			if tt.err != "" {
+				assert.Error(t, err)
+				assert.EqualError(t, err, tt.err)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, m.c, tt.c)
+			}
+		})
+	}
+}
+
+func TestCollectRolling(t *testing.T) {
+	// Remove existing files
+	err := filepath.Walk(".", func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if filepath.Ext(path) == ".log" {
+			fmt.Println("Deleting file:", path)
+			return os.Remove(path)
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	conf.IsTesting = true
+	tests := []struct {
+		name     string
+		fname    string
+		contents [2][]byte
+	}{
+		{
+			name:  "lines",
+			fname: "test_lines.log",
+			contents: [2][]byte{
+				[]byte("{\"key\":\"value0\",\"ts\":460}\n{\"key\":\"value1\",\"ts\":910}\n{\"key\":\"value2\",\"ts\":1360}"),
+				[]byte("{\"key\":\"value3\",\"ts\":1810}\n{\"key\":\"value4\",\"ts\":2260}"),
+			},
+		},
+	}
+
+	// Create a stream context for testing
+	ctx := mockContext.NewMockContext("rule", "testRolling")
+	hook := &MockRollingHook{}
+	modules.RegisterFileRollHook("mock", func() modules.RollHook {
+		return hook
+	})
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create a file sink with the temporary file path
+			sink := &fileSink{}
+			err := sink.Provision(ctx, map[string]interface{}{
+				"path":               tt.fname,
+				"fileType":           LINES_TYPE,
+				"rollingInterval":    1000,
+				"checkInterval":      500,
+				"rollingCount":       0,
+				"rollingNamePattern": "suffix",
+				"rollingHook":        "mock",
+				"rollingHookProps": map[string]any{
+					"custom": 123,
+				},
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			mockclock.ResetClock(10)
+			err = sink.Connect(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+			c := mockclock.GetMockClock()
+
+			for i := 0; i < 5; i++ {
+				c.Add(450 * time.Millisecond)
+				m := map[string]interface{}{"key": "value" + strconv.Itoa(i), "ts": c.Now().UnixMilli()}
+				b, err := json.Marshal(m)
+				assert.NoError(t, err)
+				if err := sink.Collect(ctx, b); err != nil {
+					t.Errorf("unexpected error: %s", err)
+				}
+			}
+			c.After(2000 * time.Millisecond)
+			if err = sink.Close(ctx); err != nil {
+				t.Errorf("unexpected close error: %s", err)
+			}
+			// Should write to 2 files
+			for i := 0; i < 2; i++ {
+				// Read the contents of the temporary file and check if they match the collected items
+				var fn string
+				fn = fmt.Sprintf("test_%s-%d.log", LINES_TYPE, 460+1350*i)
+
+				var contents []byte
+				contents, err := os.ReadFile(fn)
+				if err != nil {
+					t.Fatal(err)
+				}
+				assert.Equal(t, contents, tt.contents[i])
+			}
+			exp := []string{"test_lines-460.log", "test_lines-1810.log"}
+			assert.Equal(t, hook.result, exp)
+		})
+	}
+}
+
+type MockRollingHook struct {
+	result []string
+}
+
+func (m *MockRollingHook) Provision(ctx api.StreamContext, props map[string]any) error {
+	if _, ok := props["custom"]; !ok {
+		return fmt.Errorf("missing props custom")
+	}
+	return nil
+}
+
+func (m *MockRollingHook) RollDone(ctx api.StreamContext, filePath string) error {
+	ctx.GetLogger().Infof("roll done for %s", filePath)
+	m.result = append(m.result, filePath)
+	return nil
+}
+
+func (m *MockRollingHook) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("roll hook close")
+	return nil
+}
+
+var _ modules.RollHook = &MockRollingHook{}
diff --git a/internal/io/mqtt/sink_test.go b/internal/io/mqtt/sink_test.go
index bcaf23c7..83afccd0 100644
--- a/internal/io/mqtt/sink_test.go
+++ b/internal/io/mqtt/sink_test.go
@@ -16,9 +16,9 @@ package mqtt
 
 import (
 	"fmt"
-	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
@@ -28,7 +28,7 @@ func TestSinkConfigure(t *testing.T) {
 	tests := []struct {
 		name           string
 		input          map[string]interface{}
-		expectedErr    error
+		expectedErr    string
 		expectedAdConf *AdConf
 	}{
 		{
@@ -41,26 +41,48 @@ func TestSinkConfigure(t *testing.T) {
 				"privateKeyRaw": "MTIz",
 				"server":        "123",
 			},
-			expectedErr: fmt.Errorf("tls: failed to find any PEM data in certificate input"),
+			expectedErr: "tls: failed to find any PEM data in certificate input",
+		},
+		{
+			name: "Invalid configuration",
+			input: map[string]any{
+				"server":     "123",
+				"datasource": "demo",
+				"qos":        "dd",
+			},
+			expectedErr: "1 error(s) decoding:\n\n* 'qos' expected type 'uint8', got unconvertible type 'string', value: 'dd'",
 		},
 		{
 			name: "Missing topic",
 			input: map[string]interface{}{
+				"server":      "123",
+				"qos":         1,
+				"retained":    false,
+				"compression": "zlib",
+			},
+			expectedErr: "mqtt sink is missing property topic",
+		},
+		{
+			name: "Wrong topic",
+			input: map[string]interface{}{
+				"server":      "123",
+				"topic":       "test/#",
 				"qos":         1,
 				"retained":    false,
 				"compression": "zlib",
 			},
-			expectedErr: fmt.Errorf("mqtt sink is missing property topic"),
+			expectedErr: "mqtt sink topic shouldn't contain # or +",
 		},
 		{
 			name: "Invalid QoS",
 			input: map[string]interface{}{
+				"server":      "123",
 				"topic":       "testTopic",
 				"qos":         3,
 				"retained":    false,
 				"compression": "gzip",
 			},
-			expectedErr: fmt.Errorf("invalid qos value %v, the value could be only int 0 or 1 or 2", 3),
+			expectedErr: fmt.Sprintf("invalid qos value %v, the value could be only int 0 or 1 or 2", 3),
 		},
 		{
 			name: "Valid configuration with QoS 0 and no compression",
@@ -71,7 +93,6 @@ func TestSinkConfigure(t *testing.T) {
 				"compression": "",
 				"server":      "123",
 			},
-			expectedErr: nil,
 			expectedAdConf: &AdConf{
 				Tpc:         "testTopic3",
 				Qos:         0,
@@ -88,7 +109,6 @@ func TestSinkConfigure(t *testing.T) {
 				"compression": "zlib",
 				"server":      "123",
 			},
-			expectedErr: nil,
 			expectedAdConf: &AdConf{
 				Tpc:         "testTopic4",
 				Qos:         1,
@@ -101,17 +121,14 @@ func TestSinkConfigure(t *testing.T) {
 	ctx := mockContext.NewMockContext("testsinkconfigure", "sink1")
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			ms := &MQTTSink{}
+			ms := &Sink{}
 			err := ms.Provision(ctx, tt.input)
-			if !reflect.DeepEqual(err, tt.expectedErr) {
-				t.Errorf("\n Expected error: \t%v\n \t\t\tgot: \t%v", tt.expectedErr, err)
-				return
-			}
-			if tt.expectedErr == nil {
-				if !reflect.DeepEqual(ms.adconf, tt.expectedAdConf) {
-					t.Errorf("\n Expected adConf: \t%v\n \t\t\tgot: \t%v", tt.expectedAdConf, ms.adconf)
-					return
-				}
+			if tt.expectedErr != "" {
+				assert.Error(t, err)
+				assert.EqualError(t, err, tt.expectedErr)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, ms.adconf, tt.expectedAdConf)
 			}
 		})
 	}
diff --git a/internal/io/mqtt/source_sink_test.go b/internal/io/mqtt/source_sink_test.go
new file mode 100644
index 00000000..2efd9978
--- /dev/null
+++ b/internal/io/mqtt/source_sink_test.go
@@ -0,0 +1,67 @@
+// 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"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"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) {
+	sc := GetSource().(api.BytesSource)
+	sk := GetSink().(api.BytesCollector)
+	mc := mockclock.GetMockClock()
+
+	data := [][]byte{
+		[]byte("{\"humidity\":50,\"status\":\"green\",\"temperature\":22}"),
+		[]byte("{\"humidity\":82,\"status\":\"wet\",\"temperature\":25}"),
+		[]byte("{\"humidity\":60,\"status\":\"hot\",\"temperature\":33}"),
+	}
+	result := []api.Tuple{
+		model.NewDefaultRawTuple([]byte("{\"humidity\":50,\"status\":\"green\",\"temperature\":22}"), model.DefaultMessage{
+			"topic":     "demo",
+			"messageId": uint16(0),
+			"qos":       byte(0),
+		}, mc.Now()),
+		model.NewDefaultRawTuple([]byte("{\"humidity\":82,\"status\":\"wet\",\"temperature\":25}"), model.DefaultMessage{
+			"topic":     "demo",
+			"messageId": uint16(0),
+			"qos":       byte(0),
+		}, mc.Now()),
+		model.NewDefaultRawTuple([]byte("{\"humidity\":60,\"status\":\"hot\",\"temperature\":33}"), model.DefaultMessage{
+			"topic":     "demo",
+			"messageId": uint16(0),
+			"qos":       byte(0),
+		}, mc.Now()),
+	}
+
+	// Open and subscribe before sending data
+	mock.TestSourceConnector(t, sc, map[string]any{
+		"server":     url,
+		"datasource": "demo",
+	}, result, func() {
+		err := mock.RunBytesSinkCollect(sk, data, map[string]any{
+			"server": url,
+			"topic":  "demo",
+		})
+		assert.NoError(t, err)
+	})
+}
diff --git a/internal/io/mqtt/source_test.go b/internal/io/mqtt/source_test.go
index 869288d4..2bc05f24 100644
--- a/internal/io/mqtt/source_test.go
+++ b/internal/io/mqtt/source_test.go
@@ -15,29 +15,20 @@
 package mqtt
 
 import (
-	"log"
 	"strings"
 	"testing"
-	"time"
 
-	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/connection/factory"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mockclock"
-	"github.com/lf-edge/ekuiper/v2/pkg/mock"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
-	"github.com/lf-edge/ekuiper/v2/pkg/model"
 )
 
 // 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 init() {
-	factory.InitClientsFactory()
 	testx.InitEnv("mqtt_source_connector")
 }
 
@@ -77,39 +68,3 @@ func TestProvision(t *testing.T) {
 		})
 	}
 }
-
-func TestOpen(t *testing.T) {
-	sc := &SourceConnector{}
-	mc := mockclock.GetMockClock()
-
-	// Open and subscribe before sending data
-	mock.TestSourceConnector(t, sc, map[string]any{
-		"server":     url,
-		"datasource": "demo",
-	}, []api.Tuple{
-		model.NewDefaultRawTuple([]byte("hello"), model.DefaultMessage{
-			"topic":     "demo",
-			"messageId": uint16(0),
-			"qos":       byte(0),
-		}, mc.Now()),
-	}, func() {
-		opts := mqtt.NewClientOptions().AddBroker(url)
-		client := mqtt.NewClient(opts)
-		defer client.Disconnect(0)
-		if token := client.Connect(); token.Wait() && token.Error() != nil {
-			panic(token.Error())
-		}
-		data := [][]byte{
-			[]byte("hello"),
-		}
-		topic := "demo"
-		for _, d := range data {
-			time.Sleep(time.Duration(10) * time.Millisecond)
-			if token := client.Publish(topic, 0, false, d); token.Wait() && token.Error() != nil {
-				log.Println(token.Error())
-			} else {
-				log.Println("publish success")
-			}
-		}
-	})
-}
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 517f5491..1d5a6dac 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -37,7 +37,6 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/processor"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/connection/factory"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 )
@@ -51,7 +50,6 @@ func init() {
 	uploadsDb, _ = store.GetKV("uploads")
 	uploadsStatusDb, _ = store.GetKV("uploadsStatusDb")
 	sysMetrics = NewMetrics()
-	factory.InitClientsFactory()
 }
 
 type RestTestSuite struct {
@@ -424,7 +422,6 @@ func assertErrorCode(code errorx.ErrorCode, resp []byte) (bool, error) {
 
 func (suite *RestTestSuite) Test_ruleTestHandler() {
 	suite.T().Skip()
-	factory.InitClientsFactory()
 	buf1 := bytes.NewBuffer([]byte(`{"sql":"CREATE stream alert() WITH (DATASOURCE=\"0\", TYPE=\"mqtt\")"}`))
 	req1, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/streams", buf1)
 	w1 := httptest.NewRecorder()
diff --git a/internal/topo/connection/clients/mqtt/client_mqtt_test.go b/internal/topo/connection/clients/mqtt/client_mqtt_test.go
deleted file mode 100644
index 95f0f107..00000000
--- a/internal/topo/connection/clients/mqtt/client_mqtt_test.go
+++ /dev/null
@@ -1,148 +0,0 @@
-// Copyright 2022 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 (
-	"crypto/tls"
-	"reflect"
-	"testing"
-
-	"github.com/stretchr/testify/require"
-)
-
-func TestMQttClient(t *testing.T) {
-	ms := &MQTTClient{}
-	ms.srv = "mock"
-	ms.tls = &tls.Config{InsecureSkipVerify: false}
-	err := ms.checkMQTTServer()
-	require.Error(t, err)
-	require.Equal(t, "mqtt server should start with ssl:// when tls enabled", err.Error())
-}
-
-func TestMQTTClient_CfgValidate(t *testing.T) {
-	type args struct {
-		props map[string]interface{}
-	}
-	tests := []struct {
-		name    string
-		args    args
-		wantErr bool
-	}{
-		{
-			name: "config pass",
-			args: args{
-				props: map[string]interface{}{
-					"server": "tcp:127.0.0.1",
-				},
-			},
-			wantErr: false,
-		},
-		{
-			name: "config are not case sensitive",
-			args: args{
-				props: map[string]interface{}{
-					"SERVER": "tcp:127.0.0.1",
-				},
-			},
-			wantErr: false,
-		},
-		{
-			name: "config server addr key error",
-			args: args{
-				props: map[string]interface{}{
-					"server": "tcp:127.0.0.1",
-				},
-			},
-			wantErr: false,
-		},
-		{
-			name: "config have unwanted topic fields",
-			args: args{
-				props: map[string]interface{}{
-					"server": "tcp:127.0.0.1",
-					"topic":  "demo",
-				},
-			},
-			wantErr: false,
-		},
-		{
-			name: "config no server addr",
-			args: args{
-				props: map[string]interface{}{
-					"username": "user1",
-				},
-			},
-			wantErr: true,
-		},
-		{
-			name: "config no server addr",
-			args: args{
-				props: map[string]interface{}{
-					"server": "",
-				},
-			},
-			wantErr: true,
-		},
-		{
-			name: "config miss cert key file",
-			args: args{
-				props: map[string]interface{}{
-					"server":            "tcp:127.0.0.1",
-					"certificationPath": "./not_exist.crt",
-					"privateKeyPath":    "./not_exist.key",
-				},
-			},
-			wantErr: true,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			ms := &MQTTClient{}
-			err := ms.CfgValidate(tt.args.props)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("CfgValidate() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-}
-
-func TestMQTTClient_CfgResult(t *testing.T) {
-	props := map[string]interface{}{
-		"server":   "tcp:127.0.0.1:1883",
-		"USERNAME": "demo",
-		"Password": "password",
-		"clientID": "clientid",
-	}
-
-	ms := &MQTTClient{}
-
-	_ = ms.CfgValidate(props)
-
-	if !reflect.DeepEqual("tcp:127.0.0.1:1883", ms.srv) {
-		t.Errorf("result mismatch:\n\n got=%#v\n\n", ms.srv)
-	}
-	if !reflect.DeepEqual("demo", ms.uName) {
-		t.Errorf("result mismatch:\n\n got=%#v\n\n", ms.uName)
-	}
-	if !reflect.DeepEqual("password", ms.password) {
-		t.Errorf("result mismatch:\n\n got=%#v\n\n", ms.password)
-	}
-	if !reflect.DeepEqual("clientid", ms.clientid) {
-		t.Errorf("result mismatch:\n\n got=%#v\n\n", ms.clientid)
-	}
-	if !reflect.DeepEqual(uint(4), ms.pVersion) {
-		t.Errorf("result mismatch:\n\n got=%#v\n\n", ms.pVersion)
-	}
-}