Repository: lf-edge/ekuiper
Pull request #2812 (2024-04-26 01:54:23)

refactor(converter): add context to converter API


Patch:
diff --git a/internal/converter/binary/converter.go b/internal/converter/binary/converter.go
index 6ff4e8f8..670b1789 100644
--- a/internal/converter/binary/converter.go
+++ b/internal/converter/binary/converter.go
@@ -17,6 +17,7 @@ package binary
 import (
 	"fmt"
 
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 )
@@ -29,7 +30,7 @@ func GetConverter() (message.Converter, error) {
 	return converter, nil
 }
 
-func (c *Converter) Encode(d interface{}) (b []byte, err error) {
+func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
@@ -39,7 +40,7 @@ func (c *Converter) Encode(d interface{}) (b []byte, err error) {
 	return nil, fmt.Errorf("not supported")
 }
 
-func (c *Converter) Decode(b []byte) (m interface{}, err error) {
+func (c *Converter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 	result := make(map[string]interface{})
 	result[message.DefaultField] = b
 	return result, nil
diff --git a/internal/converter/custom/converter.go b/internal/converter/custom/converter.go
deleted file mode 100644
index 0cac3f1e..00000000
--- a/internal/converter/custom/converter.go
+++ /dev/null
@@ -1,41 +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 custom
-
-import (
-	"fmt"
-
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/converter/static"
-	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
-	"github.com/lf-edge/ekuiper/v2/internal/schema"
-	"github.com/lf-edge/ekuiper/v2/pkg/message"
-)
-
-type Converter struct{}
-
-var converter = &Converter{}
-
-func LoadConverter(schemaFile string, messageName string, _ string) (message.Converter, error) {
-	conf.Log.Infof("Load custom converter from file %s, for symbol Get%s", schemaFile, messageName)
-	ffs, err := schema.GetSchemaFile(def.CUSTOM, schemaFile)
-	if err != nil {
-		return nil, err
-	}
-	if ffs.SoFile == "" {
-		return nil, fmt.Errorf("no so file found for custom schema %s", messageName)
-	}
-	return static.LoadStaticConverter(ffs.SoFile, messageName)
-}
diff --git a/internal/converter/delimited/converter.go b/internal/converter/delimited/converter.go
index 88e434b0..e9160b42 100644
--- a/internal/converter/delimited/converter.go
+++ b/internal/converter/delimited/converter.go
@@ -21,6 +21,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 )
@@ -42,7 +43,7 @@ func (c *Converter) SetColumns(cols []string) {
 }
 
 // Encode If no columns defined, the default order is sort by key
-func (c *Converter) Encode(d interface{}) (b []byte, err error) {
+func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
@@ -74,7 +75,7 @@ func (c *Converter) Encode(d interface{}) (b []byte, err error) {
 
 // Decode If the cols is not set, the default key name is col1, col2, col3...
 // The return value is always a map
-func (c *Converter) Decode(b []byte) (ma interface{}, err error) {
+func (c *Converter) Decode(ctx api.StreamContext, b []byte) (ma any, err error) {
 	tokens := strings.Split(string(b), c.delimiter)
 	m := make(map[string]interface{})
 	if len(c.cols) == 0 {
diff --git a/internal/converter/ext_converter.go b/internal/converter/ext_converter.go
index 4166d326..28564c27 100644
--- a/internal/converter/ext_converter.go
+++ b/internal/converter/ext_converter.go
@@ -17,7 +17,6 @@
 package converter
 
 import (
-	"github.com/lf-edge/ekuiper/v2/internal/converter/custom"
 	"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"
@@ -33,5 +32,4 @@ func init() {
 		}
 		return protobuf.NewConverter(ffs.SchemaFile, ffs.SoFile, schemaMessageName)
 	})
-	modules.RegisterConverter(message.FormatCustom, custom.LoadConverter)
 }
diff --git a/internal/converter/ext_mock.go b/internal/converter/ext_mock.go
index b3f34b8c..dd6fcdc5 100644
--- a/internal/converter/ext_mock.go
+++ b/internal/converter/ext_mock.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
@@ -34,13 +35,13 @@ func init() {
 // MockConverter mocks a slow converter for benchmark test
 type MockConverter struct{}
 
-func (m MockConverter) Encode(d interface{}) ([]byte, error) {
+func (m MockConverter) Encode(ctx api.StreamContext, d any) ([]byte, error) {
 	time.Sleep(10 * time.Millisecond)
 	now := timex.GetNowInMilli()
 	return []byte(fmt.Sprintf(`{"temperature":23.4,"humidity":76,"ts": %d}`, now)), nil
 }
 
-func (m MockConverter) Decode(b []byte) (interface{}, error) {
+func (m MockConverter) Decode(ctx api.StreamContext, b []byte) (any, error) {
 	time.Sleep(10 * time.Millisecond)
 	return map[string]any{
 		"temperature": 23.4,
diff --git a/internal/converter/json/converter.go b/internal/converter/json/converter.go
index bce44941..e1f9ef94 100644
--- a/internal/converter/json/converter.go
+++ b/internal/converter/json/converter.go
@@ -21,6 +21,7 @@ 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"
@@ -36,11 +37,11 @@ func GetConverter() (message.Converter, error) {
 	return converter, nil
 }
 
-func (c *Converter) Encode(d interface{}) (b []byte, err error) {
+func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 	return json.Marshal(d)
 }
 
-func (c *Converter) Decode(b []byte) (m interface{}, err error) {
+func (c *Converter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
@@ -144,11 +145,11 @@ func mergeSchema(originSchema, newSchema map[string]*ast.JsonStreamField) (map[s
 	return merge.MergeSchema(originSchema, newSchema)
 }
 
-func (c *FastJsonConverter) Encode(d interface{}) (b []byte, err error) {
+func (c *FastJsonConverter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 	return json.Marshal(d)
 }
 
-func (c *FastJsonConverter) Decode(b []byte) (m interface{}, err error) {
+func (c *FastJsonConverter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
@@ -157,7 +158,7 @@ func (c *FastJsonConverter) Decode(b []byte) (m interface{}, err error) {
 	c.RLock()
 	defer c.RUnlock()
 	if len(c.wildcardMap) > 0 {
-		return converter.Decode(b)
+		return converter.Decode(ctx, b)
 	}
 	return c.decodeWithSchema(b, c.schema)
 }
diff --git a/internal/converter/protobuf/converter.go b/internal/converter/protobuf/converter.go
index dc788e19..c2113f36 100644
--- a/internal/converter/protobuf/converter.go
+++ b/internal/converter/protobuf/converter.go
@@ -20,6 +20,7 @@ import (
 	"github.com/jhump/protoreflect/desc"
 	"github.com/jhump/protoreflect/desc/protoparse"
 
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	kconf "github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/converter/static"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
@@ -58,7 +59,7 @@ func NewConverter(schemaFile string, soFile string, messageName string) (message
 	}
 }
 
-func (c *Converter) Encode(d interface{}) (b []byte, err error) {
+func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
@@ -76,7 +77,7 @@ func (c *Converter) Encode(d interface{}) (b []byte, err error) {
 	}
 }
 
-func (c *Converter) Decode(b []byte) (m interface{}, err error) {
+func (c *Converter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
diff --git a/internal/schema/ext_inferer_custom.go b/internal/schema/ext_inferer_custom.go
deleted file mode 100644
index 6c2a8cd9..00000000
--- a/internal/schema/ext_inferer_custom.go
+++ /dev/null
@@ -1,71 +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.
-
-//go:build schema || !core
-
-package schema
-
-import (
-	"encoding/json"
-	"fmt"
-	"plugin"
-
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
-	"github.com/lf-edge/ekuiper/v2/pkg/ast"
-	"github.com/lf-edge/ekuiper/v2/pkg/cast"
-	"github.com/lf-edge/ekuiper/v2/pkg/message"
-)
-
-func init() {
-	inferes[message.FormatCustom] = InferCustom
-}
-
-func InferCustom(schemaFile string, messageName string) (ast.StreamFields, error) {
-	conf.Log.Infof("Load custom schema from file %s, for symbol Get%s", schemaFile, messageName)
-	ffs, err := GetSchemaFile(def.CUSTOM, schemaFile)
-	if err != nil {
-		return nil, err
-	}
-	if ffs.SoFile == "" {
-		return nil, fmt.Errorf("no so file found for custom schema %s", messageName)
-	}
-	sp, err := plugin.Open(ffs.SoFile)
-	if err != nil {
-		conf.Log.Errorf(fmt.Sprintf("custom schema file %s open error: %v", ffs.SoFile, err))
-		return nil, fmt.Errorf("cannot open %s: %v", ffs.SoFile, err)
-	}
-	nf, err := sp.Lookup("Get" + messageName)
-	if err != nil {
-		conf.Log.Warnf(fmt.Sprintf("cannot find schemaId %s, please check if it is exported: Get%v", messageName, err))
-		return nil, nil
-	}
-	nff, ok := nf.(func() interface{})
-	if !ok {
-		conf.Log.Errorf("exported symbol Get%s is not func to return interface{}", messageName)
-		return nil, fmt.Errorf("load custom schema %s, message %s error", ffs.SoFile, messageName)
-	}
-	mc, ok := nff().(message.SchemaProvider)
-	if ok {
-		sj := mc.GetSchemaJson()
-		var result ast.StreamFields
-		err := json.Unmarshal(cast.StringToBytes(sj), &result)
-		if err != nil {
-			return nil, fmt.Errorf("invalid schema json %s: %v", sj, err)
-		}
-		return result, nil
-	} else {
-		return nil, fmt.Errorf("get schema converter failed, exported symbol %s is not type of message.Converter", messageName)
-	}
-}
diff --git a/internal/server/meta_init.go b/internal/server/meta_init.go
index ea6b4d61..83ff0ae2 100644
--- a/internal/server/meta_init.go
+++ b/internal/server/meta_init.go
@@ -13,7 +13,6 @@
 // limitations under the License.
 
 //go:build ui || !core
-// +build ui !core
 
 package server
 
diff --git a/internal/topo/context/decoder.go b/internal/topo/context/decoder.go
deleted file mode 100644
index 36eb4121..00000000
--- a/internal/topo/context/decoder.go
+++ /dev/null
@@ -1,71 +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 context
-
-import (
-	"fmt"
-
-	"github.com/lf-edge/ekuiper/v2/pkg/message"
-)
-
-const (
-	DecodeKey = "$$decode"
-)
-
-func (c *DefaultContext) Decode(data []byte) (map[string]interface{}, error) {
-	v := c.Value(DecodeKey)
-	f, ok := v.(message.Converter)
-	if ok {
-		t, err := f.Decode(data)
-		if err != nil {
-			return nil, fmt.Errorf("decode failed: %v", err)
-		}
-		if result, ok := t.(map[string]interface{}); ok {
-			return result, nil
-		} else {
-			return nil, fmt.Errorf("only map[string]interface{} is supported but got: %v", t)
-		}
-	}
-	return nil, fmt.Errorf("no decoder configured")
-}
-
-func (c *DefaultContext) DecodeIntoList(data []byte) ([]map[string]interface{}, error) {
-	v := c.Value(DecodeKey)
-	f, ok := v.(message.Converter)
-	if ok {
-		t, err := f.Decode(data)
-		if err != nil {
-			return nil, fmt.Errorf("decode failed: %v", err)
-		}
-		switch r := t.(type) {
-		case map[string]interface{}:
-			return []map[string]interface{}{r}, nil
-		case []map[string]interface{}:
-			return r, nil
-		case []interface{}:
-			rs := make([]map[string]interface{}, len(r))
-			for i, v := range r {
-				if vc, ok := v.(map[string]interface{}); ok {
-					rs[i] = vc
-				} else {
-					return nil, fmt.Errorf("only map[string]interface{} and []map[string]interface{} is supported but got: %v", t)
-				}
-			}
-			return rs, nil
-		}
-		return nil, fmt.Errorf("only map[string]interface{} and []map[string]interface{} is supported but got: %v", t)
-	}
-	return nil, fmt.Errorf("no decoder configured")
-}
diff --git a/internal/topo/context/transform.go b/internal/topo/context/transform.go
deleted file mode 100644
index 64075369..00000000
--- a/internal/topo/context/transform.go
+++ /dev/null
@@ -1,33 +0,0 @@
-// Copyright 2021-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 context
-
-import (
-	"fmt"
-
-	"github.com/lf-edge/ekuiper/v2/internal/topo/transform"
-)
-
-const TransKey = "$$trans"
-
-// TransformOutput Lazy transform output to bytes
-func (c *DefaultContext) TransformOutput(data interface{}) ([]byte, bool, error) {
-	v := c.Value(TransKey)
-	f, ok := v.(transform.TransFunc)
-	if ok {
-		return f(data)
-	}
-	return nil, false, fmt.Errorf("no transform configured")
-}
diff --git a/internal/topo/node/compress_op.go b/internal/topo/node/compress_op.go
index 5ee1850b..3bf48aac 100644
--- a/internal/topo/node/compress_op.go
+++ b/internal/topo/node/compress_op.go
@@ -53,7 +53,7 @@ func (o *CompressOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func (o *CompressOp) Worker(_ api.Logger, item any) []any {
+func (o *CompressOp) Worker(_ api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
diff --git a/internal/topo/node/concurrent.go b/internal/topo/node/concurrent.go
index 3e600504..b9cdfec3 100644
--- a/internal/topo/node/concurrent.go
+++ b/internal/topo/node/concurrent.go
@@ -22,7 +22,7 @@ import (
 // WorkerFunc is the function to process the data
 // The function do not need to process error and control messages
 // The function must return a slice of data for each input. To omit the data, return nil
-type workerFunc func(logger api.Logger, item any) []any
+type workerFunc func(ctx api.StreamContext, item any) []any
 
 func runWithOrder(ctx api.StreamContext, node *defaultSinkNode, numWorkers int, wf workerFunc) {
 	workerChans := make([]chan any, numWorkers)
@@ -107,7 +107,7 @@ func worker(ctx api.StreamContext, node *defaultSinkNode, i int, wf workerFunc,
 				result = []any{item}
 			default:
 				node.statManager.IncTotalRecordsIn()
-				result = wf(ctx.GetLogger(), item)
+				result = wf(ctx, item)
 			}
 			select {
 			case output <- result:
diff --git a/internal/topo/node/decode_op.go b/internal/topo/node/decode_op.go
index de582abc..0a9b05c0 100644
--- a/internal/topo/node/decode_op.go
+++ b/internal/topo/node/decode_op.go
@@ -84,14 +84,14 @@ func (o *DecodeOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func (o *DecodeOp) Worker(_ api.Logger, item any) []any {
+func (o *DecodeOp) Worker(ctx api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
 	case error:
 		return []any{d}
 	case *xsql.Tuple:
-		result, err := o.converter.Decode(d.Raw)
+		result, err := o.converter.Decode(ctx, d.Raw)
 		if err != nil {
 			return []any{err}
 		}
diff --git a/internal/topo/node/decompress_op.go b/internal/topo/node/decompress_op.go
index cc56e605..e663d9ce 100644
--- a/internal/topo/node/decompress_op.go
+++ b/internal/topo/node/decompress_op.go
@@ -54,7 +54,7 @@ func (o *DecompressOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func (o *DecompressOp) Worker(_ api.Logger, item any) []any {
+func (o *DecompressOp) Worker(_ api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
diff --git a/internal/topo/node/encode_op.go b/internal/topo/node/encode_op.go
index 440b01e8..ef4bb2a7 100644
--- a/internal/topo/node/encode_op.go
+++ b/internal/topo/node/encode_op.go
@@ -54,14 +54,14 @@ func (o *EncodeOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func (o *EncodeOp) Worker(_ api.Logger, item any) []any {
+func (o *EncodeOp) Worker(ctx api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
 	case []byte:
 		return []any{d}
 	default:
-		r, err := o.converter.Encode(item)
+		r, err := o.converter.Encode(ctx, item)
 		if err != nil {
 			return []any{err}
 		} else {
diff --git a/internal/topo/node/encrypt_op.go b/internal/topo/node/encrypt_op.go
index 87306f0e..2eedb3dc 100644
--- a/internal/topo/node/encrypt_op.go
+++ b/internal/topo/node/encrypt_op.go
@@ -53,7 +53,7 @@ func (o *EncryptNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func (o *EncryptNode) Worker(_ api.Logger, item any) []any {
+func (o *EncryptNode) Worker(_ api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
diff --git a/internal/topo/node/transform_op.go b/internal/topo/node/transform_op.go
index 3d6387ef..76b9d46a 100644
--- a/internal/topo/node/transform_op.go
+++ b/internal/topo/node/transform_op.go
@@ -74,16 +74,16 @@ func (t *TransformOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 }
 
 // Worker do not need to process error and control messages
-func (t *TransformOp) Worker(logger api.Logger, item any) []any {
+func (t *TransformOp) Worker(ctx api.StreamContext, item any) []any {
 	t.statManager.ProcessTimeStart()
 	defer t.statManager.ProcessTimeEnd()
 	if ic, ok := item.(xsql.Collection); ok && t.omitIfEmpty && ic.Len() == 0 {
-		logger.Debugf("receive empty collection, dropped")
+		ctx.GetLogger().Debugf("receive empty collection, dropped")
 		return nil
 	}
 	outs := itemToMap(item)
 	if t.omitIfEmpty && (item == nil || len(outs) == 0) {
-		logger.Debugf("receive empty result %v in sink, dropped", outs)
+		ctx.GetLogger().Debugf("receive empty result %v in sink, dropped", outs)
 		return nil
 	}
 	var result []any
diff --git a/internal/topo/transform/template.go b/internal/topo/transform/template.go
index 2f941139..ffe37ff0 100644
--- a/internal/topo/transform/template.go
+++ b/internal/topo/transform/template.go
@@ -15,107 +15,13 @@
 package transform
 
 import (
-	"bytes"
 	"encoding/json"
 	"fmt"
 	"text/template"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/converter"
-	"github.com/lf-edge/ekuiper/v2/internal/converter/delimited"
-	"github.com/lf-edge/ekuiper/v2/pkg/ast"
-	"github.com/lf-edge/ekuiper/v2/pkg/message"
 )
 
-// TransFunc is the function to transform data
-type TransFunc func(interface{}) ([]byte, bool, error)
-
-func GenTransform(dt string, format string, schemaId string, delimiter string, dataField string, fields []string) (TransFunc, error) {
-	var (
-		tp  *template.Template = nil
-		c   message.Converter
-		err error
-	)
-	switch format {
-	case message.FormatProtobuf, message.FormatCustom:
-		c, err = converter.GetOrCreateConverter(&ast.Options{FORMAT: format, SCHEMAID: schemaId})
-		if err != nil {
-			return nil, err
-		}
-	case message.FormatDelimited:
-		c, err = converter.GetOrCreateConverter(&ast.Options{FORMAT: format, DELIMITER: delimiter})
-		if err != nil {
-			return nil, err
-		}
-		c.(*delimited.Converter).SetColumns(fields)
-	case message.FormatJson:
-		c, err = converter.GetOrCreateConverter(&ast.Options{FORMAT: format})
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	if dt != "" {
-		temp, err := template.New("sink").Funcs(conf.FuncMap).Parse(dt)
-		if err != nil {
-			return nil, err
-		}
-		tp = temp
-	}
-	return func(d interface{}) ([]byte, bool, error) {
-		var (
-			bs          []byte
-			transformed bool
-			selected    bool
-			m           interface{}
-			e           error
-		)
-		if tp != nil {
-			var output bytes.Buffer
-			err := tp.Execute(&output, d)
-			if err != nil {
-				return nil, false, fmt.Errorf("fail to encode data %v with dataTemplate for error %v", d, err)
-			}
-			bs = output.Bytes()
-			transformed = true
-		}
-
-		if transformed {
-			m, selected, e = TransItem(bs, dataField, fields)
-		} else {
-			m, selected, e = TransItem(d, dataField, fields)
-		}
-		if e != nil {
-			return nil, false, fmt.Errorf("fail to TransItem data %v for error %v", d, e)
-		}
-		if selected {
-			d = m
-		}
-
-		switch format {
-		case message.FormatJson:
-			if transformed && !selected {
-				return bs, true, nil
-			}
-			outBytes, err := c.Encode(d)
-			return outBytes, transformed || selected, err
-		case message.FormatProtobuf, message.FormatCustom, message.FormatDelimited:
-			if transformed && !selected {
-				m := make(map[string]interface{})
-				err := json.Unmarshal(bs, &m)
-				if err != nil {
-					return nil, false, fmt.Errorf("fail to decode data %s after applying dataTemplate for error %v", string(bs), err)
-				}
-				d = m
-			}
-			outBytes, err := c.Encode(d)
-			return outBytes, transformed || selected, err
-		default: // should not happen
-			return nil, false, fmt.Errorf("unsupported format %v", format)
-		}
-	}, nil
-}
-
 func GenTp(dt string) (*template.Template, error) {
 	return template.New("sink").Funcs(conf.FuncMap).Parse(dt)
 }
diff --git a/internal/xsql/parser.go b/internal/xsql/parser.go
index 3b63f104..e971a062 100644
--- a/internal/xsql/parser.go
+++ b/internal/xsql/parser.go
@@ -25,6 +25,7 @@ import (
 	"github.com/golang-collections/collections/stack"
 
 	"github.com/lf-edge/ekuiper/v2/internal/binder/function"
+	_ "github.com/lf-edge/ekuiper/v2/internal/converter"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
diff --git a/pkg/message/artifacts.go b/pkg/message/artifacts.go
index 3708a42c..a8b90e54 100644
--- a/pkg/message/artifacts.go
+++ b/pkg/message/artifacts.go
@@ -15,6 +15,7 @@
 package message
 
 import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 )
 
@@ -31,8 +32,8 @@ const (
 
 // Converter converts bytes & map or []map according to the schema
 type Converter interface {
-	Encode(d interface{}) ([]byte, error)
-	Decode(b []byte) (interface{}, error)
+	Encode(ctx api.StreamContext, d any) ([]byte, error)
+	Decode(ctx api.StreamContext, b []byte) (any, error)
 }
 
 // ConverterProvider The format, schema information are passed in by stream options
diff --git a/pkg/mock/context/mock.go b/pkg/mock/context/mock.go
index 3d6076e5..dcd63c10 100644
--- a/pkg/mock/context/mock.go
+++ b/pkg/mock/context/mock.go
@@ -20,14 +20,11 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/state"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/transform"
 )
 
 func NewMockContext(ruleId string, opId string) api.StreamContext {
 	contextLogger := conf.Log.WithField("rule", ruleId)
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
-	tf, _ := transform.GenTransform("", "json", "", "", "", []string{})
-	ctx = context.WithValue(ctx, context.TransKey, tf)
 	tempStore, _ := state.CreateStore(ruleId, def.AtMostOnce)
 	return ctx.WithMeta(ruleId, opId, tempStore)
 }
diff --git a/pkg/timex/time.go b/pkg/timex/time.go
index 58fbf14b..0bddadd7 100644
--- a/pkg/timex/time.go
+++ b/pkg/timex/time.go
@@ -69,3 +69,17 @@ func GetNowInMilli() int64 {
 func GetNow() time.Time {
 	return Clock.Now()
 }
+
+// Mock time, only use in test
+
+func Set(t int64) {
+	if IsTesting {
+		Clock.(*clock.Mock).Set(time.UnixMilli(t))
+	}
+}
+
+func Add(d time.Duration) {
+	if IsTesting {
+		Clock.(*clock.Mock).Add(d)
+	}
+}


Test patch:
diff --git a/.github/workflows/run_test_case.yaml b/.github/workflows/run_test_case.yaml
index 688719ac..495cb222 100644
--- a/.github/workflows/run_test_case.yaml
+++ b/.github/workflows/run_test_case.yaml
@@ -45,12 +45,6 @@ jobs:
         mkdir -p log
         mkdir -p plugins/sinks
         sed -i -r "s/debug: .*/debug: true/1" etc/kuiper.yaml
-        go build --buildmode=plugin -o data/test/helloworld.so internal/converter/protobuf/test/*.go
-        go build --buildmode=plugin -o data/myFormat.so internal/converter/custom/test/*.go
-        mkdir -p data/custom
-        cp data/myFormat.so data/custom/myFormat.so
-        mkdir -p data/schema
-        cp data/myFormat.so data/schema/myFormat.so
     - name: Run test case
       run: |
         make failpoint-enable
diff --git a/internal/converter/binary/converter_test.go b/internal/converter/binary/converter_test.go
index 2f4ffda4..ef544ea5 100644
--- a/internal/converter/binary/converter_test.go
+++ b/internal/converter/binary/converter_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestMessageDecode(t *testing.T) {
@@ -42,8 +43,9 @@ func TestMessageDecode(t *testing.T) {
 		},
 	}
 	conv, _ := GetConverter()
+	ctx := mockContext.NewMockContext("test", "op1")
 	for i, tt := range tests {
-		result, err := conv.Decode(tt.payload)
+		result, err := conv.Decode(ctx, tt.payload)
 		if err != nil {
 			t.Errorf("%d decode error: %v", i, err)
 		}
@@ -54,7 +56,8 @@ func TestMessageDecode(t *testing.T) {
 }
 
 func TestError(t *testing.T) {
-	_, err := converter.Encode(nil)
+	ctx := mockContext.NewMockContext("test", "op1")
+	_, err := converter.Encode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok := err.(errorx.ErrorWithCode)
 	require.True(t, ok)
diff --git a/internal/converter/custom/converter_test.go b/internal/converter/custom/converter_test.go
deleted file mode 100644
index 01cc3bbf..00000000
--- a/internal/converter/custom/converter_test.go
+++ /dev/null
@@ -1,142 +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 custom
-
-import (
-	"fmt"
-	"os"
-	"path/filepath"
-	"reflect"
-	"testing"
-
-	"github.com/gdexlab/go-render/render"
-
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/schema"
-	"github.com/lf-edge/ekuiper/v2/internal/testx"
-)
-
-func init() {
-	testx.InitEnv("custom")
-}
-
-func TestCustomConverter(t *testing.T) {
-	dataDir, err := conf.GetDataLoc()
-	if err != nil {
-		t.Fatal(err)
-	}
-	etcDir := filepath.Join(dataDir, "schemas", "custom")
-	err = os.MkdirAll(etcDir, os.ModePerm)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = os.RemoveAll(etcDir)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}()
-	// build the so file into data/test prior to running the test
-	// Copy the helloworld.so
-	bytesRead, err := os.ReadFile(filepath.Join(dataDir, "myFormat.so"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = os.WriteFile(filepath.Join(etcDir, "myFormat.so"), bytesRead, 0o755)
-	if err != nil {
-		t.Fatal(err)
-	}
-	schema.InitRegistry()
-	testEncode(t)
-	testDecode(t)
-}
-
-func testEncode(t *testing.T) {
-	c, err := LoadConverter("myFormat", "Sample", "")
-	if err != nil {
-		t.Fatal(err)
-	}
-	tests := []struct {
-		m map[string]interface{}
-		r []byte
-		e string
-	}{
-		{
-			m: map[string]interface{}{
-				"id":   12,
-				"name": "test",
-			},
-			r: []byte(`{"id":12,"name":"test","age":0,"hobbies":{"indoor":null,"outdoor":null}}`),
-		}, {
-			m: map[string]interface{}{
-				"id":   7,
-				"name": "John Doe",
-				"age":  22,
-				"hobbies": map[string]interface{}{
-					"indoor": []string{
-						"Chess",
-					},
-					"outdoor": []string{
-						"Basketball",
-					},
-				},
-			},
-			r: []byte(`{"id":7,"name":"John Doe","age":22,"hobbies":{"indoor":["Chess"],"outdoor":["Basketball"]}}`),
-		},
-	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-	for i, tt := range tests {
-		a, err := c.Encode(tt.m)
-		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
-			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
-		} else if tt.e == "" && !reflect.DeepEqual(tt.r, a) {
-			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%s\n\ngot=%s\n\n", i, tt.r, a)
-		}
-	}
-}
-
-func testDecode(t *testing.T) {
-	c, err := LoadConverter("myFormat", "Sample", "")
-	if err != nil {
-		t.Fatal(err)
-	}
-	tests := []struct {
-		m map[string]interface{}
-		r []byte
-		e string
-	}{
-		{
-			m: map[string]interface{}{
-				"name": "test",
-				"age":  int64(0),
-				"id":   int64(0),
-				"hobbies": map[string]interface{}{
-					"indoor":  []string(nil),
-					"outdoor": []string(nil),
-				},
-			},
-			r: []byte(`{"name":"test"}`),
-		},
-	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-	for i, tt := range tests {
-		a, err := c.Decode(tt.r)
-		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
-			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
-		} else if tt.e == "" && !reflect.DeepEqual(tt.m, a) {
-			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%v\n\ngot=%v\n\n", i, render.AsCode(tt.m), render.AsCode(a))
-		}
-	}
-}
diff --git a/internal/converter/custom/test/Makefile b/internal/converter/custom/test/Makefile
deleted file mode 100644
index 22f23892..00000000
--- a/internal/converter/custom/test/Makefile
+++ /dev/null
@@ -1,5 +0,0 @@
-.PHONY: build
-build:
-	@echo "Start build"
-	CGO_ENABLED=1 go build -trimpath --buildmode=plugin
-	@echo "Build successfully"
\ No newline at end of file
diff --git a/internal/converter/custom/test/go.mod b/internal/converter/custom/test/go.mod
deleted file mode 100644
index cf3a2978..00000000
--- a/internal/converter/custom/test/go.mod
+++ /dev/null
@@ -1,5 +0,0 @@
-module test
-
-go 1.22
-
-require github.com/mitchellh/mapstructure v1.5.0
diff --git a/internal/converter/custom/test/go.sum b/internal/converter/custom/test/go.sum
deleted file mode 100644
index 59f4b8e6..00000000
--- a/internal/converter/custom/test/go.sum
+++ /dev/null
@@ -1,2 +0,0 @@
-github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
-github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
diff --git a/internal/converter/custom/test/myformat.go b/internal/converter/custom/test/myformat.go
deleted file mode 100644
index 6c33ef78..00000000
--- a/internal/converter/custom/test/myformat.go
+++ /dev/null
@@ -1,117 +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 main
-
-import (
-	"encoding/json"
-	"fmt"
-
-	"github.com/mitchellh/mapstructure"
-)
-
-type Hobbies struct {
-	Indoor  []string `json:"indoor"`
-	Outdoor []string `json:"outdoor"`
-}
-
-type Sample struct {
-	Id      int64   `json:"id"`
-	Name    string  `json:"name"`
-	Age     int64   `json:"age"`
-	Hobbies Hobbies `json:"hobbies"`
-}
-
-func (x *Sample) GetSchemaJson() string {
-	// return a static schema
-	return `{
-		"id": {
-			"type": "bigint"
-	},
-		"name": {
-			"type": "string"
-	},
-		"age": {
-			"type": "bigint"
-	},
-		"hobbies": {
-			"type": "struct",
-			"properties": {
-			"indoor": {
-				"type": "array",
-					"items": {
-						"type": "string"
-				}
-			},
-			"outdoor": {
-				"type": "array",
-					"items": {
-						"type": "string"
-				}
-			}
-		}
-	}
-	}`
-}
-
-func (x *Sample) Encode(d interface{}) ([]byte, error) {
-	switch r := d.(type) {
-	case map[string]interface{}:
-		result := &Sample{}
-		err := MapToStructStrict(r, result)
-		if err != nil {
-			return nil, err
-		}
-		return json.Marshal(result)
-	default:
-		return nil, fmt.Errorf("unsupported type %v, must be a map", d)
-	}
-}
-
-func (x *Sample) Decode(b []byte) (interface{}, error) {
-	result := &Sample{}
-	// check error
-	err := json.Unmarshal(b, &result)
-	if err != nil {
-		return nil, err
-	}
-	// convert struct to map
-	hobbyMap := make(map[string]interface{}, 2)
-	hobbyMap["indoor"] = result.Hobbies.Indoor
-	hobbyMap["outdoor"] = result.Hobbies.Outdoor
-	resultMap := make(map[string]interface{}, 4)
-	resultMap["id"] = result.Id
-	resultMap["name"] = result.Name
-	resultMap["age"] = result.Age
-	resultMap["hobbies"] = hobbyMap
-	return resultMap, err
-}
-
-func MapToStructStrict(input, output interface{}) error {
-	config := &mapstructure.DecoderConfig{
-		ErrorUnused: true,
-		TagName:     "json",
-		Result:      output,
-	}
-	decoder, err := mapstructure.NewDecoder(config)
-	if err != nil {
-		return err
-	}
-
-	return decoder.Decode(input)
-}
-
-func GetSample() interface{} {
-	return &Sample{}
-}
diff --git a/internal/converter/delimited/converter_test.go b/internal/converter/delimited/converter_test.go
index 5c93740f..bd3f37d6 100644
--- a/internal/converter/delimited/converter_test.go
+++ b/internal/converter/delimited/converter_test.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestEncode(t *testing.T) {
@@ -55,12 +56,13 @@ func TestEncode(t *testing.T) {
 		},
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	ctx := mockContext.NewMockContext("test", "op1")
 	for i, tt := range tests {
 		c, err := NewConverter(":")
 		if err != nil {
 			t.Fatal(err)
 		}
-		a, err := c.Encode(tt.m)
+		a, err := c.Encode(ctx, tt.m)
 		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
 			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
 		} else if tt.e == "" && !reflect.DeepEqual(tt.r, a) {
@@ -102,14 +104,15 @@ func TestDecode(t *testing.T) {
 		},
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	ctx := mockContext.NewMockContext("test", "op1")
 	for i, tt := range tests {
-		a, err := c.Decode(tt.r)
+		a, err := c.Decode(ctx, tt.r)
 		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
 			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
 		} else if tt.e == "" && !reflect.DeepEqual(tt.m, a) {
 			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%v\n\ngot=%v\n\n", i, tt.m, a)
 		}
-		b, err := ch.Decode(tt.r)
+		b, err := ch.Decode(ctx, tt.r)
 		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
 			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
 		} else if tt.e == "" && !reflect.DeepEqual(tt.nm, b) {
@@ -121,7 +124,8 @@ func TestDecode(t *testing.T) {
 func TestError(t *testing.T) {
 	converter, err := NewConverter(",")
 	require.NoError(t, err)
-	_, err = converter.Encode(nil)
+	ctx := mockContext.NewMockContext("test", "op1")
+	_, err = converter.Encode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok := err.(errorx.ErrorWithCode)
 	require.True(t, ok)
diff --git a/internal/converter/ext_mock_test.go b/internal/converter/ext_mock_test.go
index 1e733641..666554fb 100644
--- a/internal/converter/ext_mock_test.go
+++ b/internal/converter/ext_mock_test.go
@@ -20,15 +20,18 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestEncodeWithMockConverter(t *testing.T) {
 	mockConverter := MockConverter{}
+	ctx := mockContext.NewMockContext("test", "op1")
 	data := map[string]interface{}{
 		"temperature": 23.4,
 		"humidity":    76,
 	}
-	encodedData, err := mockConverter.Encode(data)
+	encodedData, err := mockConverter.Encode(ctx, data)
 
 	assert.Nil(t, err)
 	assert.Contains(t, string(encodedData), `"temperature":23.4`)
@@ -37,8 +40,9 @@ func TestEncodeWithMockConverter(t *testing.T) {
 
 func TestDecodeWithMockConverter(t *testing.T) {
 	mockConverter := MockConverter{}
+	ctx := mockContext.NewMockContext("test", "op1")
 	data := []byte(`{"temperature":23.4,"humidity":76,"ts":1633027200000}`)
-	decodedData, err := mockConverter.Decode(data)
+	decodedData, err := mockConverter.Decode(ctx, data)
 
 	assert.Nil(t, err)
 	assert.Equal(t, 23.4, decodedData.(map[string]interface{})["temperature"])
diff --git a/internal/converter/json/convert_bench_test.go b/internal/converter/json/convert_bench_test.go
index 3989b310..732a1171 100644
--- a/internal/converter/json/convert_bench_test.go
+++ b/internal/converter/json/convert_bench_test.go
@@ -19,6 +19,7 @@ import (
 	"testing"
 
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func BenchmarkSimpleTuples(b *testing.B) {
@@ -100,6 +101,7 @@ func BenchmarkComplexTuplesWithSchema(b *testing.B) {
 }
 
 func benchmarkByFiles(filePath string, b *testing.B, schema map[string]*ast.JsonStreamField) {
+	ctx := mockContext.NewMockContext("test", "test")
 	payload, err := os.ReadFile(filePath)
 	if err != nil {
 		b.Fatalf(err.Error())
@@ -108,12 +110,12 @@ func benchmarkByFiles(filePath string, b *testing.B, schema map[string]*ast.Json
 		f := NewFastJsonConverter("", "", schema, false, false)
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
-			f.Decode(payload)
+			f.Decode(ctx, payload)
 		}
 	} else {
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
-			converter.Decode(payload)
+			converter.Decode(ctx, payload)
 		}
 	}
 }
diff --git a/internal/converter/json/converter_test.go b/internal/converter/json/converter_test.go
index 5996ff97..62014b37 100644
--- a/internal/converter/json/converter_test.go
+++ b/internal/converter/json/converter_test.go
@@ -29,6 +29,7 @@ import (
 	"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"
 )
 
 func TestMessageDecode(t *testing.T) {
@@ -65,8 +66,9 @@ func TestMessageDecode(t *testing.T) {
 		},
 	}
 	conv, _ := GetConverter()
+	ctx := mockContext.NewMockContext("test", "op1")
 	for i, tt := range tests {
-		result, err := conv.Decode(tt.payload)
+		result, err := conv.Decode(ctx, tt.payload)
 		if err != nil {
 			t.Errorf("%d decode error: %v", i, err)
 		}
@@ -214,9 +216,10 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 			},
 		},
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
 		f := NewFastJsonConverter("", "", tc.schema, false, false)
-		v, err := f.Decode(tc.payload)
+		v, err := f.Decode(ctx, tc.payload)
 		require.NoError(t, err)
 		require.Equal(t, v, tc.require)
 	}
@@ -227,7 +230,7 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 			tc.require,
 		}
 		f := NewFastJsonConverter("", "", tc.schema, false, false)
-		v, err := f.Decode(arrayPayload)
+		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
 	}
@@ -384,10 +387,10 @@ func TestFastJsonConverterWithSchemaError(t *testing.T) {
 			err: fmt.Errorf("array has wrong type:true, expect:datetime"),
 		},
 	}
-
+	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
 		f := NewFastJsonConverter("", "", tc.schema, false, false)
-		_, err := f.Decode(tc.payload)
+		_, err := f.Decode(ctx, tc.payload)
 		require.Error(t, err)
 		require.Equal(t, err.Error(), tc.err.Error())
 	}
@@ -396,8 +399,9 @@ func TestFastJsonConverterWithSchemaError(t *testing.T) {
 func TestFastJsonEncode(t *testing.T) {
 	a := make(map[string]int)
 	a["a"] = 1
+	ctx := mockContext.NewMockContext("test", "op1")
 	f := NewFastJsonConverter("", "", nil, false, false)
-	v, err := f.Encode(a)
+	v, err := f.Encode(ctx, a)
 	require.NoError(t, err)
 	require.Equal(t, v, []byte(`{"a":1}`))
 }
@@ -428,8 +432,9 @@ func TestArrayWithArray(t *testing.T) {
 			},
 		},
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	f := NewFastJsonConverter("", "", schema, false, false)
-	v, err := f.Decode(payload)
+	v, err := f.Decode(ctx, payload)
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
 		"a": []interface{}{
@@ -603,13 +608,14 @@ func TestTypeNull(t *testing.T) {
 			},
 		},
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
 		arrayPayload := []byte(fmt.Sprintf("[%s]", string(tc.payload)))
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
 		f := NewFastJsonConverter("", "", tc.schema, false, false)
-		v, err := f.Decode(arrayPayload)
+		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
 	}
@@ -619,7 +625,7 @@ func TestTypeNull(t *testing.T) {
 			tc.require,
 		}
 		f := NewFastJsonConverter("", "", tc.schema, false, false)
-		v, err := f.Decode(arrayPayload)
+		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
 	}
@@ -634,8 +640,9 @@ func TestConvertBytea(t *testing.T) {
 			Type: "bytea",
 		},
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	f := NewFastJsonConverter("", "", schema, false, false)
-	v, err := f.Decode([]byte(payload))
+	v, err := f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
 		"a": []byte(origin),
@@ -651,7 +658,7 @@ func TestConvertBytea(t *testing.T) {
 		},
 	}
 	f = NewFastJsonConverter("", "", schema, false, false)
-	v, err = f.Decode([]byte(payload))
+	v, err = f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
 		"a": []interface{}{[]byte(origin)},
@@ -846,12 +853,13 @@ func TestMergeWildcardSchema(t *testing.T) {
 		"b": float64(2),
 		"c": float64(3),
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	bs, _ := json.Marshal(data)
-	d, err := f.Decode(bs)
+	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(bs)
+	d, err = f.Decode(ctx, bs)
 	require.NoError(t, err)
 	require.Equal(t, map[string]interface{}{
 		"a": int64(1),
@@ -901,23 +909,25 @@ func TestSchemaless(t *testing.T) {
 			},
 		},
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
 		bs, _ := json.Marshal(tc.data)
-		v, err := f.Decode(bs)
+		v, err := f.Decode(ctx, bs)
 		require.NoError(t, err)
 		require.Equal(t, tc.expect, v)
 	}
 }
 
 func TestJsonError(t *testing.T) {
-	_, err := converter.Decode(nil)
+	ctx := mockContext.NewMockContext("test", "op1")
+	_, err := converter.Decode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok := err.(errorx.ErrorWithCode)
 	require.True(t, ok)
 	require.Equal(t, errorx.CovnerterErr, errWithCode.Code())
 	// fastjson
 	c := NewFastJsonConverter("", "", nil, false, true)
-	_, err = c.Decode(nil)
+	_, err = c.Decode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok = err.(errorx.ErrorWithCode)
 	require.True(t, ok)
diff --git a/internal/converter/protobuf/converter_test.go b/internal/converter/protobuf/converter_test.go
index 37bea85c..ba2b0247 100644
--- a/internal/converter/protobuf/converter_test.go
+++ b/internal/converter/protobuf/converter_test.go
@@ -16,30 +16,28 @@ package protobuf
 
 import (
 	"fmt"
-	"os"
-	"path/filepath"
 	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/schema"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestOneOfDecode(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test5.proto", "", "Book")
 	require.NoError(t, err)
-	v, err := c.Decode([]byte{0x0A, 0x03, 0x31, 0x32, 0x33, 0x1A, 0x04, 0x31, 0x32, 0x33, 0x34})
+	v, err := c.Decode(ctx, []byte{0x0A, 0x03, 0x31, 0x32, 0x33, 0x1A, 0x04, 0x31, 0x32, 0x33, 0x34})
 	require.NoError(t, err)
 	require.Equal(t, map[string]interface{}{
 		"a": "123", "c": "1234",
 	}, v)
 
-	v, err = c.Decode([]byte{0x0A, 0x03, 0x31, 0x32, 0x33, 0x22, 0x04, 0x31, 0x32, 0x33, 0x34})
+	v, err = c.Decode(ctx, []byte{0x0A, 0x03, 0x31, 0x32, 0x33, 0x22, 0x04, 0x31, 0x32, 0x33, 0x34})
 	require.NoError(t, err)
 	require.Equal(t, map[string]interface{}{
 		"a": "123", "d": "1234",
@@ -47,6 +45,7 @@ func TestOneOfDecode(t *testing.T) {
 }
 
 func TestEncode(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test1.proto", "", "Person")
 	if err != nil {
 		t.Fatal(err)
@@ -84,7 +83,7 @@ func TestEncode(t *testing.T) {
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 	for i, tt := range tests {
-		a, err := c.Encode(tt.m)
+		a, err := c.Encode(ctx, tt.m)
 		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
 			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
 		} else if tt.e == "" && !reflect.DeepEqual(tt.r, a) {
@@ -94,6 +93,7 @@ func TestEncode(t *testing.T) {
 }
 
 func TestEmbedType(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test3.proto", "", "DrivingData")
 	if err != nil {
 		t.Fatal(err)
@@ -126,16 +126,17 @@ func TestEmbedType(t *testing.T) {
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 	for _, tt := range tests {
-		a, err := c.Encode(tt.m)
+		a, err := c.Encode(ctx, tt.m)
 		assert.NoError(t, err)
 		assert.Equal(t, tt.r, a)
-		m, err := c.Decode(a)
+		m, err := c.Decode(ctx, a)
 		assert.NoError(t, err)
 		assert.Equal(t, tt.d, m)
 	}
 }
 
 func TestDecode(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test1.proto", "", "Person")
 	if err != nil {
 		t.Fatal(err)
@@ -169,79 +170,15 @@ func TestDecode(t *testing.T) {
 
 	for i, tt := range tests {
 		t.Run(fmt.Sprintf("test %d", i), func(t *testing.T) {
-			a, err := c.Decode(tt.r)
+			a, err := c.Decode(ctx, tt.r)
 			assert.NoError(t, err)
 			assert.Equal(t, tt.m, a)
 		})
 	}
 }
 
-func TestStatic(t *testing.T) {
-	dataDir, err := conf.GetDataLoc()
-	if err != nil {
-		t.Fatal(err)
-	}
-	etcDir := filepath.Join(dataDir, "schemas", "custom")
-	err = os.MkdirAll(etcDir, os.ModePerm)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = os.RemoveAll(etcDir)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}()
-	// build the so file into data/test prior to running the test
-	// Copy the helloworld.so
-	bytesRead, err := os.ReadFile(filepath.Join(dataDir, "helloworld.so"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = os.WriteFile(filepath.Join(etcDir, "helloworld.so"), bytesRead, 0o755)
-	if err != nil {
-		t.Fatal(err)
-	}
-	schema.InitRegistry()
-	c, err := NewConverter("../../schema/test/test1.proto", "../../../data/test/schemas/custom/helloworld.so", "HelloReply")
-	if err != nil {
-		t.Fatal(err)
-	}
-	tests := []struct {
-		m map[string]interface{}
-		r []byte
-		e string
-	}{
-		{
-			m: map[string]interface{}{
-				"message": "test",
-			},
-			r: []byte{0x0a, 0x04, 0x74, 0x65, 0x73, 0x74},
-		}, {
-			m: map[string]interface{}{
-				"message": "another test 2",
-			},
-			r: []byte{0x0a, 0x0e, 0x61, 0x6e, 0x6f, 0x74, 0x68, 0x65, 0x72, 0x20, 0x74, 0x65, 0x73, 0x74, 0x20, 0x32},
-		},
-	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-	for i, tt := range tests {
-		a, err := c.Encode(tt.m)
-		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
-			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
-		} else if tt.e == "" && !reflect.DeepEqual(tt.r, a) {
-			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%x\n\ngot=%x\n\n", i, tt.r, a)
-		}
-		m, err := c.Decode(tt.r)
-		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
-			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
-		} else if tt.e == "" && !reflect.DeepEqual(tt.m, m) {
-			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%v\n\ngot=%v\n\n", i, tt.m, m)
-		}
-	}
-}
-
 func TestDecodeProto3(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test4.proto", "", "Classroom")
 	if err != nil {
 		t.Fatal(err)
@@ -276,7 +213,7 @@ func TestDecodeProto3(t *testing.T) {
 
 	for i, tt := range tests {
 		t.Run(fmt.Sprintf("test %d", i), func(t *testing.T) {
-			a, err := c.Decode(tt.r)
+			a, err := c.Decode(ctx, tt.r)
 			assert.NoError(t, err)
 			assert.Equal(t, tt.m, a)
 		})
@@ -284,6 +221,7 @@ func TestDecodeProto3(t *testing.T) {
 }
 
 func TestEncodeDecodeForAllTypes(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/alltypes.proto", "", "AllTypesTest")
 	if err != nil {
 		t.Fatal(err)
@@ -337,10 +275,10 @@ func TestEncodeDecodeForAllTypes(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			a, err := c.Encode(tt.m)
+			a, err := c.Encode(ctx, tt.m)
 			assert.NoError(t, err)
 			assert.Equal(t, tt.b, a)
-			m, err := c.Decode(a)
+			m, err := c.Decode(ctx, a)
 			assert.NoError(t, err)
 			assert.Equal(t, tt.r, m)
 		})
@@ -348,14 +286,15 @@ func TestEncodeDecodeForAllTypes(t *testing.T) {
 }
 
 func TestErr(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test1.proto", "", "Person")
 	require.NoError(t, err)
-	_, err = c.Encode("123")
+	_, err = c.Encode(ctx, "123")
 	require.Error(t, err)
 	errWithCode, ok := err.(errorx.ErrorWithCode)
 	require.True(t, ok)
 	require.Equal(t, errorx.CovnerterErr, errWithCode.Code())
-	_, err = c.Decode(nil)
+	_, err = c.Decode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok = err.(errorx.ErrorWithCode)
 	require.True(t, ok)
diff --git a/internal/converter/protobuf/test/Makefile b/internal/converter/protobuf/test/Makefile
deleted file mode 100644
index 22f23892..00000000
--- a/internal/converter/protobuf/test/Makefile
+++ /dev/null
@@ -1,5 +0,0 @@
-.PHONY: build
-build:
-	@echo "Start build"
-	CGO_ENABLED=1 go build -trimpath --buildmode=plugin
-	@echo "Build successfully"
\ No newline at end of file
diff --git a/internal/converter/protobuf/test/go.mod b/internal/converter/protobuf/test/go.mod
deleted file mode 100644
index 86185d65..00000000
--- a/internal/converter/protobuf/test/go.mod
+++ /dev/null
@@ -1,8 +0,0 @@
-module helloworld
-
-go 1.22
-
-require (
-	github.com/golang/protobuf v1.5.2
-	google.golang.org/protobuf v1.33.0
-)
diff --git a/internal/converter/protobuf/test/go.sum b/internal/converter/protobuf/test/go.sum
deleted file mode 100644
index 0052f576..00000000
--- a/internal/converter/protobuf/test/go.sum
+++ /dev/null
@@ -1,11 +0,0 @@
-github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
-github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw=
-github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
-github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
-github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
-google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
-google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
diff --git a/internal/converter/protobuf/test/helloworld.pb.go b/internal/converter/protobuf/test/helloworld.pb.go
deleted file mode 100644
index 789002cd..00000000
--- a/internal/converter/protobuf/test/helloworld.pb.go
+++ /dev/null
@@ -1,217 +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.
-
-// Code generated by protoc-gen-go. DO NOT EDIT.
-// versions:
-// 	protoc-gen-go v1.28.1
-// 	protoc        v3.6.1
-// source: helloworld.proto
-
-package main
-
-import (
-	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
-	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
-	reflect "reflect"
-	sync "sync"
-)
-
-const (
-	// Verify that this generated code is sufficiently up-to-date.
-	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
-	// Verify that runtime/protoimpl is sufficiently up-to-date.
-	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
-)
-
-type HelloRequest struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
-}
-
-func (x *HelloRequest) Reset() {
-	*x = HelloRequest{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_helloworld_proto_msgTypes[0]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *HelloRequest) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*HelloRequest) ProtoMessage() {}
-
-func (x *HelloRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_helloworld_proto_msgTypes[0]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use HelloRequest.ProtoReflect.Descriptor instead.
-func (*HelloRequest) Descriptor() ([]byte, []int) {
-	return file_helloworld_proto_rawDescGZIP(), []int{0}
-}
-
-func (x *HelloRequest) GetName() string {
-	if x != nil {
-		return x.Name
-	}
-	return ""
-}
-
-type HelloReply struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Message string `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"`
-}
-
-func (x *HelloReply) Reset() {
-	*x = HelloReply{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_helloworld_proto_msgTypes[1]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *HelloReply) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*HelloReply) ProtoMessage() {}
-
-func (x *HelloReply) ProtoReflect() protoreflect.Message {
-	mi := &file_helloworld_proto_msgTypes[1]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use HelloReply.ProtoReflect.Descriptor instead.
-func (*HelloReply) Descriptor() ([]byte, []int) {
-	return file_helloworld_proto_rawDescGZIP(), []int{1}
-}
-
-func (x *HelloReply) GetMessage() string {
-	if x != nil {
-		return x.Message
-	}
-	return ""
-}
-
-var File_helloworld_proto protoreflect.FileDescriptor
-
-var file_helloworld_proto_rawDesc = []byte{
-	0x0a, 0x10, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x77, 0x6f, 0x72, 0x6c, 0x64, 0x2e, 0x70, 0x72, 0x6f,
-	0x74, 0x6f, 0x22, 0x22, 0x0a, 0x0c, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x52, 0x65, 0x71, 0x75, 0x65,
-	0x73, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
-	0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x22, 0x26, 0x0a, 0x0a, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x52,
-	0x65, 0x70, 0x6c, 0x79, 0x12, 0x18, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18,
-	0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x62, 0x06,
-	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
-}
-
-var (
-	file_helloworld_proto_rawDescOnce sync.Once
-	file_helloworld_proto_rawDescData = file_helloworld_proto_rawDesc
-)
-
-func file_helloworld_proto_rawDescGZIP() []byte {
-	file_helloworld_proto_rawDescOnce.Do(func() {
-		file_helloworld_proto_rawDescData = protoimpl.X.CompressGZIP(file_helloworld_proto_rawDescData)
-	})
-	return file_helloworld_proto_rawDescData
-}
-
-var file_helloworld_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
-var file_helloworld_proto_goTypes = []interface{}{
-	(*HelloRequest)(nil), // 0: HelloRequest
-	(*HelloReply)(nil),   // 1: HelloReply
-}
-var file_helloworld_proto_depIdxs = []int32{
-	0, // [0:0] is the sub-list for method output_type
-	0, // [0:0] is the sub-list for method input_type
-	0, // [0:0] is the sub-list for extension type_name
-	0, // [0:0] is the sub-list for extension extendee
-	0, // [0:0] is the sub-list for field type_name
-}
-
-func init() { file_helloworld_proto_init() }
-func file_helloworld_proto_init() {
-	if File_helloworld_proto != nil {
-		return
-	}
-	if !protoimpl.UnsafeEnabled {
-		file_helloworld_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*HelloRequest); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_helloworld_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*HelloReply); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-	}
-	type x struct{}
-	out := protoimpl.TypeBuilder{
-		File: protoimpl.DescBuilder{
-			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
-			RawDescriptor: file_helloworld_proto_rawDesc,
-			NumEnums:      0,
-			NumMessages:   2,
-			NumExtensions: 0,
-			NumServices:   0,
-		},
-		GoTypes:           file_helloworld_proto_goTypes,
-		DependencyIndexes: file_helloworld_proto_depIdxs,
-		MessageInfos:      file_helloworld_proto_msgTypes,
-	}.Build()
-	File_helloworld_proto = out.File
-	file_helloworld_proto_rawDesc = nil
-	file_helloworld_proto_goTypes = nil
-	file_helloworld_proto_depIdxs = nil
-}
diff --git a/internal/converter/protobuf/test/helloworld.proto b/internal/converter/protobuf/test/helloworld.proto
deleted file mode 100644
index c0d57e1e..00000000
--- a/internal/converter/protobuf/test/helloworld.proto
+++ /dev/null
@@ -1,9 +0,0 @@
-syntax = "proto3";
-
-message HelloRequest {
-  string name = 1;
-}
-
-message HelloReply {
-  string message = 1;
-}
\ No newline at end of file
diff --git a/internal/converter/protobuf/test/helloworld_wrapper.go b/internal/converter/protobuf/test/helloworld_wrapper.go
deleted file mode 100644
index e68c37b3..00000000
--- a/internal/converter/protobuf/test/helloworld_wrapper.go
+++ /dev/null
@@ -1,89 +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 main
-
-import (
-	"fmt"
-
-	"github.com/golang/protobuf/proto"
-)
-
-func (x *HelloReply) Encode(d interface{}) ([]byte, error) {
-	switch r := d.(type) {
-	case map[string]interface{}:
-		t, ok := r["message"]
-		if ok {
-			if v, ok := t.(string); ok {
-				x.Message = v
-			} else {
-				fmt.Println("message is not string")
-			}
-		} else {
-			// if required, return error
-			fmt.Println("message is not found")
-		}
-		return proto.Marshal(x)
-	default:
-		return nil, fmt.Errorf("unsupported type %v, must be a map", d)
-	}
-}
-
-func (x *HelloReply) Decode(b []byte) (interface{}, error) {
-	err := proto.Unmarshal(b, x)
-	if err != nil {
-		return nil, err
-	}
-	result := make(map[string]interface{}, 1)
-	result["message"] = x.Message
-	return result, nil
-}
-
-func GetHelloReply() interface{} {
-	return &HelloReply{}
-}
-
-func (x *HelloRequest) Encode(d interface{}) ([]byte, error) {
-	switch r := d.(type) {
-	case map[string]interface{}:
-		t, ok := r["name"]
-		if ok {
-			if v, ok := t.(string); ok {
-				x.Name = v
-			} else {
-				return nil, fmt.Errorf("name is not string")
-			}
-		} else {
-			// if required, return error
-			fmt.Println("message is not found")
-		}
-		return proto.Marshal(x)
-	default:
-		return nil, fmt.Errorf("unsupported type %v, must be a map", d)
-	}
-}
-
-func (x *HelloRequest) Decode(b []byte) (interface{}, error) {
-	err := proto.Unmarshal(b, x)
-	if err != nil {
-		return nil, err
-	}
-	result := make(map[string]interface{}, 1)
-	result["name"] = x.Name
-	return result, nil
-}
-
-func GetHelloRequest() interface{} {
-	return &HelloRequest{}
-}
diff --git a/internal/converter/protobuf/test/wrapper_test.go b/internal/converter/protobuf/test/wrapper_test.go
deleted file mode 100644
index 40869821..00000000
--- a/internal/converter/protobuf/test/wrapper_test.go
+++ /dev/null
@@ -1,35 +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 main
-
-import (
-	"fmt"
-	"testing"
-)
-
-func TestWrapper(t *testing.T) {
-	r := HelloReply{}
-	m := map[string]interface{}{"message": "hello"}
-	bytes, err := r.Encode(m)
-	if err != nil {
-		t.Errorf("encode error: %v", err)
-	}
-	fmt.Printf("bytes: %X\n", bytes)
-	mf, err := r.Decode(bytes)
-	if err != nil {
-		t.Errorf("decode error: %v", err)
-	}
-	fmt.Printf("mf: %v\n", mf)
-}
diff --git a/internal/processor/stream_test.go b/internal/processor/stream_test.go
index b58fd263..726a7939 100644
--- a/internal/processor/stream_test.go
+++ b/internal/processor/stream_test.go
@@ -284,33 +284,13 @@ func TestInferredStream(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	etcDir := filepath.Join(dataDir, "schemas", "custom")
-	err = os.MkdirAll(etcDir, os.ModePerm)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = os.RemoveAll(etcDir)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}()
-	// build the so file into data/test prior to running the test
-	bytesRead, err := os.ReadFile(filepath.Join(dataDir, "myFormat.so"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = os.WriteFile(filepath.Join(etcDir, "myFormat.so"), bytesRead, 0o755)
-	if err != nil {
-		t.Fatal(err)
-	}
 	petcDir := filepath.Join(dataDir, "schemas", "protobuf")
 	err = os.MkdirAll(petcDir, os.ModePerm)
 	if err != nil {
 		t.Fatal(err)
 	}
 	// Copy test2.proto
-	bytesRead, err = os.ReadFile("../schema/test/test2.proto")
+	bytesRead, err := os.ReadFile("../schema/test/test2.proto")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -343,20 +323,6 @@ func TestInferredStream(t *testing.T) {
 				"name":   {Type: "string"},
 				"author": {Type: "string"},
 			},
-		}, {
-			s: `CREATE STREAM demo2 () WITH (FORMAT="custom", DATASOURCE="demo", SCHEMAID="myFormat.Sample")`,
-			r: map[string]*ast.JsonStreamField{
-				"id":   {Type: "bigint"},
-				"name": {Type: "string"},
-				"age":  {Type: "bigint"},
-				"hobbies": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"indoor":  {Type: "array", Items: &ast.JsonStreamField{Type: "string"}},
-						"outdoor": {Type: "array", Items: &ast.JsonStreamField{Type: "string"}},
-					},
-				},
-			},
 		},
 	}
 
diff --git a/internal/schema/ext_inferer_custom_test.go b/internal/schema/ext_inferer_custom_test.go
deleted file mode 100644
index 8722fa8f..00000000
--- a/internal/schema/ext_inferer_custom_test.go
+++ /dev/null
@@ -1,80 +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 schema
-
-import (
-	"os"
-	"path/filepath"
-	"testing"
-
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/testx"
-	"github.com/lf-edge/ekuiper/v2/pkg/ast"
-)
-
-func init() {
-	testx.InitEnv("schema")
-}
-
-func TestInferCustom(t *testing.T) {
-	// Prepare test schema file
-	dataDir, err := conf.GetDataLoc()
-	if err != nil {
-		t.Fatal(err)
-	}
-	etcDir := filepath.Join(dataDir, "schemas", "custom")
-	err = os.MkdirAll(etcDir, os.ModePerm)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = os.RemoveAll(etcDir)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}()
-	// build the so file into data/test prior to running the test
-	// Copy the helloworld.so
-	bytesRead, err := os.ReadFile(filepath.Join(dataDir, "myFormat.so"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = os.WriteFile(filepath.Join(etcDir, "myFormat.so"), bytesRead, 0o755)
-	if err != nil {
-		t.Fatal(err)
-	}
-	InitRegistry()
-
-	// Test
-	result, err := InferCustom("myFormat", "Sample")
-	if err != nil {
-		t.Errorf("Infer custom format error: %v", err)
-		return
-	}
-	expected := ast.StreamFields{
-		{Name: "id", FieldType: &ast.BasicType{Type: ast.BIGINT}},
-		{Name: "name", FieldType: &ast.BasicType{Type: ast.STRINGS}},
-		{Name: "age", FieldType: &ast.BasicType{Type: ast.BIGINT}},
-		{Name: "hobbies", FieldType: &ast.RecType{
-			StreamFields: []ast.StreamField{
-				{Name: "indoor", FieldType: &ast.ArrayType{Type: ast.STRINGS}},
-				{Name: "outdoor", FieldType: &ast.ArrayType{Type: ast.STRINGS}},
-			},
-		}},
-	}
-	if len(result) != len(expected) {
-		t.Errorf("InferProtobuf result is not expected, got %v, expected %v", result, expected)
-	}
-}
diff --git a/internal/schema/registry_test.go b/internal/schema/registry_test.go
index f8adec20..4983939c 100644
--- a/internal/schema/registry_test.go
+++ b/internal/schema/registry_test.go
@@ -23,8 +23,13 @@ import (
 	"testing"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
 )
 
+func init() {
+	testx.InitEnv("schema")
+}
+
 func TestProtoRegistry(t *testing.T) {
 	// Move test schema file to etc dir
 	etcDir, err := conf.GetDataLoc()
diff --git a/internal/topo/context/default_test.go b/internal/topo/context/default_test.go
index 9485569e..b9d37667 100644
--- a/internal/topo/context/default_test.go
+++ b/internal/topo/context/default_test.go
@@ -26,7 +26,6 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/state"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/transform"
 )
 
 func TestState(t *testing.T) {
@@ -229,34 +228,3 @@ func TestParseTemplate(t *testing.T) {
 		}
 	}
 }
-
-func TestTransition(t *testing.T) {
-	var mockFunc transform.TransFunc = func(d interface{}) ([]byte, bool, error) {
-		return []byte(fmt.Sprintf("%v", d)), true, nil
-	}
-	tests := []struct {
-		data interface{}
-		r    []byte
-	}{
-		{
-			data: "hello",
-			r:    []byte(`hello`),
-		}, {
-			data: "world",
-			r:    []byte(`world`),
-		}, {
-			data: map[string]interface{}{"a": "hello"},
-			r:    []byte(`map[a:hello]`),
-		},
-	}
-
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-	ctx := Background().WithMeta("testTransRule", "op1", &state.MemoryStore{}).(*DefaultContext)
-	nc := WithValue(ctx, TransKey, mockFunc)
-	for i, tt := range tests {
-		r, _, _ := nc.TransformOutput(tt.data)
-		if !reflect.DeepEqual(tt.r, r) {
-			t.Errorf("%d\n\nstmt mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, string(tt.r), string(r))
-		}
-	}
-}
diff --git a/internal/topo/operator/preprocessor_test.go b/internal/topo/operator/preprocessor_test.go
index 73ef603a..b82747ce 100644
--- a/internal/topo/operator/preprocessor_test.go
+++ b/internal/topo/operator/preprocessor_test.go
@@ -1070,12 +1070,12 @@ func TestPreprocessorForBinary(t *testing.T) {
 		pp.streamFields = tt.stmt.StreamFields.ToJsonSchema()
 		format := message.FormatJson
 		ccc, _ := converter.GetOrCreateConverter(&ast.Options{FORMAT: format})
-		nCtx := context.WithValue(ctx, context.DecodeKey, ccc)
-		if dm, e := nCtx.Decode(tt.data); e != nil {
+		if dm, e := ccc.Decode(ctx, tt.data); e != nil {
 			log.Fatal(e)
 			return
 		} else {
-			tuple := &xsql.Tuple{Message: dm}
+			dmm := dm.(map[string]any)
+			tuple := &xsql.Tuple{Message: xsql.Message(dmm)}
 			fv, afv := xsql.NewFunctionValuersForOp(nil)
 			result := pp.Apply(ctx, tuple, fv, afv)
 			if !reflect.DeepEqual(tt.result, result) {