Repository: lf-edge/ekuiper
Pull request #3047 (2024-07-24 01:49:04)

feat(cli): restore rpc

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `26.81452%` with `363 lines` in your changes missing coverage. Please review.
> Project coverage is 67.78%. Comparing base [(`88fd5da`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/88fd5da26daaa8b607b01cb013f2d77a39818e40?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) to head [(`40c7b73`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/40c7b7398853428a9b8ebcdd52a591e1c2da4d56?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).
> Report is 60 commits behind head on master.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/server/rpc.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&filepath=internal%2Fserver%2Frpc.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3JwYy5nbw==) | 51.37% | [81 Missing and 43 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/rpc\_service.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&filepath=internal%2Fserver%2Frpc_service.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3JwY19zZXJ2aWNlLmdv) | 0.00% | [65 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/rpc\_plugin.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&filepath=internal%2Fserver%2Frpc_plugin.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3JwY19wbHVnaW4uZ28=) | 0.00% | [60 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/rpc\_schema.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&filepath=internal%2Fserver%2Frpc_schema.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3JwY19zY2hlbWEuZ28=) | 0.00% | [45 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/rpc\_plugin\_both.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&filepath=internal%2Fserver%2Frpc_plugin_both.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3JwY19wbHVnaW5fYm90aC5nbw==) | 0.00% | [37 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/rpc\_plugin\_hasnative.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&filepath=internal%2Fserver%2Frpc_plugin_hasnative.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3JwY19wbHVnaW5faGFzbmF0aXZlLmdv) | 0.00% | [31 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/sink/log\_sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&filepath=internal%2Fio%2Fsink%2Flog_sink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vc2luay9sb2dfc2luay5nbw==) | 66.67% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #3047      +/-   ##
==========================================
- Coverage   67.97%   67.78%   -0.19%     
==========================================
  Files         353      380      +27     
  Lines       34242    36283    +2041     
==========================================
+ Hits        23273    24592    +1319     
- Misses       8898     9448     +550     
- Partials     2071     2243     +172     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3047?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/extensions/impl/kafka/sink.go b/extensions/impl/kafka/sink.go
new file mode 100644
index 00000000..f22b722f
--- /dev/null
+++ b/extensions/impl/kafka/sink.go
@@ -0,0 +1,262 @@
+// 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 kafka
+
+import (
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/pingcap/failpoint"
+	kafkago "github.com/segmentio/kafka-go"
+	"github.com/segmentio/kafka-go/sasl"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/cert"
+)
+
+type KafkaSink struct {
+	writer         *kafkago.Writer
+	kc             *kafkaConf
+	tlsConfig      *tls.Config
+	headersMap     map[string]string
+	headerTemplate string
+	saslConf       *saslConf
+	mechanism      sasl.Mechanism
+}
+
+type kafkaConf struct {
+	Brokers      string      `json:"brokers"`
+	Topic        string      `json:"topic"`
+	MaxAttempts  int         `json:"maxAttempts"`
+	RequiredACKs int         `json:"requiredACKs"`
+	Key          string      `json:"key"`
+	Headers      interface{} `json:"headers"`
+}
+
+func (c *kafkaConf) validate() error {
+	if c.Topic == "" {
+		return fmt.Errorf("topic can not be empty")
+	}
+	if len(c.Brokers) < 1 {
+		return fmt.Errorf("brokers can not be empty")
+	}
+	return nil
+}
+
+func (k *KafkaSink) Provision(ctx api.StreamContext, configs map[string]any) error {
+	c := &kafkaConf{
+		RequiredACKs: -1,
+		MaxAttempts:  1,
+	}
+	err := cast.MapToStruct(configs, c)
+	failpoint.Inject("kafkaErr", func(val failpoint.Value) {
+		err = mockKakfaSourceErr(val.(int), castConfErr)
+	})
+	if err != nil {
+		return err
+	}
+	err = c.validate()
+	if err != nil {
+		return err
+	}
+	sc, err := getSaslConf(configs)
+	failpoint.Inject("kafkaErr", func(val failpoint.Value) {
+		err = mockKakfaSourceErr(val.(int), saslConfErr)
+	})
+	if err != nil {
+		return err
+	}
+	if err := sc.Validate(); err != nil {
+		return err
+	}
+	k.saslConf = sc
+	tlsConfig, err := cert.GenTLSConfig(configs, "kafka-sink")
+	if err != nil {
+		return err
+	}
+	mechanism, err := k.saslConf.GetMechanism()
+	failpoint.Inject("kafkaErr", func(val failpoint.Value) {
+		err = mockKakfaSourceErr(val.(int), mechanismErr)
+	})
+	if err != nil {
+		return err
+	}
+	k.mechanism = mechanism
+	k.tlsConfig = tlsConfig
+	k.kc = c
+	err = k.setHeaders()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (k *KafkaSink) buildKafkaWriter() error {
+	brokers := strings.Split(k.kc.Brokers, ",")
+	w := &kafkago.Writer{
+		Addr:  kafkago.TCP(brokers...),
+		Topic: k.kc.Topic,
+		// kafka java-client default balancer
+		Balancer:               &kafkago.Murmur2Balancer{},
+		Async:                  false,
+		AllowAutoTopicCreation: true,
+		MaxAttempts:            k.kc.MaxAttempts,
+		RequiredAcks:           kafkago.RequiredAcks(k.kc.RequiredACKs),
+		BatchSize:              1,
+		Transport: &kafkago.Transport{
+			SASL: k.mechanism,
+			TLS:  k.tlsConfig,
+		},
+	}
+	k.writer = w
+	return nil
+}
+
+func (k *KafkaSink) Close(ctx api.StreamContext) error {
+	return k.writer.Close()
+}
+
+func (k *KafkaSink) Connect(ctx api.StreamContext) error {
+	return k.buildKafkaWriter()
+}
+
+func (k *KafkaSink) Collect(ctx api.StreamContext, item api.MessageTuple) error {
+	msgs, err := k.collect(ctx, item)
+	if err != nil {
+		return err
+	}
+	return k.writer.WriteMessages(ctx, msgs...)
+}
+
+func (k *KafkaSink) CollectList(ctx api.StreamContext, items api.MessageTupleList) error {
+	allMsgs := make([]kafkago.Message, 0)
+	items.RangeOfTuples(func(index int, tuple api.MessageTuple) bool {
+		msgs, err := k.collect(ctx, tuple)
+		if err != nil {
+			return false
+		}
+		allMsgs = append(allMsgs, msgs...)
+		return true
+	})
+	return k.writer.WriteMessages(ctx, allMsgs...)
+}
+
+func (k *KafkaSink) collect(ctx api.StreamContext, item api.MessageTuple) ([]kafkago.Message, error) {
+	ds, err := json.Marshal(item.ToMap())
+	if err != nil {
+		return nil, err
+	}
+	var messages []kafkago.Message
+	msg, err := k.buildMsg(ctx, item, ds)
+	if err != nil {
+		return nil, err
+	}
+	messages = append(messages, msg)
+	return messages, nil
+}
+
+func (k *KafkaSink) buildMsg(ctx api.StreamContext, item api.MessageTuple, decodedBytes []byte) (kafkago.Message, error) {
+	msg := kafkago.Message{Value: decodedBytes}
+	if len(k.kc.Key) > 0 {
+		newKey := k.kc.Key
+		if dp, ok := item.(api.HasDynamicProps); ok {
+			key, ok := dp.DynamicProps("key")
+			if ok {
+				newKey = key
+			}
+		}
+		msg.Key = []byte(newKey)
+	}
+	headers, err := k.parseHeaders(ctx, item)
+	if err != nil {
+		return kafkago.Message{}, fmt.Errorf("parse kafka headers error: %v", err)
+	}
+	msg.Headers = headers
+	return msg, nil
+}
+
+func (k *KafkaSink) setHeaders() error {
+	if k.kc.Headers == nil {
+		return nil
+	}
+	switch h := k.kc.Headers.(type) {
+	case map[string]interface{}:
+		kafkaHeaders := make(map[string]string)
+		for key, value := range h {
+			if sv, ok := value.(string); ok {
+				kafkaHeaders[key] = sv
+			}
+		}
+		k.headersMap = kafkaHeaders
+		return nil
+	case string:
+		k.headerTemplate = h
+		return nil
+	default:
+		return fmt.Errorf("kafka headers must be a map[string]string or a string")
+	}
+}
+
+func (k *KafkaSink) parseHeaders(ctx api.StreamContext, item api.MessageTuple) ([]kafkago.Header, error) {
+	if len(k.headersMap) > 0 {
+		var kafkaHeaders []kafkago.Header
+		for k, v := range k.headersMap {
+			value := v
+			dp, ok := item.(api.HasDynamicProps)
+			if ok {
+				nv, ok := dp.DynamicProps(k)
+				if ok {
+					value = nv
+				}
+			}
+			kafkaHeaders = append(kafkaHeaders, kafkago.Header{
+				Key:   k,
+				Value: []byte(value),
+			})
+		}
+		return kafkaHeaders, nil
+	} else if len(k.headerTemplate) > 0 {
+		raw := k.headerTemplate
+		dp, ok := item.(api.HasDynamicProps)
+		if ok {
+			nv, ok := dp.DynamicProps("headers")
+			if ok {
+				raw = nv
+			}
+		}
+		headers := make(map[string]string)
+		if err := json.Unmarshal([]byte(raw), &headers); err != nil {
+			return nil, err
+		}
+		var kafkaHeaders []kafkago.Header
+		for key, value := range headers {
+			kafkaHeaders = append(kafkaHeaders, kafkago.Header{
+				Key:   key,
+				Value: []byte(value),
+			})
+		}
+		return kafkaHeaders, nil
+	}
+	return nil, nil
+}
+
+func GetSink() api.Sink {
+	return &KafkaSink{}
+}
+
+var _ api.TupleCollector = &KafkaSink{}
diff --git a/extensions/impl/kafka/source.go b/extensions/impl/kafka/source.go
new file mode 100644
index 00000000..4819eca3
--- /dev/null
+++ b/extensions/impl/kafka/source.go
@@ -0,0 +1,271 @@
+// 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 kafka
+
+import (
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/pingcap/failpoint"
+	kafkago "github.com/segmentio/kafka-go"
+	"github.com/segmentio/kafka-go/sasl"
+	"github.com/segmentio/kafka-go/sasl/plain"
+	"github.com/segmentio/kafka-go/sasl/scram"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/cert"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+type KafkaSource struct {
+	reader    *kafkago.Reader
+	offset    int64
+	tlsConfig *tls.Config
+	sc        *kafkaSourceConf
+	saslConf  *saslConf
+	mechanism sasl.Mechanism
+}
+
+type kafkaSourceConf struct {
+	Topic       string `json:"datasource"`
+	Brokers     string `json:"brokers"`
+	GroupID     string `json:"groupID"`
+	Partition   int    `json:"partition"`
+	MaxAttempts int    `json:"maxAttempts"`
+	MaxBytes    int    `json:"maxBytes"`
+}
+
+func (c *kafkaSourceConf) validate() error {
+	if c.Topic == "" {
+		return fmt.Errorf("kafkaSourceConf topic is required")
+	}
+	if len(c.Brokers) < 1 {
+		return fmt.Errorf("brokers can not be empty")
+	}
+	return nil
+}
+
+func (c *kafkaSourceConf) GetReaderConfig() kafkago.ReaderConfig {
+	return kafkago.ReaderConfig{
+		Brokers:     strings.Split(c.Brokers, ","),
+		GroupID:     c.GroupID,
+		Topic:       c.Topic,
+		Partition:   c.Partition,
+		MaxBytes:    c.MaxBytes,
+		MaxAttempts: c.MaxAttempts,
+	}
+}
+
+func getSourceConf(props map[string]interface{}) (*kafkaSourceConf, error) {
+	c := &kafkaSourceConf{
+		MaxBytes:    1e6,
+		MaxAttempts: 3,
+	}
+	err := cast.MapToStruct(props, c)
+	if err != nil {
+		return nil, fmt.Errorf("read properties %v fail with error: %v", props, err)
+	}
+	return c, nil
+}
+
+func (k *KafkaSource) Provision(ctx api.StreamContext, configs map[string]any) error {
+	kConf, err := getSourceConf(configs)
+	failpoint.Inject("kafkaErr", func(val failpoint.Value) {
+		err = mockKakfaSourceErr(val.(int), castConfErr)
+	})
+	if err != nil {
+		conf.Log.Errorf("kafka source config error: %v", err)
+		return err
+	}
+	if err := kConf.validate(); err != nil {
+		return err
+	}
+	k.sc = kConf
+	tlsConfig, err := cert.GenTLSConfig(configs, "kafka-source")
+	if err != nil {
+		conf.Log.Errorf("kafka tls conf error: %v", err)
+		return err
+	}
+	k.tlsConfig = tlsConfig
+	saslConf, err := getSaslConf(configs)
+	failpoint.Inject("kafkaErr", func(val failpoint.Value) {
+		err = mockKakfaSourceErr(val.(int), saslConfErr)
+	})
+	if err != nil {
+		conf.Log.Errorf("kafka sasl error: %v", err)
+		return err
+	}
+	if err := saslConf.Validate(); err != nil {
+		conf.Log.Errorf("kafka validate sasl error: %v", err)
+		return err
+	}
+	k.saslConf = saslConf
+	mechanism, err := k.saslConf.GetMechanism()
+	failpoint.Inject("kafkaErr", func(val failpoint.Value) {
+		err = mockKakfaSourceErr(val.(int), mechanismErr)
+	})
+	if err != nil {
+		conf.Log.Errorf("kafka sasl mechanism error: %v", err)
+		return err
+	}
+	k.mechanism = mechanism
+	conf.Log.Infof("kafka source got configured.")
+	return nil
+}
+
+func (k *KafkaSource) Close(ctx api.StreamContext) error {
+	return k.reader.Close()
+}
+
+func (k *KafkaSource) Connect(ctx api.StreamContext) error {
+	readerConfig := k.sc.GetReaderConfig()
+	conf.Log.Infof("topic: %s, brokers: %v", readerConfig.Topic, readerConfig.Brokers)
+	readerConfig.Dialer = &kafkago.Dialer{
+		Timeout:       10 * time.Second,
+		DualStack:     true,
+		TLS:           k.tlsConfig,
+		SASLMechanism: k.mechanism,
+	}
+	reader := kafkago.NewReader(readerConfig)
+	k.reader = reader
+	err := k.reader.SetOffset(kafkago.LastOffset)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (k *KafkaSource) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+		}
+		msg, err := k.reader.ReadMessage(ctx)
+		if err != nil {
+			ingestError(ctx, err)
+			continue
+		}
+		ingest(ctx, msg.Value, nil, timex.GetNow())
+	}
+}
+
+func (k *KafkaSource) Rewind(offset interface{}) error {
+	conf.Log.Infof("set kafka source offset: %v", offset)
+	offsetV := k.offset //nolint:staticcheck
+	switch v := offset.(type) {
+	case int64:
+		offsetV = v
+	case int:
+		offsetV = int64(v)
+	case float64:
+		offsetV = int64(v)
+	default:
+		return fmt.Errorf("%v can't be set as offset", offset)
+	}
+	if err := k.reader.SetOffset(offsetV); err != nil {
+		conf.Log.Errorf("kafka offset error: %v", err)
+		return fmt.Errorf("set kafka offset failed, err:%v", err)
+	}
+	return nil
+}
+
+func (k *KafkaSource) ResetOffset(input map[string]interface{}) error {
+	return errors.New("kafka source not support reset offset")
+}
+
+func (k *KafkaSource) GetOffset() (interface{}, error) {
+	return k.offset, nil
+}
+
+const (
+	SASL_NONE  = "none"
+	SASL_PLAIN = "plain"
+	SASL_SCRAM = "scram"
+)
+
+type saslConf struct {
+	SaslAuthType string `json:"saslAuthType"`
+	SaslUserName string `json:"saslUserName"`
+	SaslPassword string `json:"saslPassword"`
+}
+
+func getSaslConf(props map[string]interface{}) (*saslConf, error) {
+	sc := &saslConf{
+		SaslAuthType: SASL_NONE,
+	}
+	err := cast.MapToStruct(props, &sc)
+	return sc, err
+}
+
+func (c *saslConf) Validate() error {
+	if !(c.SaslAuthType == SASL_NONE || c.SaslAuthType == SASL_SCRAM || c.SaslAuthType == SASL_PLAIN) {
+		return fmt.Errorf("saslAuthType incorrect")
+	}
+	if (c.SaslAuthType == SASL_SCRAM || c.SaslAuthType == SASL_PLAIN) && (c.SaslUserName == "" || c.SaslPassword == "") {
+		return fmt.Errorf("username and password can not be empty")
+	}
+	return nil
+}
+
+func (c *saslConf) GetMechanism() (sasl.Mechanism, error) {
+	var err error
+	var mechanism sasl.Mechanism
+
+	// sasl authentication type
+	switch c.SaslAuthType {
+	case SASL_PLAIN:
+		mechanism = plain.Mechanism{
+			Username: c.SaslUserName,
+			Password: c.SaslPassword,
+		}
+	case SASL_SCRAM:
+		mechanism, err = scram.Mechanism(scram.SHA512, c.SaslUserName, c.SaslPassword)
+		if err != nil {
+			return mechanism, err
+		}
+	default:
+		mechanism = nil
+	}
+	return mechanism, nil
+}
+
+const (
+	mockErrStart int = iota
+	castConfErr
+	saslConfErr
+	mechanismErr
+	mockErrEnd
+)
+
+func mockKakfaSourceErr(v, exp int) error {
+	err := errors.New("mockErr")
+	if v == exp {
+		return err
+	}
+	return nil
+}
+
+func GetSource() api.Source {
+	return &KafkaSource{}
+}
+
+var _ api.BytesSource = &KafkaSource{}
diff --git a/extensions/impl/sql/lookupSource.go b/extensions/impl/sql/lookupSource.go
new file mode 100644
index 00000000..ee4fdc28
--- /dev/null
+++ b/extensions/impl/sql/lookupSource.go
@@ -0,0 +1,221 @@
+// 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 sql
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/pingcap/failpoint"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	client2 "github.com/lf-edge/ekuiper/v2/extensions/impl/sql/client"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+)
+
+type SqlLookupSource struct {
+	conf          *SQLConf
+	conn          *client2.SQLConnection
+	props         map[string]any
+	driver        string
+	table         string
+	needReconnect bool
+	gen           sqlQueryGen
+}
+
+func (s *SqlLookupSource) Ping(_ string, m map[string]interface{}) error {
+	ctx := context.Background()
+	if err := s.Provision(ctx, m); err != nil {
+		return err
+	}
+	if err := s.Connect(ctx); err != nil {
+		return err
+	}
+	defer func() {
+		s.Close(ctx)
+	}()
+	return s.conn.Ping(ctx)
+}
+
+func (s *SqlLookupSource) Provision(ctx api.StreamContext, configs map[string]any) error {
+	cfg := &SQLConf{}
+	err := cast.MapToStruct(configs, cfg)
+	failpoint.Inject("MapToStructErr", func() {
+		err = errors.New("MapToStruct")
+	})
+	if err != nil {
+		return fmt.Errorf("read properties %v fail with error: %v", configs, err)
+	}
+	if len(cfg.DBUrl) < 1 {
+		return fmt.Errorf("dburl should be defined")
+	}
+	s.conf = cfg
+	s.driver, err = client2.ParseDriver(s.conf.DBUrl)
+	if err != nil {
+		return err
+	}
+	s.table = cfg.Datasource
+	s.props = configs
+	s.gen = s.buildGen()
+	return nil
+}
+
+func (s *SqlLookupSource) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Closing sql source connector url:%v", s.conf.DBUrl)
+	id := s.conf.DBUrl
+	connection.DetachConnection(ctx, id, s.props)
+	if s.conn != nil {
+		s.conn.DetachSub(ctx, s.props)
+	}
+	return nil
+}
+
+func (s *SqlLookupSource) Connect(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Connecting to sql server")
+	var cli *client2.SQLConnection
+	var err error
+	id := s.conf.DBUrl
+	cw, err := connection.FetchConnection(ctx, id, "sql", s.props)
+	if err != nil {
+		return err
+	}
+	conn, err := cw.Wait()
+	if err != nil {
+		return err
+	}
+	cli = conn.(*client2.SQLConnection)
+	s.conn = cli
+	return err
+}
+
+func (s *SqlLookupSource) Lookup(ctx api.StreamContext, fields []string, keys []string, values []any) ([]map[string]any, error) {
+	if s.needReconnect {
+		err := s.conn.Reconnect()
+		if err != nil {
+			conf.Log.Errorf("reconnect db error %v", err)
+			return nil, err
+		}
+	}
+	query := s.gen.buildQuery(fields, keys, values)
+	ctx.GetLogger().Debugf("Query is %s", query)
+	rows, err := s.conn.GetDB().Query(query)
+	if err != nil {
+		s.needReconnect = true
+		ctx.GetLogger().Errorf("sql look table failed, err:%v, query: %v", err, query)
+		return nil, err
+	} else {
+		s.needReconnect = false
+	}
+	cols, _ := rows.Columns()
+	types, err := rows.ColumnTypes()
+	if err != nil {
+		return nil, err
+	}
+	dataList := make([]map[string]any, 0)
+	for rows.Next() {
+		data := make(map[string]any)
+		columns := make([]interface{}, len(cols))
+		prepareValues(columns, types, cols)
+
+		err := rows.Scan(columns...)
+		if err != nil {
+			return nil, err
+		}
+		scanIntoMap(data, columns, cols)
+		dataList = append(dataList, data)
+	}
+	return dataList, nil
+}
+
+type sqlQueryGen interface {
+	buildQuery(fields []string, keys []string, values []interface{}) string
+}
+
+type defaultSQLGen struct{ table string }
+
+func (g defaultSQLGen) buildQuery(fields []string, keys []string, values []interface{}) string {
+	query := "SELECT "
+	if len(fields) == 0 {
+		query += "*"
+	} else {
+		for i, f := range fields {
+			if i > 0 {
+				query += ","
+			}
+			query += f
+		}
+	}
+	query += fmt.Sprintf(" FROM %s WHERE ", g.table)
+	for i, k := range keys {
+		if i > 0 {
+			query += " AND "
+		}
+		switch v := values[i].(type) {
+		case string:
+			query += fmt.Sprintf("`%s` = '%s'", k, v)
+		default:
+			query += fmt.Sprintf("`%s` = %v", k, v)
+		}
+	}
+	return query
+}
+
+type noQuoteSQLGen struct{ table string }
+
+func (g noQuoteSQLGen) buildQuery(fields []string, keys []string, values []interface{}) string {
+	query := "SELECT "
+	if len(fields) == 0 {
+		query += "*"
+	} else {
+		for i, f := range fields {
+			if i > 0 {
+				query += ","
+			}
+			query += f
+		}
+	}
+	query += fmt.Sprintf(" FROM %s WHERE ", g.table)
+	for i, k := range keys {
+		if i > 0 {
+			query += " AND "
+		}
+		switch v := values[i].(type) {
+		case string:
+			query += fmt.Sprintf("%s = '%s'", k, v)
+		default:
+			query += fmt.Sprintf("%s = %v", k, v)
+		}
+	}
+	return query
+}
+
+func (s *SqlLookupSource) buildGen() sqlQueryGen {
+	switch strings.ToLower(s.driver) {
+	case "sqlserver", "mssql", "postgres":
+		return noQuoteSQLGen{table: s.table}
+	default:
+		return defaultSQLGen{table: s.table}
+	}
+}
+
+func GetLookupSource() api.Source {
+	return &SqlLookupSource{}
+}
+
+var _ api.LookupSource = &SqlLookupSource{}
diff --git a/extensions/impl/sql/sink.go b/extensions/impl/sql/sink.go
index ec37c585..a3ac2c0e 100644
--- a/extensions/impl/sql/sink.go
+++ b/extensions/impl/sql/sink.go
@@ -28,10 +28,11 @@ import (
 )
 
 type SQLSinkConnector struct {
-	config *sqlSinkConfig
-	cw     *connection.ConnWrapper
-	conn   *client.SQLConnection
-	props  map[string]any
+	config        *sqlSinkConfig
+	cw            *connection.ConnWrapper
+	conn          *client.SQLConnection
+	props         map[string]any
+	needReconnect bool
 }
 
 type sqlSinkConfig struct {
@@ -86,8 +87,13 @@ func (c *sqlSinkConfig) getKeyValues(ctx api.StreamContext, mapData map[string]i
 }
 
 func (s *SQLSinkConnector) Provision(ctx api.StreamContext, configs map[string]any) error {
-	c := &sqlSinkConfig{}
-	err := cast.MapToStruct(configs, c)
+	sc := &SQLConf{}
+	err := cast.MapToStruct(configs, sc)
+	if err != nil {
+		return err
+	}
+	c := &sqlSinkConfig{SQLConf: sc}
+	err = cast.MapToStruct(configs, c)
 	if err != nil {
 		return err
 	}
@@ -246,10 +252,18 @@ func (s *SQLSinkConnector) save(ctx api.StreamContext, table string, data map[st
 
 func (s *SQLSinkConnector) writeToDB(ctx api.StreamContext, sqlStr string) error {
 	ctx.GetLogger().Debugf(sqlStr)
+	if s.needReconnect {
+		err := s.conn.Reconnect()
+		if err != nil {
+			return errorx.NewIOErr(err.Error())
+		}
+	}
 	r, err := s.conn.GetDB().Exec(sqlStr)
 	if err != nil {
+		s.needReconnect = true
 		return errorx.NewIOErr(err.Error())
 	}
+	s.needReconnect = false
 	d, err := r.RowsAffected()
 	if err != nil {
 		ctx.GetLogger().Errorf("get rows affected error: %s", err.Error())
diff --git a/extensions/impl/sql/source.go b/extensions/impl/sql/source.go
index 876b73a9..dba99f74 100644
--- a/extensions/impl/sql/source.go
+++ b/extensions/impl/sql/source.go
@@ -41,9 +41,10 @@ type SQLSourceConnector struct {
 }
 
 type SQLConf struct {
-	Interval cast.DurationConf `json:"interval"`
-	DBUrl    string            `json:"dburl"`
-	URL      string            `json:"url,omitempty"`
+	Interval   cast.DurationConf `json:"interval"`
+	DBUrl      string            `json:"dburl"`
+	URL        string            `json:"url,omitempty"`
+	Datasource string            `json:"datasource"`
 }
 
 func (s *SQLSourceConnector) Provision(ctx api.StreamContext, props map[string]any) error {
diff --git a/extensions/sinks/kafka/kafka.go b/extensions/sinks/kafka/kafka.go
new file mode 100644
index 00000000..a7a017d4
--- /dev/null
+++ b/extensions/sinks/kafka/kafka.go
@@ -0,0 +1,22 @@
+// 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 kafka
+
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/kafka"
+)
+
+func Kafka() api.Sink { return kafka.GetSink() }
diff --git a/extensions/sources/kafka/kafka.go b/extensions/sources/kafka/kafka.go
new file mode 100644
index 00000000..d9a0d880
--- /dev/null
+++ b/extensions/sources/kafka/kafka.go
@@ -0,0 +1,22 @@
+// 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 kafka
+
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql"
+)
+
+func Kafka() api.Source { return sql.GetSource() }
diff --git a/extensions/sources/kafka/kafka.json b/extensions/sources/kafka/kafka.json
new file mode 100644
index 00000000..ad7a4783
--- /dev/null
+++ b/extensions/sources/kafka/kafka.json
@@ -0,0 +1,250 @@
+{
+  "about": {
+    "trial": true,
+    "author": {
+      "name": "carlclone",
+      "email": "carlclone@gmail.com",
+      "company": "personal",
+      "website": "https://www.github.com/carlclone"
+    },
+    "helpUrl": {
+      "en_US": "https://ekuiper.org/docs/en/latest/guide/sinks/plugin/kafka.html",
+      "zh_CN": "https://ekuiper.org/docs/zh/latest/guide/sinks/plugin/kafka.html"
+    },
+    "description": {
+      "en_US": "This a sink for Kafka, it can be used for saving the analysis data into Kafka.",
+      "zh_CN": "为 Kafka 的持久化插件，可以用于将分析数据存入 Kafka 中"
+    }
+  },
+  "libs": [
+    "github.com/segmentio/kafka-go@v0.4.39"
+  ],
+  "properties": [
+    {
+      "name": "brokers",
+      "default": "127.0.0.1:9092,127.0.0.2:9092",
+      "optional": false,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "The url of the Kafka broker list",
+        "zh_CN": "Kafka brokers 的 URL 列表"
+      },
+      "label": {
+        "en_US": "broker list",
+        "zh_CN": "Broker URL 列表"
+      }
+    },
+    {
+      "name": "topic",
+      "default": "",
+      "optional": false,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "The topic to publish to.",
+        "zh_CN": "订阅主题"
+      },
+      "label": {
+        "en_US": "Topic",
+        "zh_CN": "主题"
+      }
+    },
+    {
+      "name": "requiredACKs",
+      "default": "",
+      "optional": false,
+      "control": "text",
+      "type": "int",
+      "hint": {
+        "en_US": "The mechanism for Kafka client to confirm messages",
+        "zh_CN": "Kafka 客户端确认消息的机制"
+      },
+      "label": {
+        "en_US": "requiredACKs",
+        "zh_CN": "确认机制"
+      }
+    },
+    {
+      "name": "saslAuthType",
+      "default": "none",
+      "optional": false,
+      "control": "select",
+      "values": [
+        "none",
+        "plain",
+        "scram"
+      ],
+      "type": "string",
+      "hint": {
+        "en_US": "Sasl auth type of Kafka",
+        "zh_CN": "Kafka 的 Sasl 认证类型"
+      },
+      "label": {
+        "en_US": "Sasl auth type",
+        "zh_CN": "Sasl 认证类型"
+      }
+    },
+    {
+      "name": "saslUserName",
+      "default": "",
+      "optional": true,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "Sasl username for authentication",
+        "zh_CN": "Sasl 认证使用的用户名"
+      },
+      "label": {
+        "en_US": "Sasl username",
+        "zh_CN": "Sasl 用户名"
+      }
+    },
+    {
+      "name": "saslPassword",
+      "default": "",
+      "optional": true,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "Sasl password for authentication",
+        "zh_CN": "Sasl 认证使用的密码"
+      },
+      "label": {
+        "en_US": "Sasl password",
+        "zh_CN": "Sasl 密码"
+      }
+    },
+    {
+      "name": "certificationPath",
+      "default": "",
+      "optional": true,
+      "connection_related": true,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "The location of certification path. It can be an absolute path, or a relative path.",
+        "zh_CN": "证书路径。可以为绝对路径，也可以为相对路径。如果指定的是相对路径，那么父目录为执行 server 命令的路径。"
+      },
+      "label": {
+        "en_US": "Certification path",
+        "zh_CN": "证书路径"
+      }
+    },
+    {
+      "name": "privateKeyPath",
+      "default": "",
+      "optional": true,
+      "connection_related": true,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "The location of private key path. It can be an absolute path, or a relative path. ",
+        "zh_CN": "私钥路径。可以为绝对路径，也可以为相对路径。"
+      },
+      "label": {
+        "en_US": "Private key path",
+        "zh_CN": "私钥路径"
+      }
+    },
+    {
+      "name": "rootCaPath",
+      "default": "",
+      "optional": true,
+      "connection_related": true,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "The location of root ca path. It can be an absolute path, or a relative path. ",
+        "zh_CN": "根证书路径，用以验证服务器证书。可以为绝对路径，也可以为相对路径。"
+      },
+      "label": {
+        "en_US": "Root CA path",
+        "zh_CN": "根证书路径"
+      }
+    },
+    {
+      "name": "insecureSkipVerify",
+      "default": true,
+      "optional": true,
+      "control": "radio",
+      "type": "bool",
+      "hint": {
+        "en_US": "Control if to skip the certification verification. If it is set to true, then skip certification verification; Otherwise, verify the certification.",
+        "zh_CN": "控制是否跳过证书认证。如果被设置为 true，那么跳过证书认证；否则进行证书验证。"
+      },
+      "label": {
+        "en_US": "Skip Certification verification",
+        "zh_CN": "跳过证书验证"
+      }
+    },
+    {
+      "name": "batchSize",
+      "default": 1,
+      "optional": true,
+      "control": "text",
+      "type": "int",
+      "hint": {
+        "en_US": "Limit on how many messages will be buffered before being sent to kafka in one request, default 1",
+        "zh_CN": "一次性向 kafka 发送批量消息的数量"
+      },
+      "label": {
+        "en_US": "Batch Size",
+        "zh_CN": "批量消息发送"
+      }
+    },
+    {
+      "name": "maxAttempts",
+      "default": 1,
+      "optional": true,
+      "control": "text",
+      "type": "int",
+      "hint": {
+        "en_US": "Limit on how many attempts will be made to deliver a message, default 1",
+        "zh_CN": "发送消息失败的重试次数"
+      },
+      "label": {
+        "en_US": "Max Attempts",
+        "zh_CN": "消息发送重发次数"
+      }
+    },
+    {
+      "name": "headers",
+      "default": {},
+      "optional": true,
+      "control": "list",
+      "type": "object",
+      "hint": {
+        "en_US": "Message Headers",
+        "zh_CN": "Headers in Kafka Message"
+      },
+      "label": {
+        "en_US": "Headers for the message",
+        "zh_CN": "Kafka 消息 Headers"
+      }
+    },
+    {
+      "name": "key",
+      "default": "",
+      "optional": true,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "Message Key",
+        "zh_CN": "Kafka Message Key"
+      },
+      "label": {
+        "en_US": "key for the message",
+        "zh_CN": "Kafka 消息 Key"
+      }
+    }
+  ],
+  "node": {
+    "category": "sink",
+    "icon": "iconPath",
+    "label": {
+      "en": "Kafka",
+      "zh": "Kafka"
+    }
+  }
+}
\ No newline at end of file
diff --git a/extensions/sources/kafka/kafka.yaml b/extensions/sources/kafka/kafka.yaml
new file mode 100644
index 00000000..debf7457
--- /dev/null
+++ b/extensions/sources/kafka/kafka.yaml
@@ -0,0 +1,5 @@
+default:
+  brokers: "127.0.0.1:9091,127.0.0.1:9092"
+  groupID: ""
+  partition: 0
+  maxBytes: 1000000
\ No newline at end of file
diff --git a/go.mod b/go.mod
index acb935e7..8db0eef3 100644
--- a/go.mod
+++ b/go.mod
@@ -43,6 +43,7 @@ require (
 	github.com/prometheus/client_golang v1.19.0
 	github.com/redis/go-redis/v9 v9.5.1
 	github.com/robfig/cron/v3 v3.0.1
+	github.com/segmentio/kafka-go v0.4.47
 	github.com/shirou/gopsutil v3.21.11+incompatible
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.9.0
@@ -119,6 +120,7 @@ require (
 	github.com/oapi-codegen/runtime v1.0.0 // indirect
 	github.com/onsi/ginkgo v1.16.4 // indirect
 	github.com/onsi/gomega v1.13.0 // indirect
+	github.com/pierrec/lz4/v4 v4.1.15 // indirect
 	github.com/pingcap/errors v0.11.4 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
@@ -136,6 +138,9 @@ require (
 	github.com/tklauser/numcpus v0.7.0 // indirect
 	github.com/u2takey/go-utils v0.3.1 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
+	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
+	github.com/xdg-go/scram v1.1.2 // indirect
+	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/yuin/gopher-lua v1.1.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/otel v1.7.0 // indirect
diff --git a/go.sum b/go.sum
index 2178bf9d..01e1d4d0 100644
--- a/go.sum
+++ b/go.sum
@@ -321,6 +321,7 @@ github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1/go.mod h1:
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/klauspost/compress v1.17.7 h1:ehO88t2UGzQK66LMdE8tibEd1ErmzZjNEqWkjLAKQQg=
 github.com/klauspost/compress v1.17.7/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
@@ -447,6 +448,8 @@ github.com/pebbe/zmq4 v1.2.11/go.mod h1:nqnPueOapVhE2wItZ0uOErngczsJdLOGkebMxaO8
 github.com/performancecopilot/speed v3.0.0+incompatible/go.mod h1:/CLtqpZ5gBg1M9iaPbIdPPGyKcA8hKdoy6hAWba7Yac=
 github.com/pierrec/lz4 v1.0.2-0.20190131084431-473cd7ce01a1/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0=
+github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4=
 github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8=
 github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c h1:CgbKAHto5CQgWM9fSBIvaxsJHuGP0uM74HXtv3MyyGQ=
@@ -505,6 +508,8 @@ github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
+github.com/segmentio/kafka-go v0.4.47 h1:IqziR4pA3vrZq7YdRxaT3w1/5fvIH5qpCwstUanQQB0=
+github.com/segmentio/kafka-go v0.4.47/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
 github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI=
 github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
@@ -569,6 +574,12 @@ github.com/valyala/fastjson v1.6.4 h1:uAUNq9Z6ymTgGhcm0UynUAB6tlbakBrz6CQFax3BXV
 github.com/valyala/fastjson v1.6.4/go.mod h1:CLCAqky6SMuOcxStkYQvblddUtoRxhYMGLrsQns1aXY=
 github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
+github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
+github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/xo/dburl v0.23.2 h1:Fl88cvayrgE56JA/sqhNMLljCW/b7RmG1mMkKMZUFgA=
 github.com/xo/dburl v0.23.2/go.mod h1:uazlaAQxj4gkshhfuuYyvwCBouOmNnG2aDxTCFZpmL4=
@@ -614,6 +625,7 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.3.0/go.mod h1:hebNnKkNXi2UzZN1eVRvBB7co0a+JxK6XbPiWVs/3J4=
+golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
 golang.org/x/crypto v0.21.0 h1:X31++rzVUdKhX5sWmSOFZxx8UW/ldWx55cbf08iNAMA=
 golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -629,6 +641,7 @@ golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKG
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.16.0 h1:QX4fJ0Rr5cPQCF7O9lh9Se4pmwfwskqZfq5moyldzic=
 golang.org/x/mod v0.16.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -655,6 +668,9 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.2.0/go.mod h1:KqCZLdyyvdV855qA2rE3GC2aiw5xGR5TEjj8smXukLY=
+golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/net v0.23.0 h1:7EYJ93RZ9vYSZAIb2x3lnuvqO5zneoD6IvWjuhfxjTs=
 golang.org/x/net v0.23.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -667,6 +683,7 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ=
 golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -706,12 +723,18 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
 golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.2.0/go.mod h1:TVmDHMZPmdnySmBfhjOoOdhjzdE1h4u1VwSiw2l1Nuc=
+golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
+golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
+golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -719,6 +742,9 @@ golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -742,6 +768,7 @@ golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20200103221440-774c71fcf114/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
 golang.org/x/tools v0.19.0 h1:tfGCXNR1OsFG+sVdLAitlpjAvD/I6dHDKnYrpEZUHkw=
 golang.org/x/tools v0.19.0/go.mod h1:qoJWxmGSIBmAeriMx19ogtrEPrGtDbPK634QFIcLAhc=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/internal/binder/io/ext_full.go b/internal/binder/io/ext_full.go
index 82de753d..35abb78f 100644
--- a/internal/binder/io/ext_full.go
+++ b/internal/binder/io/ext_full.go
@@ -21,6 +21,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/extensions/impl/image"
 	"github.com/lf-edge/ekuiper/v2/extensions/impl/influx"
 	"github.com/lf-edge/ekuiper/v2/extensions/impl/influx2"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/kafka"
 	sql2 "github.com/lf-edge/ekuiper/v2/extensions/impl/sql"
 	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/client"
 	"github.com/lf-edge/ekuiper/v2/extensions/impl/video"
@@ -30,13 +31,18 @@ import (
 func init() {
 	//modules.RegisterSource("random", func() api.Source { return random.GetSource() })
 	modules.RegisterSource("video", func() api.Source { return video.GetSource() })
-	//modules.RegisterSource("kafka", func() api.Source { return kafkaSrc.GetSource() })
+	modules.RegisterSource("kafka", func() api.Source { return kafka.GetSource() })
 	//modules.RegisterLookupSource("sql", func() api.LookupSource { return sql.GetLookup() })
+	//modules.RegisterSink("image", func() api.Sink { return image.GetSink() })
+	//modules.RegisterSink("influx", func() api.Sink { return influx.GetSink() })
+	//modules.RegisterSink("influx2", func() api.Sink { return influx2.GetSink() })
+	modules.RegisterSink("kafka", func() api.Sink { return kafka.GetSink() })
 	modules.RegisterSink("image", func() api.Sink { return image.GetSink() })
 	modules.RegisterSink("influx", func() api.Sink { return influx.GetSink() })
 	modules.RegisterSink("influx2", func() api.Sink { return influx2.GetSink() })
 	//modules.RegisterSink("kafka", func() api.Sink { return kafka.GetSink() })
 	modules.RegisterSource("sql", sql2.GetSource)
+	modules.RegisterLookupSource("sql", sql2.GetLookupSource)
 	modules.RegisterSink("sql", sql2.GetSink)
 	modules.RegisterConnection("sql", client.CreateConnection)
 }
diff --git a/internal/io/sink/log_sink.go b/internal/io/sink/log_sink.go
index d4d67be5..363f437b 100644
--- a/internal/io/sink/log_sink.go
+++ b/internal/io/sink/log_sink.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/collector"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 )
 
 // NewLogSink log action, no properties now
@@ -43,12 +44,12 @@ var QR = &QueryResult{LastFetch: time.Now()}
 func NewLogSinkToMemory() api.Sink {
 	QR.Results = make([]string, 0, 10)
 	return collector.Func(func(ctx api.StreamContext, data any) error {
-		result, ok := data.(string)
-		if !ok {
+		r, err := cast.ToString(data, cast.CONVERT_SAMEKIND)
+		if err != nil {
 			return fmt.Errorf("result is not a string but got %v", data)
 		}
 		QR.Mux.Lock()
-		QR.Results = append(QR.Results, result)
+		QR.Results = append(QR.Results, r)
 		QR.Mux.Unlock()
 		return nil
 	})
diff --git a/internal/server/meta_init.go b/internal/server/meta_init.go
index 99097a29..c4d8ed21 100644
--- a/internal/server/meta_init.go
+++ b/internal/server/meta_init.go
@@ -63,6 +63,7 @@ func (m metaComp) rest(r *mux.Router) {
 	r.HandleFunc("/metadata/resources", resourcesHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/sources/connection/{name}", sourceConnectionHandler).Methods(http.MethodPost)
 	r.HandleFunc("/metadata/sinks/connection/{name}", sinkConnectionHandler).Methods(http.MethodPost)
+	r.HandleFunc("/metadata/lookups/connection/{name}", lookupConnectionHandler).Methods(http.MethodPost)
 	for _, endpoint := range metaEndpoints {
 		endpoint(r)
 	}
@@ -368,3 +369,25 @@ func sourceConnectionHandler(w http.ResponseWriter, r *http.Request) {
 
 	w.WriteHeader(http.StatusOK)
 }
+
+func lookupConnectionHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+
+	sourceNm := vars["name"]
+	config := map[string]interface{}{}
+	v, _ := io.ReadAll(r.Body)
+	err := json.Unmarshal(v, &config)
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+	config = replacePasswdForConfig("source", sourceNm, config)
+	err = node.LookupPing(sourceNm, config)
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+}
diff --git a/internal/server/rpc.go b/internal/server/rpc.go
new file mode 100644
index 00000000..2d229868
--- /dev/null
+++ b/internal/server/rpc.go
@@ -0,0 +1,415 @@
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build rpc || !core
+
+package server
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/rpc"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/sink"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/infra"
+)
+
+const QueryRuleId = "internal-ekuiper_query_rule"
+
+func init() {
+	servers["rpc"] = &rpcComp{}
+}
+
+type rpcComp struct {
+	s *http.Server
+}
+
+func (r *rpcComp) register() {}
+
+func (r *rpcComp) serve() {
+	// Start rpc service
+	server := new(Server)
+	portRpc := conf.Config.Basic.Port
+	ipRpc := conf.Config.Basic.Ip
+	rpcSrv := rpc.NewServer()
+	err := rpcSrv.Register(server)
+	if err != nil {
+		logger.Fatal("Format of service Server isn'restHttpType correct. ", err)
+	}
+	srvRpc := &http.Server{
+		Addr:         cast.JoinHostPortInt(ipRpc, portRpc),
+		WriteTimeout: time.Second * 15,
+		ReadTimeout:  time.Second * 15,
+		IdleTimeout:  time.Second * 60,
+		Handler:      rpcSrv,
+	}
+	r.s = srvRpc
+	go func() {
+		if err = srvRpc.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			logger.Fatal("Error serving rpc service:", err)
+		}
+	}()
+	initQuery()
+}
+
+func (r *rpcComp) close() {
+	if r.s != nil {
+		if err := r.s.Shutdown(context.TODO()); err != nil {
+			logger.Errorf("rpc server shutdown error: %v", err)
+		}
+		logger.Info("rpc server shutdown.")
+	}
+}
+
+type Server int
+
+func (t *Server) CreateQuery(sql string, reply *string) error {
+	if _, ok := registry.Load(QueryRuleId); ok {
+		stopQuery()
+	}
+	tp, err := ruleProcessor.ExecQuery(QueryRuleId, sql)
+	if err != nil {
+		return err
+	} else {
+		rs := &rule.RuleState{RuleId: QueryRuleId, Topology: tp}
+		registry.Store(QueryRuleId, rs)
+		msg := fmt.Sprintf("Query was submit successfully.")
+		logger.Println(msg)
+		*reply = fmt.Sprint(msg)
+	}
+	return nil
+}
+
+func stopQuery() {
+	if rs, ok := registry.Load(QueryRuleId); ok {
+		logger.Printf("stop the query.")
+		(*rs.Topology).Cancel()
+		registry.Delete(QueryRuleId)
+	}
+}
+
+/**
+ * qid is not currently used.
+ */
+func (t *Server) GetQueryResult(_ string, reply *string) error {
+	if rs, ok := registry.Load(QueryRuleId); ok {
+		c := (*rs.Topology).GetContext()
+		if c != nil && c.Err() != nil {
+			return c.Err()
+		}
+	}
+
+	sink.QR.LastFetch = time.Now()
+	sink.QR.Mux.Lock()
+	if len(sink.QR.Results) > 0 {
+		*reply = strings.Join(sink.QR.Results, "\n")
+		sink.QR.Results = make([]string, 0, 10)
+	} else {
+		*reply = ""
+	}
+	sink.QR.Mux.Unlock()
+	return nil
+}
+
+func (t *Server) Stream(stream string, reply *string) error {
+	content, err := streamProcessor.ExecStmt(stream)
+	if err != nil {
+		return fmt.Errorf("Stream command error: %s", err)
+	} else {
+		for _, c := range content {
+			*reply = *reply + fmt.Sprintln(c)
+		}
+	}
+	return nil
+}
+
+func (t *Server) CreateRule(rule *model.RPCArgDesc, reply *string) error {
+	id, err := createRule(rule.Name, rule.Json)
+	if err != nil {
+		return fmt.Errorf("Create rule %s error : %s.", id, err)
+	} else {
+		*reply = fmt.Sprintf("Rule %s was created successfully, please use 'bin/kuiper getstatus rule %s' command to get rule status.", rule.Name, rule.Name)
+	}
+	return nil
+}
+
+func (t *Server) GetStatusRule(name string, reply *string) error {
+	if r, err := getRuleStatus(name); err != nil {
+		return err
+	} else {
+		*reply = r
+	}
+	return nil
+}
+
+func (t *Server) GetTopoRule(name string, reply *string) error {
+	if r, err := getRuleTopo(name); err != nil {
+		return err
+	} else {
+		dst := &bytes.Buffer{}
+		if err = json.Indent(dst, cast.StringToBytes(r), "", "  "); err != nil {
+			*reply = r
+		} else {
+			*reply = dst.String()
+		}
+	}
+	return nil
+}
+
+func (t *Server) StartRule(name string, reply *string) error {
+	if err := startRule(name); err != nil {
+		return err
+	} else {
+		*reply = fmt.Sprintf("Rule %s was started", name)
+	}
+	return nil
+}
+
+func (t *Server) StopRule(name string, reply *string) error {
+	*reply, _ = stopRule(name)
+	return nil
+}
+
+func (t *Server) RestartRule(name string, reply *string) error {
+	err := restartRule(name)
+	if err != nil {
+		return err
+	}
+	*reply = fmt.Sprintf("Rule %s was restarted.", name)
+	return nil
+}
+
+func (t *Server) DescRule(name string, reply *string) error {
+	r, err := ruleProcessor.ExecDesc(name)
+	if err != nil {
+		return fmt.Errorf("Desc rule error : %s.", err)
+	} else {
+		*reply = r
+	}
+	return nil
+}
+
+func (t *Server) ShowRules(_ int, reply *string) error {
+	r, err := getAllRulesWithStatus()
+	if err != nil {
+		return fmt.Errorf("Show rule error : %s.", err)
+	}
+	if len(r) == 0 {
+		*reply = "No rule definitions are found."
+	} else {
+		result, err := json.Marshal(r)
+		if err != nil {
+			return fmt.Errorf("Show rule error : %s.", err)
+		}
+		dst := &bytes.Buffer{}
+		if err := json.Indent(dst, result, "", "  "); err != nil {
+			return fmt.Errorf("Show rule error : %s.", err)
+		}
+		*reply = dst.String()
+	}
+	return nil
+}
+
+func (t *Server) DropRule(name string, reply *string) error {
+	deleteRule(name)
+	r, err := ruleProcessor.ExecDrop(name)
+	if err != nil {
+		return fmt.Errorf("Drop rule error : %s.", err)
+	} else {
+		err := t.StopRule(name, reply)
+		if err != nil {
+			return err
+		}
+	}
+	*reply = r
+	return nil
+}
+
+func (t *Server) ValidateRule(rule *model.RPCArgDesc, reply *string) error {
+	_, s, err := validateRule(rule.Name, rule.Json)
+	if s {
+		*reply = "The rule has been successfully validated and is confirmed to be correct."
+	} else {
+		*reply = err.Error()
+	}
+	return nil
+}
+
+func (t *Server) Import(file string, reply *string) error {
+	f, err := os.Open(file)
+	if err != nil {
+		return fmt.Errorf("fail to read file %s: %v", file, err)
+	}
+	defer f.Close()
+	buf := new(bytes.Buffer)
+	_, err = io.Copy(buf, f)
+	if err != nil {
+		return fmt.Errorf("fail to convert file %s: %v", file, err)
+	}
+	content := buf.Bytes()
+	rules, counts, err := rulesetProcessor.Import(content)
+	if err != nil {
+		return fmt.Errorf("import ruleset error: %v", err)
+	}
+	infra.SafeRun(func() error {
+		for _, name := range rules {
+			rul, ee := ruleProcessor.GetRuleById(name)
+			if ee != nil {
+				logger.Error(ee)
+				continue
+			}
+			reply := recoverRule(rul)
+			if reply != "" {
+				logger.Error(reply)
+			}
+		}
+		return nil
+	})
+	*reply = fmt.Sprintf("imported %d streams, %d tables and %d rules", counts[0], counts[1], counts[2])
+	return nil
+}
+
+func (t *Server) Export(file string, reply *string) error {
+	f, err := os.Create(file)
+	if err != nil {
+		return err
+	}
+	exported, counts, err := rulesetProcessor.Export()
+	if err != nil {
+		return err
+	}
+	_, err = io.Copy(f, exported)
+	if err != nil {
+		return fmt.Errorf("fail to save to file %s:%v", file, err)
+	}
+	*reply = fmt.Sprintf("exported %d streams, %d tables and %d rules", counts[0], counts[1], counts[2])
+	return nil
+}
+
+func (t *Server) ImportConfiguration(arg *model.ImportDataDesc, reply *string) error {
+	file := arg.FileName
+	f, err := os.Open(file)
+	if err != nil {
+		return fmt.Errorf("fail to read file %s: %v", file, err)
+	}
+	defer f.Close()
+	buf := new(bytes.Buffer)
+	_, err = io.Copy(buf, f)
+	if err != nil {
+		return fmt.Errorf("fail to convert file %s: %v", file, err)
+	}
+	content := buf.Bytes()
+	partial := arg.Partial
+
+	var result ImportConfigurationStatus
+	if !partial {
+		configurationReset()
+		result = configurationImport(context.Background(), content, arg.Stop)
+	} else {
+		result = configurationPartialImport(context.Background(), content)
+	}
+	marshal, _ := json.Marshal(result)
+
+	dst := &bytes.Buffer{}
+	if err := json.Indent(dst, marshal, "", "  "); err != nil {
+		return fmt.Errorf("import configuration error: %v", err)
+	}
+	*reply = dst.String()
+
+	return nil
+}
+
+func (t *Server) GetStatusImport(_ int, reply *string) error {
+	jsonRsp := configurationStatusExport()
+	result, err := json.Marshal(jsonRsp)
+	if err != nil {
+		return fmt.Errorf("Show rule error : %s.", err)
+	}
+	dst := &bytes.Buffer{}
+	if err := json.Indent(dst, result, "", "  "); err != nil {
+		return fmt.Errorf("Show rule error : %s.", err)
+	}
+	*reply = dst.String()
+
+	return nil
+}
+
+func (t *Server) ExportConfiguration(arg *model.ExportDataDesc, reply *string) error {
+	rules := arg.Rules
+	file := arg.FileName
+	f, err := os.Create(file)
+	if err != nil {
+		return err
+	}
+	var jsonBytes []byte
+	// do not specify rules, export all
+	if len(rules) == 0 {
+		jsonBytes, err = configurationExport()
+	} else {
+		jsonBytes, err = ruleMigrationProcessor.ConfigurationPartialExport(rules)
+	}
+	if err != nil {
+		return err
+	}
+	_, err = io.Copy(f, bytes.NewReader(jsonBytes))
+	if err != nil {
+		return fmt.Errorf("fail to save to file %s:%v", file, err)
+	}
+	*reply = fmt.Sprintf("export configuration success")
+	return nil
+}
+
+func marshalDesc(m interface{}) (string, error) {
+	s, err := json.Marshal(m)
+	if err != nil {
+		return "", fmt.Errorf("invalid json %v", m)
+	}
+	dst := &bytes.Buffer{}
+	if err := json.Indent(dst, s, "", "  "); err != nil {
+		return "", fmt.Errorf("indent json error %v", err)
+	}
+	return dst.String(), nil
+}
+
+func initQuery() {
+	ticker := time.NewTicker(time.Second * 5)
+	go infra.SafeRun(func() error {
+		for {
+			<-ticker.C
+			if registry != nil {
+				if _, ok := registry.Load(QueryRuleId); !ok {
+					continue
+				}
+
+				n := time.Now()
+				w := 10 * time.Second
+				if v := n.Sub(sink.QR.LastFetch); v >= w {
+					logger.Printf("The client seems no longer fetch the query result, stop the query now.")
+					stopQuery()
+				}
+			}
+		}
+	})
+}
diff --git a/internal/server/rpc_plugin.go b/internal/server/rpc_plugin.go
new file mode 100644
index 00000000..4c441822
--- /dev/null
+++ b/internal/server/rpc_plugin.go
@@ -0,0 +1,112 @@
+// Copyright 2022-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.
+
+//go:build (rpc || !core) && (plugin || portable || !core)
+
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/internal/plugin"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+func (t *Server) CreatePlugin(arg *model.PluginDesc, reply *string) error {
+	pt := plugin.PluginType(arg.Type)
+	p, err := getPluginByJson(arg, pt)
+	if err != nil {
+		return fmt.Errorf("Create plugin error: %s", err)
+	}
+	if p.GetFile() == "" {
+		return fmt.Errorf("Create plugin error: Missing plugin file url.")
+	}
+	// define according to the build tag
+	err = t.doRegister(pt, p)
+	if err != nil {
+		return err
+	}
+	if err != nil {
+		return fmt.Errorf("Create plugin error: %s", err)
+	} else {
+		*reply = fmt.Sprintf("Plugin %s is created.", p.GetName())
+	}
+	return nil
+}
+
+func (t *Server) DropPlugin(arg *model.PluginDesc, reply *string) error {
+	pt := plugin.PluginType(arg.Type)
+	p, err := getPluginByJson(arg, pt)
+	if err != nil {
+		return fmt.Errorf("Drop plugin error: %s", err)
+	}
+	err = t.doDelete(pt, p.GetName(), arg.Stop)
+	if err != nil {
+		return fmt.Errorf("Drop plugin error: %s", err)
+	} else {
+		if pt == plugin.PORTABLE {
+			*reply = fmt.Sprintf("Plugin %s is dropped .", p.GetName())
+		} else {
+			if arg.Stop {
+				*reply = fmt.Sprintf("Plugin %s is dropped and Kuiper will be stopped.", p.GetName())
+			} else {
+				*reply = fmt.Sprintf("Plugin %s is dropped and Kuiper must restart for the change to take effect.", p.GetName())
+			}
+		}
+	}
+
+	return nil
+}
+
+func (t *Server) DescPlugin(arg *model.PluginDesc, reply *string) error {
+	pt := plugin.PluginType(arg.Type)
+	p, err := getPluginByJson(arg, pt)
+	if err != nil {
+		return fmt.Errorf("Describe plugin error: %s", err)
+	}
+	m, err := t.doDesc(pt, p.GetName())
+	if err != nil {
+		return fmt.Errorf("Describe plugin error: %s", err)
+	} else {
+		r, err := marshalDesc(m)
+		if err != nil {
+			return fmt.Errorf("Describe plugin error: %v", err)
+		}
+		*reply = r
+	}
+	return nil
+}
+
+func (t *Server) ShowPlugins(arg int, reply *string) error {
+	pt := plugin.PluginType(arg)
+	l, err := t.doShow(pt)
+	if err != nil {
+		return fmt.Errorf("Show plugin error: %s", err)
+	}
+	*reply = l
+	return nil
+}
+
+func getPluginByJson(arg *model.PluginDesc, pt plugin.PluginType) (plugin.Plugin, error) {
+	p := plugin.NewPluginByType(pt)
+	if arg.Json != "" {
+		if err := json.Unmarshal(cast.StringToBytes(arg.Json), p); err != nil {
+			return nil, fmt.Errorf("Parse plugin %s error : %s.", arg.Json, err)
+		}
+	}
+	p.SetName(arg.Name)
+	return p, nil
+}
diff --git a/internal/server/rpc_plugin_both.go b/internal/server/rpc_plugin_both.go
new file mode 100644
index 00000000..bce6f7a7
--- /dev/null
+++ b/internal/server/rpc_plugin_both.go
@@ -0,0 +1,77 @@
+// Copyright 2022-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.
+
+//go:build (!core && !wasmedge) || (rpc && portable && plugin)
+
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/lf-edge/ekuiper/v2/internal/plugin"
+)
+
+func (t *Server) doRegister(pt plugin.PluginType, p plugin.Plugin) error {
+	if pt == plugin.PORTABLE {
+		return portableManager.Register(p)
+	} else {
+		return nativeManager.Register(pt, p)
+	}
+}
+
+func (t *Server) doDelete(pt plugin.PluginType, name string, stopRun bool) error {
+	if pt == plugin.PORTABLE {
+		return portableManager.Delete(name)
+	} else {
+		return nativeManager.Delete(pt, name, stopRun)
+	}
+}
+
+func (t *Server) doDesc(pt plugin.PluginType, name string) (interface{}, error) {
+	var (
+		result interface{}
+		ok     bool
+	)
+	if pt == plugin.PORTABLE {
+		result, ok = portableManager.GetPluginInfo(name)
+	} else {
+		result, ok = nativeManager.GetPluginInfo(pt, name)
+	}
+	if !ok {
+		return nil, fmt.Errorf("not found")
+	}
+	return result, nil
+}
+
+func (t *Server) doShow(pt plugin.PluginType) (string, error) {
+	var result string
+	if pt == plugin.PORTABLE {
+		l := portableManager.List()
+		jb, err := json.Marshal(l)
+		if err != nil {
+			return "", err
+		}
+		return string(jb), nil
+	} else {
+		l := nativeManager.List(pt)
+		if len(l) == 0 {
+			result = "No plugin is found."
+		} else {
+			result = strings.Join(l, "\n")
+		}
+		return result, nil
+	}
+}
diff --git a/internal/server/rpc_plugin_hasnative.go b/internal/server/rpc_plugin_hasnative.go
new file mode 100644
index 00000000..3a7559e2
--- /dev/null
+++ b/internal/server/rpc_plugin_hasnative.go
@@ -0,0 +1,69 @@
+// Copyright 2022-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.
+
+//go:build !core || (rpc && plugin)
+
+package server
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/internal/plugin"
+)
+
+func (t *Server) RegisterPlugin(arg *model.PluginDesc, reply *string) error {
+	p, err := getPluginByJson(arg, plugin.FUNCTION)
+	if err != nil {
+		return fmt.Errorf("Register plugin functions error: %s", err)
+	}
+	if len(p.GetSymbols()) == 0 {
+		return fmt.Errorf("Register plugin functions error: Missing function list.")
+	}
+	err = nativeManager.RegisterFuncs(p.GetName(), p.GetSymbols())
+	if err != nil {
+		return fmt.Errorf("Create plugin error: %s", err)
+	} else {
+		*reply = fmt.Sprintf("Plugin %s is created.", p.GetName())
+	}
+	return nil
+}
+
+func (t *Server) ShowUdfs(_ int, reply *string) error {
+	l := nativeManager.ListSymbols()
+	if len(l) == 0 {
+		l = append(l, "No udf is found.")
+	}
+	*reply = strings.Join(l, "\n")
+	return nil
+}
+
+func (t *Server) DescUdf(arg string, reply *string) error {
+	m, ok := nativeManager.GetPluginBySymbol(plugin.FUNCTION, arg)
+	if !ok {
+		return fmt.Errorf("Describe udf error: not found")
+	} else {
+		j := map[string]string{
+			"name":   arg,
+			"plugin": m,
+		}
+		r, err := marshalDesc(j)
+		if err != nil {
+			return fmt.Errorf("Describe udf error: %v", err)
+		}
+		*reply = r
+	}
+	return nil
+}
diff --git a/internal/server/rpc_plugin_native.go b/internal/server/rpc_plugin_native.go
new file mode 100644
index 00000000..b7cf43b9
--- /dev/null
+++ b/internal/server/rpc_plugin_native.go
@@ -0,0 +1,67 @@
+// Copyright 2022-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.
+
+//go:build plugin && rpc && core && !portable
+
+package server
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/lf-edge/ekuiper/v2/internal/plugin"
+)
+
+func (t *Server) doRegister(pt plugin.PluginType, p plugin.Plugin) error {
+	if pt == plugin.PORTABLE {
+		return fmt.Errorf("portable plugin support is disabled")
+	} else {
+		return nativeManager.Register(pt, p)
+	}
+}
+
+func (t *Server) doDelete(pt plugin.PluginType, name string, stopRun bool) error {
+	if pt == plugin.PORTABLE {
+		return fmt.Errorf("portable plugin support is disabled")
+	} else {
+		return nativeManager.Delete(pt, name, stopRun)
+	}
+}
+
+func (t *Server) doDesc(pt plugin.PluginType, name string) (interface{}, error) {
+	if pt == plugin.PORTABLE {
+		return nil, fmt.Errorf("portable plugin support is disabled")
+	} else {
+		r, ok := nativeManager.GetPluginInfo(pt, name)
+		if !ok {
+			return nil, fmt.Errorf("not found")
+		}
+		return r, nil
+	}
+}
+
+func (t *Server) doShow(pt plugin.PluginType) (string, error) {
+	var result string
+	if pt == plugin.PORTABLE {
+		return "", fmt.Errorf("portable plugin support is disabled")
+	} else {
+		l := nativeManager.List(pt)
+		if len(l) == 0 {
+			result = "No plugin is found."
+		} else {
+			result = strings.Join(l, "\n")
+		}
+		return result, nil
+	}
+}
diff --git a/internal/server/rpc_plugin_portable.go b/internal/server/rpc_plugin_portable.go
new file mode 100644
index 00000000..ba0894e8
--- /dev/null
+++ b/internal/server/rpc_plugin_portable.go
@@ -0,0 +1,65 @@
+// Copyright 2022-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.
+
+//go:build portable && rpc && core && !plugin
+
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/lf-edge/ekuiper/v2/internal/plugin"
+)
+
+func (t *Server) doRegister(pt plugin.PluginType, p plugin.Plugin) error {
+	if pt == plugin.PORTABLE {
+		return portableManager.Register(p)
+	} else {
+		return fmt.Errorf("native plugin support is disabled")
+	}
+}
+
+func (t *Server) doDelete(pt plugin.PluginType, name string, stopRun bool) error {
+	if pt == plugin.PORTABLE {
+		return portableManager.Delete(name)
+	} else {
+		return fmt.Errorf("native plugin support is disabled")
+	}
+}
+
+func (t *Server) doDesc(pt plugin.PluginType, name string) (interface{}, error) {
+	if pt == plugin.PORTABLE {
+		r, ok := portableManager.GetPluginInfo(name)
+		if !ok {
+			return nil, fmt.Errorf("not found")
+		}
+		return r, nil
+	} else {
+		return nil, fmt.Errorf("native plugin support is disabled")
+	}
+}
+
+func (t *Server) doShow(pt plugin.PluginType) (string, error) {
+	if pt == plugin.PORTABLE {
+		l := portableManager.List()
+		jb, err := json.Marshal(l)
+		if err != nil {
+			return "", err
+		}
+		return string(jb), nil
+	} else {
+		return "", fmt.Errorf("native plugin support is disabled")
+	}
+}
diff --git a/internal/server/rpc_schema.go b/internal/server/rpc_schema.go
new file mode 100644
index 00000000..ae7de759
--- /dev/null
+++ b/internal/server/rpc_schema.go
@@ -0,0 +1,91 @@
+// Copyright 2022-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.
+
+//go:build !core || (rpc && schema)
+
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/internal/schema"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+func (t *Server) CreateSchema(arg *model.RPCTypedArgDesc, reply *string) error {
+	sd := &schema.Info{Type: def.SchemaType(arg.Type)}
+	if arg.Json != "" {
+		if err := json.Unmarshal(cast.StringToBytes(arg.Json), sd); err != nil {
+			return fmt.Errorf("Parse service %s error : %s.", arg.Json, err)
+		}
+	}
+	if sd.Name != arg.Name {
+		return fmt.Errorf("Create schema error: name mismatch.")
+	}
+	if sd.Content != "" && sd.FilePath != "" {
+		return fmt.Errorf("Invalid body: Cannot specify both content and file")
+	}
+	err := schema.Register(sd)
+	if err != nil {
+		return fmt.Errorf("Create schema error: %s", err)
+	} else {
+		*reply = fmt.Sprintf("Schema %s is created.", arg.Name)
+	}
+	return nil
+}
+
+func (t *Server) DescSchema(arg *model.RPCTypedArgDesc, reply *string) error {
+	j, err := schema.GetSchema(def.SchemaType(arg.Type), arg.Name)
+	if err != nil {
+		return fmt.Errorf("Desc schema error : %s.", err)
+	} else if j == nil {
+		return fmt.Errorf("Desc schema error : not found.")
+	} else {
+		r, err := marshalDesc(j)
+		if err != nil {
+			return fmt.Errorf("Describe service error: %v", err)
+		}
+		*reply = r
+	}
+	return nil
+}
+
+func (t *Server) DropSchema(arg *model.RPCTypedArgDesc, reply *string) error {
+	err := schema.DeleteSchema(def.SchemaType(arg.Type), arg.Name)
+	if err != nil {
+		return fmt.Errorf("Drop schema error : %s.", err)
+	}
+	*reply = fmt.Sprintf("Schema %s is dropped", arg.Name)
+	return nil
+}
+
+func (t *Server) ShowSchemas(schemaType string, reply *string) error {
+	l, err := schema.GetAllForType(def.SchemaType(schemaType))
+	if err != nil {
+		return fmt.Errorf("Show schemas error: %s.", err)
+	}
+	if len(l) == 0 {
+		*reply = "No schema definitions are found."
+	} else {
+		r, err := marshalDesc(l)
+		if err != nil {
+			return fmt.Errorf("Show service error: %v", err)
+		}
+		*reply = r
+	}
+	return nil
+}
diff --git a/internal/server/rpc_script.go b/internal/server/rpc_script.go
new file mode 100644
index 00000000..e7837003
--- /dev/null
+++ b/internal/server/rpc_script.go
@@ -0,0 +1,79 @@
+// Copyright 2022-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.
+
+//go:build full || (rpc && script)
+
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/lf-edge/ekuiper/v2/internal/plugin/js"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+func (t *Server) CreateScript(j string, reply *string) error {
+	sd := &js.Script{}
+	if err := json.Unmarshal(cast.StringToBytes(j), sd); err != nil {
+		return fmt.Errorf("Parse JavaScript function error : %s.", err)
+	}
+	err := js.GetManager().Create(sd)
+	if err != nil {
+		return fmt.Errorf("Create JavaScript function error: %s", err)
+	} else {
+		*reply = fmt.Sprintf("JavaScript function %s is created.", sd.Id)
+	}
+	return nil
+}
+
+func (t *Server) DescScript(name string, reply *string) error {
+	j, err := js.GetManager().GetScript(name)
+	if err != nil {
+		return fmt.Errorf("Describe JavaScript function error : %s.", err)
+	} else {
+		r, err := marshalDesc(j)
+		if err != nil {
+			return fmt.Errorf("Describe JavaScript function error : %s.", err)
+		}
+		*reply = r
+	}
+	return nil
+}
+
+func (t *Server) DropScript(name string, reply *string) error {
+	err := js.GetManager().Delete(name)
+	if err != nil {
+		return fmt.Errorf("Drop JavaScript function error : %s.", err)
+	}
+	*reply = fmt.Sprintf("JavaScript function %s is dropped", name)
+	return nil
+}
+
+func (t *Server) ShowScripts(_ int, reply *string) error {
+	content, err := js.GetManager().List()
+	if err != nil {
+		return fmt.Errorf("Show JavaScript functions error: %s.", err)
+	}
+	if len(content) == 0 {
+		*reply = "No JavaScript functions are found."
+	} else {
+		r, err := marshalDesc(content)
+		if err != nil {
+			return fmt.Errorf("Show service error: %v", err)
+		}
+		*reply = r
+	}
+	return nil
+}
diff --git a/internal/server/rpc_service.go b/internal/server/rpc_service.go
new file mode 100644
index 00000000..0097fee7
--- /dev/null
+++ b/internal/server/rpc_service.go
@@ -0,0 +1,119 @@
+// Copyright 2022-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.
+
+//go:build !core || (rpc && service)
+
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/internal/service"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+func (t *Server) CreateService(arg *model.RPCArgDesc, reply *string) error {
+	sd := &service.ServiceCreationRequest{}
+	if arg.Json != "" {
+		if err := json.Unmarshal(cast.StringToBytes(arg.Json), sd); err != nil {
+			return fmt.Errorf("Parse service %s error : %s.", arg.Json, err)
+		}
+	}
+	if sd.Name != arg.Name {
+		return fmt.Errorf("Create service error: name mismatch.")
+	}
+	if sd.File == "" {
+		return fmt.Errorf("Create service error: Missing service file url.")
+	}
+	err := serviceManager.Create(sd)
+	if err != nil {
+		return fmt.Errorf("Create service error: %s", err)
+	} else {
+		*reply = fmt.Sprintf("Service %s is created.", arg.Name)
+	}
+	return nil
+}
+
+func (t *Server) DescService(name string, reply *string) error {
+	s, err := serviceManager.Get(name)
+	if err != nil {
+		return fmt.Errorf("Desc service error : %s.", err)
+	} else {
+		r, err := marshalDesc(s)
+		if err != nil {
+			return fmt.Errorf("Describe service error: %v", err)
+		}
+		*reply = r
+	}
+	return nil
+}
+
+func (t *Server) DescServiceFunc(name string, reply *string) error {
+	s, err := serviceManager.GetFunction(name)
+	if err != nil {
+		return fmt.Errorf("Desc service func error : %s.", err)
+	} else {
+		r, err := marshalDesc(s)
+		if err != nil {
+			return fmt.Errorf("Describe service func error: %v", err)
+		}
+		*reply = r
+	}
+	return nil
+}
+
+func (t *Server) DropService(name string, reply *string) error {
+	err := serviceManager.Delete(name)
+	if err != nil {
+		return fmt.Errorf("Drop service error : %s.", err)
+	}
+	*reply = fmt.Sprintf("Service %s is dropped", name)
+	return nil
+}
+
+func (t *Server) ShowServices(_ int, reply *string) error {
+	s, err := serviceManager.List()
+	if err != nil {
+		return fmt.Errorf("Show service error: %s.", err)
+	}
+	if len(s) == 0 {
+		*reply = "No service definitions are found."
+	} else {
+		r, err := marshalDesc(s)
+		if err != nil {
+			return fmt.Errorf("Show service error: %v", err)
+		}
+		*reply = r
+	}
+	return nil
+}
+
+func (t *Server) ShowServiceFuncs(_ int, reply *string) error {
+	s, err := serviceManager.ListFunctions()
+	if err != nil {
+		return fmt.Errorf("Show service funcs error: %s.", err)
+	}
+	if len(s) == 0 {
+		*reply = "No service definitions are found."
+	} else {
+		r, err := marshalDesc(s)
+		if err != nil {
+			return fmt.Errorf("Show service funcs error: %v", err)
+		}
+		*reply = r
+	}
+	return nil
+}
diff --git a/internal/topo/node/node.go b/internal/topo/node/node.go
index 8ebb7f5f..c0438aed 100644
--- a/internal/topo/node/node.go
+++ b/internal/topo/node/node.go
@@ -266,3 +266,18 @@ func SourcePing(sourceType string, config map[string]any) error {
 	}
 	return fmt.Errorf("source %v doesn't support ping connection", sourceType)
 }
+
+func LookupPing(lookupType string, config map[string]interface{}) error {
+	lookup, err := io.LookupSource(lookupType)
+	if err != nil {
+		return err
+	}
+	dataSource := "/$$TEST_CONNECTION$$"
+	if v, ok := config["DATASOURCE"]; ok {
+		dataSource = v.(string)
+	}
+	if pingAble, ok := lookup.(util.PingableConn); ok {
+		return pingAble.Ping(dataSource, config)
+	}
+	return fmt.Errorf("lookup %v doesn't support ping connection", lookup)
+}
diff --git a/pkg/connection/connection_pool.go b/pkg/connection/connection_pool.go
index 81446bae..ecddbae0 100644
--- a/pkg/connection/connection_pool.go
+++ b/pkg/connection/connection_pool.go
@@ -120,6 +120,9 @@ func PingConnection(ctx api.StreamContext, id string) error {
 }
 
 func FetchConnection(ctx api.StreamContext, id, typ string, props map[string]interface{}) (*ConnWrapper, error) {
+	failpoint.Inject("FetchConnectionErr", func() {
+		failpoint.Return(nil, fmt.Errorf("FetchConnectionErr"))
+	})
 	if id == "" {
 		return nil, fmt.Errorf("connection id should be defined")
 	}


Test patch:
diff --git a/extensions/impl/kafka/sink_test.go b/extensions/impl/kafka/sink_test.go
new file mode 100644
index 00000000..7de0206b
--- /dev/null
+++ b/extensions/impl/kafka/sink_test.go
@@ -0,0 +1,126 @@
+// 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 kafka
+
+import (
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"testing"
+
+	"github.com/pingcap/failpoint"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestKafkaSink(t *testing.T) {
+	ks := &KafkaSink{}
+	testcases := []struct {
+		configs map[string]any
+	}{
+		{
+			configs: map[string]any{},
+		},
+		{
+			configs: map[string]any{
+				"topic": "t",
+			},
+		},
+		{
+			configs: map[string]any{
+				"topic":            "t",
+				"brokers":          "localhost:9092",
+				"certificationRaw": "mockErr",
+			},
+		},
+		{
+			configs: map[string]any{
+				"datasource":   "t",
+				"brokers":      "localhost:9092",
+				"saslAuthType": "mockErr",
+			},
+		},
+		{
+			configs: map[string]any{
+				"datasource":   "t",
+				"brokers":      "localhost:9092",
+				"saslAuthType": "plain",
+			},
+		},
+		{
+			configs: map[string]any{
+				"topic":   "t",
+				"brokers": "localhost:9092",
+				"headers": 1,
+			},
+		},
+	}
+	ctx := mockContext.NewMockContext("1", "2")
+	for index, tc := range testcases {
+		require.Error(t, ks.Provision(ctx, tc.configs), index)
+	}
+	configs := map[string]any{
+		"topic":   "t",
+		"brokers": "localhost:9092",
+	}
+	require.NoError(t, ks.Provision(ctx, configs))
+	require.NoError(t, ks.Connect(ctx))
+	mockT := testx.MockTuple{
+		Map: map[string]any{"1": 1},
+	}
+	msgs, err := ks.collect(ctx, mockT)
+	require.Len(t, msgs, 1)
+	require.NoError(t, err)
+	require.NoError(t, ks.Close(ctx))
+
+	for i := mockErrStart + 1; i < mockErrEnd; i++ {
+		failpoint.Enable("github.com/lf-edge/ekuiper/v2/extensions/impl/kafka/kafkaErr", fmt.Sprintf("return(%v)", i))
+		require.Error(t, ks.Provision(ctx, configs), i)
+	}
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/extensions/impl/kafka/kafkaErr")
+}
+
+func TestKafkaSinkBuildMsg(t *testing.T) {
+	configs := map[string]any{
+		"topic":   "t",
+		"brokers": "localhost:9092",
+		"headers": map[string]any{
+			"a": "{{.a}}",
+		},
+		"key": "{{.a}}",
+	}
+	ks := &KafkaSink{}
+	ctx := mockContext.NewMockContext("1", "2")
+	require.NoError(t, ks.Provision(ctx, configs))
+	require.NoError(t, ks.Connect(ctx))
+	item := map[string]any{
+		"a": 1,
+	}
+	d, _ := json.Marshal(item)
+	mockT := testx.MockTuple{
+		Map:      item,
+		Template: map[string]string{"a": "1", "key": "1"},
+	}
+	msg, err := ks.buildMsg(ctx, mockT, d)
+	require.NoError(t, err)
+	require.Equal(t, "a", msg.Headers[0].Key)
+	b := make([]uint8, 0, 8)
+	b = strconv.AppendInt(b, int64(1), 10)
+	require.Equal(t, b, msg.Headers[0].Value)
+	require.Equal(t, []byte("1"), msg.Key)
+	require.NoError(t, ks.Close(ctx))
+}
diff --git a/extensions/impl/kafka/source_test.go b/extensions/impl/kafka/source_test.go
new file mode 100644
index 00000000..b05c0a2e
--- /dev/null
+++ b/extensions/impl/kafka/source_test.go
@@ -0,0 +1,79 @@
+// 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 kafka
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/pingcap/failpoint"
+	"github.com/stretchr/testify/require"
+
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestKafkaSource(t *testing.T) {
+	ks := &KafkaSource{}
+	testcases := []struct {
+		configs map[string]any
+	}{
+		{
+			configs: map[string]any{},
+		},
+		{
+			configs: map[string]any{
+				"datasource": "t",
+			},
+		},
+		{
+			configs: map[string]any{
+				"datasource":       "t",
+				"brokers":          "localhost:9092",
+				"certificationRaw": "mockErr",
+			},
+		},
+		{
+			configs: map[string]any{
+				"datasource":   "t",
+				"brokers":      "localhost:9092",
+				"saslAuthType": "mockErr",
+			},
+		},
+		{
+			configs: map[string]any{
+				"datasource":   "t",
+				"brokers":      "localhost:9092",
+				"saslAuthType": "plain",
+			},
+		},
+	}
+	ctx := mockContext.NewMockContext("1", "2")
+	for _, tc := range testcases {
+		require.Error(t, ks.Provision(ctx, tc.configs))
+	}
+	configs := map[string]any{
+		"datasource": "t",
+		"brokers":    "localhost:9092",
+	}
+	require.NoError(t, ks.Provision(ctx, configs))
+	require.NoError(t, ks.Connect(ctx))
+	require.NoError(t, ks.Close(ctx))
+
+	for i := mockErrStart + 1; i < mockErrEnd; i++ {
+		failpoint.Enable("github.com/lf-edge/ekuiper/v2/extensions/impl/kafka/kafkaErr", fmt.Sprintf("return(%v)", i))
+		require.Error(t, ks.Provision(ctx, configs), i)
+	}
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/extensions/impl/kafka/kafkaErr")
+}
diff --git a/extensions/impl/sql/lookupSource_test.go b/extensions/impl/sql/lookupSource_test.go
new file mode 100644
index 00000000..9fc772bf
--- /dev/null
+++ b/extensions/impl/sql/lookupSource_test.go
@@ -0,0 +1,131 @@
+// 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 sql
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/pingcap/failpoint"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/testx"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestSQLLookupSourceErr(t *testing.T) {
+	connection.InitConnectionManager4Test()
+	ctx := mockContext.NewMockContext("1", "2")
+	props := map[string]interface{}{
+		"dburl":      fmt.Sprintf("mysql://root:@%v:%v/test", address, port),
+		"datasource": "t",
+	}
+	ls := &SqlLookupSource{}
+	require.NoError(t, ls.Provision(ctx, props))
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/pkg/connection/FetchConnectionErr", "return(true)")
+	defer failpoint.Disable("github.com/lf-edge/ekuiper/v2/pkg/connection/FetchConnectionErr")
+	require.Error(t, ls.Connect(ctx))
+}
+
+func TestSQLLookupSource(t *testing.T) {
+	connection.InitConnectionManager4Test()
+	ctx := mockContext.NewMockContext("1", "2")
+	s, err := testx.SetupEmbeddedMysqlServer(address, port)
+	require.NoError(t, err)
+	defer func() {
+		s.Close()
+	}()
+	props := map[string]interface{}{
+		"dburl":      fmt.Sprintf("mysql://root:@%v:%v/test", address, port),
+		"datasource": "t",
+	}
+	ls := &SqlLookupSource{}
+	require.NoError(t, ls.Provision(ctx, props))
+	require.NoError(t, ls.Connect(ctx))
+	got, err := ls.Lookup(ctx, []string{"a", "b"}, []string{"a"}, []any{1})
+	require.NoError(t, err)
+	require.Equal(t, []map[string]any{{"a": int64(1), "b": int64(1)}}, got)
+	got, err = ls.Lookup(ctx, []string{"a", "b"}, []string{"a", "b"}, []any{1, 1})
+	require.NoError(t, err)
+	require.Equal(t, []map[string]any{{"a": int64(1), "b": int64(1)}}, got)
+	ls.Close(ctx)
+}
+
+func TestSQLLookupSourceProvisionErr(t *testing.T) {
+	props := map[string]interface{}{}
+	ls := &SqlLookupSource{}
+	ctx := mockContext.NewMockContext("1", "2")
+	require.Error(t, ls.Provision(ctx, props))
+	props = map[string]interface{}{
+		"dburl":      fmt.Sprintf("mysql://root:@%v:%v/test", address, port),
+		"datasource": "t",
+	}
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/extensions/impl/sql/MapToStructErr", "return(true)")
+	require.Error(t, ls.Provision(ctx, props))
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/extensions/impl/sql/MapToStructErr")
+	props = map[string]interface{}{
+		"dburl":      "123",
+		"datasource": "t",
+	}
+	require.Error(t, ls.Provision(ctx, props))
+}
+
+func TestSQLLookupReconnect(t *testing.T) {
+	connection.InitConnectionManager4Test()
+	ctx := mockContext.NewMockContext("1", "2")
+	s, err := testx.SetupEmbeddedMysqlServer(address, port)
+	require.NoError(t, err)
+	props := map[string]interface{}{
+		"dburl":      fmt.Sprintf("mysql://root:@%v:%v/test", address, port),
+		"datasource": "t",
+	}
+	ls := &SqlLookupSource{}
+	require.NoError(t, ls.Provision(ctx, props))
+	require.NoError(t, ls.Connect(ctx))
+	s.Close()
+	_, err = ls.Lookup(ctx, []string{"a", "b"}, []string{"a"}, []any{1})
+	require.Error(t, err)
+	_, err = ls.Lookup(ctx, []string{"a", "b"}, []string{"a"}, []any{1})
+	require.Error(t, err)
+	s, err = testx.SetupEmbeddedMysqlServer(address, port)
+	require.NoError(t, err)
+	defer func() {
+		s.Close()
+	}()
+	got, err := ls.Lookup(ctx, []string{"a", "b"}, []string{"a"}, []any{1})
+	require.NoError(t, err)
+	require.Equal(t, []map[string]any{{"a": int64(1), "b": int64(1)}}, got)
+}
+
+func TestLookupPing(t *testing.T) {
+	connection.InitConnectionManager4Test()
+	s, err := testx.SetupEmbeddedMysqlServer(address, port)
+	require.NoError(t, err)
+	defer func() {
+		s.Close()
+	}()
+	props := map[string]interface{}{
+		"dburl":      fmt.Sprintf("mysql://root:@%v:%v/test", address, port),
+		"datasource": "t",
+	}
+	ls := &SqlLookupSource{}
+	require.NoError(t, ls.Ping("", props))
+	props = map[string]interface{}{
+		"dburl":      "",
+		"datasource": "t",
+	}
+	require.Error(t, ls.Ping("", props))
+}
diff --git a/extensions/impl/sql/sink_test.go b/extensions/impl/sql/sink_test.go
index 85b21a69..cfbbe771 100644
--- a/extensions/impl/sql/sink_test.go
+++ b/extensions/impl/sql/sink_test.go
@@ -284,3 +284,40 @@ func TestSQLSinkAction(t *testing.T) {
 	}))
 	require.NoError(t, sqlSink.Close(ctx))
 }
+
+func TestSQLSinkReconnect(t *testing.T) {
+	connection.InitConnectionManager4Test()
+	ctx := mockContext.NewMockContext("1", "2")
+	s, err := testx.SetupEmbeddedMysqlServer(address, port)
+	require.NoError(t, err)
+	tableName := "t"
+	dburl := fmt.Sprintf("mysql://root:@%v:%v/test", address, port)
+	sqlSink := &SQLSinkConnector{}
+	require.NoError(t, sqlSink.Provision(ctx, map[string]interface{}{
+		"dburl":        dburl,
+		"table":        tableName,
+		"fields":       []string{"a", "b"},
+		"rowKindField": "action",
+		"keyField":     "a",
+	}))
+	require.NoError(t, sqlSink.Connect(ctx))
+	s.Close()
+	// update
+	require.Error(t, sqlSink.collect(ctx, map[string]any{
+		"a":      1,
+		"b":      2,
+		"action": "update",
+	}))
+	require.True(t, sqlSink.needReconnect)
+	s, err = testx.SetupEmbeddedMysqlServer(address, port)
+	require.NoError(t, err)
+	defer func() {
+		s.Close()
+	}()
+	require.NoError(t, sqlSink.collect(ctx, map[string]any{
+		"a":      1,
+		"b":      2,
+		"action": "update",
+	}))
+	require.False(t, sqlSink.needReconnect)
+}
diff --git a/internal/server/meta_init_test.go b/internal/server/meta_init_test.go
index 20609aa0..67101d67 100644
--- a/internal/server/meta_init_test.go
+++ b/internal/server/meta_init_test.go
@@ -16,6 +16,7 @@ package server
 
 import (
 	"bytes"
+	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
@@ -29,8 +30,13 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/client"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/sql/testx"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/meta"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
 type MetaTestSuite struct {
@@ -56,6 +62,31 @@ func (suite *MetaTestSuite) SetupTest() {
 	}
 }
 
+var (
+	address = "localhost"
+	port    = 33061
+)
+
+func (suite *MetaTestSuite) TestLookupPing() {
+	modules.RegisterConnection("sql", client.CreateConnection)
+	modules.RegisterLookupSource("sql", sql.GetLookupSource)
+	connection.InitConnectionManager4Test()
+	s, err := testx.SetupEmbeddedMysqlServer(address, port)
+	require.NoError(suite.T(), err)
+	defer func() {
+		s.Close()
+	}()
+	props := map[string]interface{}{
+		"dburl":      fmt.Sprintf("mysql://root:@%v:%v/test", address, port),
+		"datasource": "t",
+	}
+	b, _ := json.Marshal(props)
+	req, _ := http.NewRequest(http.MethodPost, "/metadata/lookups/connection/sql", bytes.NewBuffer(b))
+	w := httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	require.Equal(suite.T(), http.StatusOK, w.Code)
+}
+
 func (suite *MetaTestSuite) TestSinksMetaHandler() {
 	req, _ := http.NewRequest(http.MethodGet, "/metadata/sinks", bytes.NewBufferString("any"))
 	w := httptest.NewRecorder()
diff --git a/internal/server/rpc_script_test.go b/internal/server/rpc_script_test.go
new file mode 100644
index 00000000..0f31945b
--- /dev/null
+++ b/internal/server/rpc_script_test.go
@@ -0,0 +1,72 @@
+// Copyright 2022-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.
+
+//go:build full || (rpc && script)
+
+package server
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/plugin/js"
+)
+
+func init() {
+	// Wait for other db tests to finish to avoid db lock
+	for i := 0; i < 10; i++ {
+		if err := js.InitManager(); err != nil {
+			time.Sleep(10 * time.Millisecond)
+		} else {
+			break
+		}
+	}
+}
+
+func TestFullLC(t *testing.T) {
+	// Create
+	s := new(Server)
+	reply := new(string)
+	err := s.CreateScript(`nojson`, reply)
+	assert.Error(t, err)
+	err = s.CreateScript(`{"id": "test", "source": "function() {}"}`, new(string))
+	assert.Error(t, err)
+	err = s.CreateScript(`{"id": "test", "script": "function test(a, b) {return a+b}"}`, reply)
+	assert.NoError(t, err)
+	assert.Equal(t, "JavaScript function test is created.", *reply)
+	// Get
+	err = s.DescScript("test", reply)
+	assert.NoError(t, err)
+	assert.Equal(t, "{\n  \"id\": \"test\",\n  \"description\": \"\",\n  \"script\": \"function test(a, b) {return a+b}\",\n  \"isAgg\": false\n}", *reply)
+	// List
+	err = s.ShowScripts(0, reply)
+	assert.NoError(t, err)
+	assert.Equal(t, "[\n  \"test\"\n]", *reply)
+	// Delete
+	err = s.DropScript("test", reply)
+	assert.NoError(t, err)
+	assert.Equal(t, "JavaScript function test is dropped", *reply)
+	// Get
+	err = s.DescScript("test", reply)
+	assert.Error(t, err)
+	// List
+	err = s.ShowScripts(0, reply)
+	assert.NoError(t, err)
+	assert.Equal(t, "No JavaScript functions are found.", *reply)
+	// Delete inexist
+	err = s.DropScript("test", reply)
+	assert.Error(t, err)
+}
diff --git a/internal/server/rpc_test.go b/internal/server/rpc_test.go
new file mode 100644
index 00000000..d196db4b
--- /dev/null
+++ b/internal/server/rpc_test.go
@@ -0,0 +1,238 @@
+// Copyright 2023-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+
+	"github.com/lf-edge/ekuiper/v2/internal/meta"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/internal/plugin/native"
+	"github.com/lf-edge/ekuiper/v2/internal/plugin/portable"
+	"github.com/lf-edge/ekuiper/v2/internal/schema"
+	"github.com/lf-edge/ekuiper/v2/internal/service"
+)
+
+type ServerTestSuite struct {
+	suite.Suite
+	s *Server
+}
+
+func (suite *ServerTestSuite) SetupTest() {
+	suite.s = new(Server)
+	nativeManager, _ = native.InitManager()
+	portableManager, _ = portable.InitManager()
+	serviceManager, _ = service.InitManager()
+	_ = schema.InitRegistry()
+	meta.InitYamlConfigManager()
+}
+
+func (suite *ServerTestSuite) TestStream() {
+	sql := `Create Stream test () WITH (FORMAT="JSON", type="simulator");`
+	var reply string
+	err := suite.s.Stream(sql, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "Stream test is created.\n", reply)
+
+	reply = ""
+	sql = "show streams;"
+	err = suite.s.Stream(sql, &reply)
+	assert.Nil(suite.T(), err)
+
+	reply = ""
+	sql = "SELECT * FROM test;"
+	err = suite.s.CreateQuery(sql, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "Query was submit successfully.", reply)
+
+	var result string = ""
+	for i := 0; i < 5; i++ {
+		var queryresult string
+		time.Sleep(time.Second)
+		err = suite.s.GetQueryResult("test", &queryresult)
+		assert.Nil(suite.T(), err)
+		if queryresult != "" {
+			result += queryresult
+			break
+		}
+	}
+	assert.Equal(suite.T(), "[{\"humidity\":50,\"temperature\":22.5}]", result)
+	stopQuery()
+}
+
+func (suite *ServerTestSuite) TestRule() {
+	sql := `Create Stream test () WITH (DATASOURCE="../internal/server/rpc_test_data/test.json", FORMAT="JSON", type="file");`
+	var reply string
+	err := suite.s.Stream(sql, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "Stream test is created.\n", reply)
+
+	reply = ""
+	rule := `{
+			  "sql": "SELECT * from test;",
+			  "actions": [{
+				"file": {
+				  "path": "../internal/server/rpc_test_data/data/result.txt",
+				  "interval": 5000,
+				  "fileType": "lines",
+				  "format": "json"
+				}
+			  }]
+	}`
+	ruleId := "myRule"
+	args := &model.RPCArgDesc{Name: ruleId, Json: rule}
+	err = suite.s.ValidateRule(args, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "The rule has been successfully validated and is confirmed to be correct.", reply)
+
+	reply = ""
+	rule = `{
+			  "sql": "SELECT * from test;"
+			}`
+	args = &model.RPCArgDesc{Name: ruleId, Json: rule}
+	err = suite.s.ValidateRule(args, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "invalid rule json: Missing rule actions.", reply)
+
+	reply = ""
+	rule = `{
+			  "sql": "SELECT * from test;",
+			  "actions": [{
+				"file": {
+				  "path": "../internal/server/rpc_test_data/data/result.txt",
+				  "interval": 5000,
+				  "fileType": "lines",
+				  "format": "json"
+				}
+			  }]
+	}`
+	args = &model.RPCArgDesc{Name: ruleId, Json: rule}
+	err = suite.s.CreateRule(args, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "Rule myRule was created successfully, please use 'bin/kuiper getstatus rule myRule' command to get rule status.", reply)
+
+	reply = ""
+	err = suite.s.GetStatusRule(ruleId, &reply)
+	assert.Nil(suite.T(), err)
+
+	reply = ""
+	err = suite.s.ShowRules(1, &reply)
+	assert.Nil(suite.T(), err)
+
+	reply = ""
+	err = suite.s.DescRule(ruleId, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "{\n  \"sql\": \"SELECT * from test;\",\n  \"actions\": [\n    {\n      \"file\": {\n        \"path\": \"../internal/server/rpc_test_data/data/result.txt\",\n        \"interval\": 5000,\n        \"fileType\": \"lines\",\n        \"format\": \"json\"\n      }\n    }\n  ]\n}\n", reply)
+
+	reply = ""
+	err = suite.s.GetTopoRule(ruleId, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "{\n  \"sources\": [\n    \"source_test\"\n  ],\n  \"edges\": {\n    \"op_2_decoder\": [\n      \"op_3_project\"\n    ],\n    \"op_3_project\": [\n      \"op_file_0_0_transform\"\n    ],\n    \"op_file_0_0_transform\": [\n      \"op_file_0_1_encode\"\n    ],\n    \"op_file_0_1_encode\": [\n      \"sink_file_0\"\n    ],\n    \"source_test\": [\n      \"op_2_decoder\"\n    ]\n  }\n}", reply)
+
+	reply = ""
+	err = suite.s.StopRule(ruleId, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "Rule myRule was stopped.", reply)
+	fmt.Println("rule stopped")
+
+	reply = ""
+	err = suite.s.StartRule(ruleId, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "Rule myRule was started", reply)
+	fmt.Println("rule started")
+
+	reply = ""
+	err = suite.s.RestartRule(ruleId, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "Rule myRule was restarted.", reply)
+	fmt.Println("rule restarted")
+
+	reply = ""
+	err = suite.s.DropRule(ruleId, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "Rule myRule is dropped.", reply)
+}
+
+func (suite *ServerTestSuite) TestImportAndExport() {
+	file := "rpc_test_data/import.json"
+	var reply string
+	err := suite.s.Import(file, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "imported 1 streams, 0 tables and 1 rules", reply)
+
+	reply = ""
+	file = "rpc_test_data/export.json"
+	err = suite.s.Export(file, &reply)
+	assert.Nil(suite.T(), err)
+	os.Remove(file)
+}
+
+func (suite *ServerTestSuite) TestConfiguration() {
+	importArg := model.ImportDataDesc{
+		FileName: "rpc_test_data/import_configuration.json",
+		Stop:     false,
+		Partial:  false,
+	}
+	var reply string
+	err := suite.s.ImportConfiguration(&importArg, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "{\n  \"ErrorMsg\": \"\",\n  \"ConfigResponse\": {\n    \"streams\": {},\n    \"tables\": {},\n    \"rules\": {},\n    \"nativePlugins\": {},\n    \"portablePlugins\": {},\n    \"sourceConfig\": {},\n    \"sinkConfig\": {},\n    \"connectionConfig\": {},\n    \"Service\": {},\n    \"Schema\": {},\n    \"uploads\": {},\n    \"scripts\": {}\n  }\n}", reply)
+
+	reply = ""
+	err = suite.s.GetStatusImport(1, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "{\n  \"streams\": {},\n  \"tables\": {},\n  \"rules\": {},\n  \"nativePlugins\": {},\n  \"portablePlugins\": {},\n  \"sourceConfig\": {},\n  \"sinkConfig\": {},\n  \"connectionConfig\": {},\n  \"Service\": {},\n  \"Schema\": {},\n  \"uploads\": {},\n  \"scripts\": {}\n}", reply)
+
+	reply = ""
+	exportArg := model.ExportDataDesc{
+		FileName: "rpc_test_data/export_configuration.json",
+		Rules:    []string{},
+	}
+	err = suite.s.ExportConfiguration(&exportArg, &reply)
+	assert.Nil(suite.T(), err)
+	assert.Equal(suite.T(), "export configuration success", reply)
+	os.Remove("rpc_test_data/export_configuration.json")
+}
+
+func (suite *ServerTestSuite) TearDownTest() {
+	// Clean up
+	sql := "DROP STREAM test;"
+	var reply string
+	_ = suite.s.Stream(sql, &reply)
+	_ = suite.s.DropRule("myRule", &reply)
+}
+
+func TestServerTestSuite(t *testing.T) {
+	suite.Run(t, new(ServerTestSuite))
+}
+
+func TestGetStoppedMessage(t *testing.T) {
+	message := `"123","123","123"`
+	r, err := getStoppedState(message)
+	require.NoError(t, err)
+	v := map[string]string{}
+	err = json.Unmarshal([]byte(r), &v)
+	require.NoError(t, err)
+	require.Equal(t, "stopped", v["status"])
+	require.Equal(t, message, v["message"])
+}
diff --git a/internal/testx/testUtil.go b/internal/testx/testUtil.go
index 4cec0ee9..d74dbe98 100644
--- a/internal/testx/testUtil.go
+++ b/internal/testx/testUtil.go
@@ -114,3 +114,26 @@ func CreateWebsocketClient(ip string, port int, endpoint string) (*websocket.Con
 	conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
 	return conn, err
 }
+
+type MockTuple struct {
+	Map      map[string]any
+	Template map[string]string
+}
+
+func (m MockTuple) DynamicProps(template string) (string, bool) {
+	r, ok := m.Template[template]
+	return r, ok
+}
+
+func (m MockTuple) AllProps() map[string]string {
+	return m.Template
+}
+
+func (m MockTuple) Value(key, table string) (any, bool) {
+	v, ok := m.Map[key]
+	return v, ok
+}
+
+func (m MockTuple) ToMap() map[string]any {
+	return m.Map
+}