Repository: lf-edge/ekuiper
Pull request #2866 (2024-05-21 08:45:46)

feat(io): add file source in lines mode 

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

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2866?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/io/file/source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2866?src=pr&el=tree&filepath=internal%2Fio%2Ffile%2Fsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZmlsZS9zb3VyY2UuZ28=) | 25.38% | [127 Missing and 20 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2866?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [pkg/mock/test\_source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2866?src=pr&el=tree&filepath=pkg%2Fmock%2Ftest_source.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-cGtnL21vY2svdGVzdF9zb3VyY2UuZ28=) | 55.56% | [4 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2866?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

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


```diff
@@             Coverage Diff              @@
##             master-v2    #2866   +/-   ##
============================================
  Coverage             ?   68.45%           
============================================
  Files                ?      274           
  Lines                ?    26038           
  Branches             ?        0           
============================================
  Hits                 ?    17823           
  Misses               ?     6604           
  Partials             ?     1611           
```



</details>

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


Patch:
diff --git a/etc/kuiper.yaml b/etc/kuiper.yaml
index 266043c5..cef56fc5 100644
--- a/etc/kuiper.yaml
+++ b/etc/kuiper.yaml
@@ -63,6 +63,7 @@ basic:
   enableOpenZiti: false
   # AES Key, base64 encoded
   aesKey: MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3
+  gracefulShutdownTimeout: 3s
 
 # The default options for all rules. Each rule can override this setting by defining its own option
 rule:
diff --git a/go.mod b/go.mod
index 824b3dba..77c570dc 100644
--- a/go.mod
+++ b/go.mod
@@ -23,7 +23,7 @@ require (
 	github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1
 	github.com/klauspost/compress v1.17.7
 	github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible
-	github.com/lf-edge/ekuiper/contract/v2 v2.0.0-alpha.1
+	github.com/lf-edge/ekuiper/contract/v2 v2.0.0-alpha.2
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/modern-go/reflect2 v1.0.2
 	github.com/montanaflynn/stats v0.7.1
@@ -40,6 +40,7 @@ require (
 	github.com/urfave/cli v1.22.14
 	github.com/valyala/fastjson v1.6.4
 	go.uber.org/automaxprocs v1.5.3
+	golang.org/x/exp v0.0.0-20231108232855-2478ac86f678
 	golang.org/x/text v0.14.0
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237
 	google.golang.org/grpc v1.62.1
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index 7e1bebb6..c021f983 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -27,7 +27,7 @@ func init() {
 	modules.RegisterSource("mqtt", mqtt.GetSource)
 	// modules.RegisterSource("httppull", func() api.Source { return &http.PullSource{} })
 	// modules.RegisterSource("httppush", func() api.Source { return &http.PushSource{} })
-	// modules.RegisterSource("file", func() api.Source { return &file.FileSource{} })
+	modules.RegisterSource("file", file.GetSource)
 	modules.RegisterSource("memory", func() api.Source { return memory.GetSource() })
 	// modules.RegisterSource("neuron", func() api.Source { return neuron.GetSource() })
 	// modules.RegisterSource("websocket", func() api.Source { return &websocket.WebsocketSource{} })
diff --git a/internal/conf/conf.go b/internal/conf/conf.go
index 9ca2c2c7..9c9e5736 100644
--- a/internal/conf/conf.go
+++ b/internal/conf/conf.go
@@ -169,32 +169,33 @@ func (s *syslogConf) Validate() error {
 
 type KuiperConf struct {
 	Basic struct {
-		LogLevel            string            `yaml:"logLevel"`
-		Debug               bool              `yaml:"debug"`
-		ConsoleLog          bool              `yaml:"consoleLog"`
-		FileLog             bool              `yaml:"fileLog"`
-		LogDisableTimestamp bool              `yaml:"logDisableTimestamp"`
-		Syslog              *syslogConf       `yaml:"syslog"`
-		RotateTime          int               `yaml:"rotateTime"`
-		MaxAge              int               `yaml:"maxAge"`
-		RotateSize          int64             `yaml:"rotateSize"`
-		RotateCount         int               `yaml:"rotateCount"`
-		TimeZone            string            `yaml:"timezone"`
-		Ip                  string            `yaml:"ip"`
-		Port                int               `yaml:"port"`
-		RestIp              string            `yaml:"restIp"`
-		RestPort            int               `yaml:"restPort"`
-		RestTls             *tlsConf          `yaml:"restTls"`
-		Prometheus          bool              `yaml:"prometheus"`
-		PrometheusPort      int               `yaml:"prometheusPort"`
-		PluginHosts         string            `yaml:"pluginHosts"`
-		Authentication      bool              `yaml:"authentication"`
-		IgnoreCase          bool              `yaml:"ignoreCase"`
-		SQLConf             *SQLConf          `yaml:"sql"`
-		RulePatrolInterval  cast.DurationConf `yaml:"rulePatrolInterval"`
-		CfgStorageType      string            `yaml:"cfgStorageType"`
-		EnableOpenZiti      bool              `yaml:"enableOpenZiti"`
-		AesKey              string            `yaml:"aesKey"`
+		LogLevel                string            `yaml:"logLevel"`
+		Debug                   bool              `yaml:"debug"`
+		ConsoleLog              bool              `yaml:"consoleLog"`
+		FileLog                 bool              `yaml:"fileLog"`
+		LogDisableTimestamp     bool              `yaml:"logDisableTimestamp"`
+		Syslog                  *syslogConf       `yaml:"syslog"`
+		RotateTime              int               `yaml:"rotateTime"`
+		MaxAge                  int               `yaml:"maxAge"`
+		RotateSize              int64             `yaml:"rotateSize"`
+		RotateCount             int               `yaml:"rotateCount"`
+		TimeZone                string            `yaml:"timezone"`
+		Ip                      string            `yaml:"ip"`
+		Port                    int               `yaml:"port"`
+		RestIp                  string            `yaml:"restIp"`
+		RestPort                int               `yaml:"restPort"`
+		RestTls                 *tlsConf          `yaml:"restTls"`
+		Prometheus              bool              `yaml:"prometheus"`
+		PrometheusPort          int               `yaml:"prometheusPort"`
+		PluginHosts             string            `yaml:"pluginHosts"`
+		Authentication          bool              `yaml:"authentication"`
+		IgnoreCase              bool              `yaml:"ignoreCase"`
+		SQLConf                 *SQLConf          `yaml:"sql"`
+		RulePatrolInterval      cast.DurationConf `yaml:"rulePatrolInterval"`
+		CfgStorageType          string            `yaml:"cfgStorageType"`
+		EnableOpenZiti          bool              `yaml:"enableOpenZiti"`
+		AesKey                  string            `yaml:"aesKey"`
+		GracefulShutdownTimeout cast.DurationConf `yaml:"gracefulShutdownTimeout"`
 	}
 	Rule   def.RuleOption
 	Sink   *SinkConf
@@ -356,6 +357,10 @@ func InitConf() {
 		}
 	}
 
+	if time.Duration(Config.Basic.GracefulShutdownTimeout) < 1 {
+		Config.Basic.GracefulShutdownTimeout = cast.DurationConf(3 * time.Second)
+	}
+
 	if Config.Basic.TimeZone != "" {
 		if err := cast.SetTimeZone(Config.Basic.TimeZone); err != nil {
 			Log.Fatal(err)
diff --git a/internal/io/file/source.go b/internal/io/file/source.go
new file mode 100644
index 00000000..25483b3c
--- /dev/null
+++ b/internal/io/file/source.go
@@ -0,0 +1,347 @@
+// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package file
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"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/timex"
+)
+
+type SourceConfig struct {
+	FileName         string        `json:"datasource"`
+	FileType         FileType      `json:"fileType"`
+	Path             string        `json:"path"`
+	Interval         time.Duration `json:"interval"`
+	IsTable          bool          `json:"isTable"`
+	Parallel         bool          `json:"parallel"`
+	SendInterval     time.Duration `json:"sendInterval"`
+	ActionAfterRead  int           `json:"actionAfterRead"`
+	MoveTo           string        `json:"moveTo"`
+	IgnoreStartLines int           `json:"ignoreStartLines"`
+	IgnoreEndLines   int           `json:"ignoreEndLines"`
+	// For csv only
+	HasHeader bool     `json:"hasHeader"`
+	Columns   []string `json:"columns"`
+	Delimiter string   `json:"delimiter"`
+	// Only use for planning
+	Decompression string `json:"decompression"`
+}
+
+// Source load data from file system.
+// If file type is lines, it reads line by line.
+// Otherwise, it reads the file as a whole and send to company reader node to read and split.
+// The planner need to plan according to the file type.(currently, it is hardcoded to deal with file source only)
+type Source struct {
+	file   string
+	isDir  bool
+	config *SourceConfig
+	eof    api.EOFIngest
+}
+
+func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
+	cfg := &SourceConfig{
+		FileType: JSON_TYPE,
+	}
+	err := cast.MapToStruct(props, cfg)
+	if err != nil {
+		return fmt.Errorf("read properties %v fail with error: %v", props, err)
+	}
+	if cfg.FileType == "" {
+		return errors.New("missing or invalid property fileType, must be 'json'")
+	}
+	if _, ok := fileTypes[cfg.FileType]; !ok {
+		return fmt.Errorf("invalid property fileType: %s", cfg.FileType)
+	}
+	if cfg.Path == "" {
+		return errors.New("missing property Path")
+	}
+	if !filepath.IsAbs(cfg.Path) {
+		p, err := conf.GetLoc(cfg.Path)
+		if err != nil {
+			return fmt.Errorf("invalid path %s", cfg.Path)
+		}
+		cfg.Path = p
+	}
+	fs.file = filepath.Join(cfg.Path, cfg.FileName)
+	fi, err := os.Stat(fs.file)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return fmt.Errorf("file %s not exist", fs.file)
+		}
+	}
+	if fi.IsDir() {
+		fs.isDir = true
+	}
+	if cfg.IgnoreStartLines < 0 {
+		cfg.IgnoreStartLines = 0
+	}
+	if cfg.IgnoreEndLines < 0 {
+		cfg.IgnoreEndLines = 0
+	}
+	if cfg.ActionAfterRead < 0 || cfg.ActionAfterRead > 2 {
+		return fmt.Errorf("invalid actionAfterRead: %d", cfg.ActionAfterRead)
+	}
+	if cfg.ActionAfterRead == 2 {
+		if cfg.MoveTo == "" {
+			return fmt.Errorf("missing moveTo when actionAfterRead is 2")
+		} else {
+			if !filepath.IsAbs(cfg.MoveTo) {
+				cfg.MoveTo, err = conf.GetLoc(cfg.MoveTo)
+				if err != nil {
+					return fmt.Errorf("invalid moveTo %s: %v", cfg.MoveTo, err)
+				}
+			}
+			fileInfo, err := os.Stat(cfg.MoveTo)
+			if err != nil {
+				err := os.MkdirAll(cfg.MoveTo, os.ModePerm)
+				if err != nil {
+					return fmt.Errorf("fail to create dir for moveTo %s: %v", cfg.MoveTo, err)
+				}
+			} else if !fileInfo.IsDir() {
+				return fmt.Errorf("moveTo %s is not a directory", cfg.MoveTo)
+			}
+		}
+	}
+	if cfg.Delimiter == "" {
+		cfg.Delimiter = ","
+	}
+
+	fs.config = cfg
+	return nil
+}
+
+func (fs *Source) Connect(ctx api.StreamContext) error {
+	// do nothing
+	return nil
+}
+
+// Subscribe file source always ingest bytes
+// For stream source, it ingest one line
+// For batch source, it ingest the whole file, thus it need a reader node to coordinate and read the content into lines/array
+func (fs *Source) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error {
+	fs.Load(ctx, ingest, ingestError)
+	if fs.config.Interval > 0 {
+		ticker := time.NewTicker(fs.config.Interval)
+		logger := ctx.GetLogger()
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				logger.Debugf("Load file source again at %v", timex.GetNowInMilli())
+				fs.Load(ctx, ingest, ingestError)
+			case <-ctx.Done():
+				return nil
+			}
+		}
+	} else { // If not pulling is set, it is a bounded source
+		if fs.eof != nil {
+			ctx.GetLogger().Debugf("send eof")
+			fs.eof(ctx)
+		}
+	}
+	return nil
+}
+
+func (fs *Source) SetEofIngest(eof api.EOFIngest) {
+	fs.eof = eof
+}
+
+func (fs *Source) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Close file source")
+	// do nothing
+	return nil
+}
+
+func (fs *Source) Load(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) {
+	if fs.isDir {
+		ctx.GetLogger().Debugf("Monitor dir %s", fs.file)
+		entries, err := os.ReadDir(fs.file)
+		// may be just forget to put in the file
+		if err != nil {
+			ingestError(ctx, err)
+		}
+		if fs.config.Parallel {
+			var wg sync.WaitGroup
+			for _, entry := range entries {
+				if entry.IsDir() {
+					continue
+				}
+				wg.Add(1)
+				go func(file string) {
+					defer wg.Done()
+					fs.parseFile(ctx, file, ingest, ingestError)
+				}(filepath.Join(fs.file, entry.Name()))
+			}
+			wg.Wait()
+		} else {
+			for _, entry := range entries {
+				if entry.IsDir() {
+					continue
+				}
+				file := filepath.Join(fs.file, entry.Name())
+				fs.parseFile(ctx, file, ingest, ingestError)
+			}
+		}
+	} else {
+		fs.parseFile(ctx, fs.file, ingest, ingestError)
+	}
+	// TODO deal with table case
+	//if fs.config.IsTable {
+	//	select {
+	//	case consumer <- api.NewDefaultSourceTupleWithTime(nil, nil, rcvTime):
+	//		// do nothing
+	//	case <-ctx.Done():
+	//		return nil
+	//	}
+	//}
+	ctx.GetLogger().Debug("All tuples sent")
+}
+
+func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.BytesIngest, ingestError api.ErrorIngest) {
+	var (
+		err error
+		r   io.Reader
+	)
+	r, err = os.Open(file)
+	if err != nil {
+		ctx.GetLogger().Debugf("prepare file %s error: %v", file, err)
+		ingestError(ctx, err)
+	}
+	if fs.config.IgnoreStartLines > 0 || fs.config.IgnoreEndLines > 0 {
+		r = ignoreLines(ctx, r, fs.config.IgnoreStartLines, fs.config.IgnoreEndLines)
+	}
+	if closer, ok := r.(io.Closer); ok {
+		defer func() {
+			ctx.GetLogger().Debugf("Close reader")
+			closer.Close()
+		}()
+	}
+	meta := map[string]any{"file": file}
+	// Read line or read all
+	switch fs.config.FileType {
+	case LINES_TYPE: // read line by line
+		scanner := bufio.NewScanner(r)
+		scanner.Split(bufio.ScanLines)
+		for scanner.Scan() {
+			rcvTime := timex.GetNow()
+			fmt.Println("received", rcvTime.UnixMilli())
+			line := scanner.Bytes()
+			ingest(ctx, line, meta, rcvTime)
+			if fs.config.SendInterval > 0 {
+				time.Sleep(fs.config.SendInterval)
+			}
+		}
+		// Check for any errors during scanning
+		if err := scanner.Err(); err != nil {
+			ingestError(ctx, err)
+			return
+		}
+	default: // read all
+		rcvTime := timex.GetNow()
+		content, err := os.ReadFile(file)
+		if err != nil {
+			ingestError(ctx, err)
+			// have error, do not need to do action after read
+			return
+		} else {
+			ingest(ctx, content, meta, rcvTime)
+		}
+	}
+	ctx.GetLogger().Debugf("Finish loading from file %s", file)
+	switch fs.config.ActionAfterRead {
+	case 1:
+		if err := os.Remove(file); err != nil {
+			ingestError(ctx, err)
+		}
+		ctx.GetLogger().Debugf("Remove file %s", file)
+	case 2:
+		targetFile := filepath.Join(fs.config.MoveTo, filepath.Base(file))
+		if err := os.Rename(file, targetFile); err != nil {
+			ingestError(ctx, err)
+		}
+		ctx.GetLogger().Debugf("Move file %s to %s", file, targetFile)
+	}
+}
+
+func ignoreLines(ctx api.StreamContext, reader io.Reader, ignoreStartLines int, ignoreEndLines int) io.Reader {
+	r, w := io.Pipe()
+	go func() {
+		defer func() {
+			w.Close()
+			reader.(io.ReadCloser).Close()
+		}()
+		scanner := bufio.NewScanner(reader)
+		scanner.Split(bufio.ScanLines)
+
+		ln := 0
+		// This is a queue to store the lines that should be ignored
+		tempLines := make([][]byte, 0, ignoreEndLines)
+		for scanner.Scan() {
+			if ln >= ignoreStartLines {
+				if ignoreEndLines > 0 { // the last n line are left in the tempLines
+					slot := (ln - ignoreStartLines) % ignoreEndLines
+					if len(tempLines) <= slot { // first round
+						tempLines = append(tempLines, scanner.Bytes())
+					} else {
+						_, err := w.Write(tempLines[slot])
+						if err != nil {
+							ctx.GetLogger().Error(err)
+							break
+						}
+						_, err = w.Write([]byte{'\n'})
+						if err != nil {
+							ctx.GetLogger().Error(err)
+							break
+						}
+						tempLines[slot] = scanner.Bytes()
+					}
+				} else {
+					_, err := w.Write(scanner.Bytes())
+					if err != nil {
+						ctx.GetLogger().Error(err)
+						break
+					}
+					_, err = w.Write([]byte{'\n'})
+					if err != nil {
+						ctx.GetLogger().Error(err)
+						break
+					}
+				}
+			}
+			ln++
+		}
+	}()
+	return r
+}
+
+func GetSource() api.Source {
+	return &Source{}
+}
+
+var (
+	_ api.BytesSource = &Source{}
+	_ api.Bounded     = &Source{}
+)
diff --git a/internal/server/rule_manager.go b/internal/server/rule_manager.go
index a9c544f8..8acf0520 100644
--- a/internal/server/rule_manager.go
+++ b/internal/server/rule_manager.go
@@ -254,6 +254,16 @@ func stopRuleInternal(name string) {
 	}
 }
 
+func stopRuleWhenServerStop(name string) {
+	var err error
+	if rs, ok := registry.Load(name); ok {
+		err = rs.Stop()
+		if err != nil {
+			conf.Log.Warn(err)
+		}
+	}
+}
+
 func stopRule(name string) (result string, err error) {
 	if rs, ok := registry.Load(name); ok {
 		err = rs.Stop()
diff --git a/internal/server/server.go b/internal/server/server.go
index 0a3c4b13..bff7b80a 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -24,6 +24,7 @@ import (
 	"os/signal"
 	"path/filepath"
 	"sort"
+	"sync"
 	"syscall"
 	"time"
 
@@ -251,13 +252,30 @@ func StartUp(Version string) {
 		conf.Log.Info("eKuiper stopped by Stop request")
 	}
 	exit <- struct{}{}
-	conf.Log.Info("start to stop rest server")
-	ctx, cancel := context.WithTimeout(context.TODO(), 3*time.Second)
+	// wait rule checker exit
+	time.Sleep(10 * time.Millisecond)
+
+	ctx, cancel := context.WithTimeout(context.TODO(), time.Duration(conf.Config.Basic.GracefulShutdownTimeout))
 	defer cancel()
-	if err = srvRest.Shutdown(ctx); err != nil {
-		logger.Errorf("rest server shutdown error: %v", err)
-	}
-	logger.Info("rest server successfully shutdown.")
+	wg := sync.WaitGroup{}
+	// wait all service stop
+	wg.Add(2)
+	go func() {
+		conf.Log.Info("start to stop all rules")
+		if err := waitAllRuleStop(ctx); err != nil {
+			conf.Log.Warnf(err.Error())
+		}
+		wg.Done()
+	}()
+	go func() {
+		conf.Log.Info("start to stop rest server")
+		if err = srvRest.Shutdown(ctx); err != nil {
+			logger.Errorf("rest server shutdown error: %v", err)
+		}
+		logger.Info("rest server successfully shutdown.")
+		wg.Done()
+	}()
+	wg.Wait()
 
 	// close extend services
 	for k, v := range servers {
@@ -439,3 +457,43 @@ func handleScheduleRule(now time.Time, r *def.Rule, state string) scheduleRuleAc
 	}
 	return scheduleRuleActionDoNothing
 }
+
+func waitAllRuleStop(ctx context.Context) error {
+	rules, _ := ruleProcessor.GetAllRules()
+	for _, r := range rules {
+		stopRuleWhenServerStop(r)
+	}
+	wg := &sync.WaitGroup{}
+	m := &sync.Map{}
+	for _, r := range rules {
+		rs, ok := registry.Load(r)
+		if ok {
+			m.Store(r, struct{}{})
+			wg.Add(1)
+			go func() {
+				defer func() {
+					m.Delete(r)
+					wg.Done()
+				}()
+				rs.Topology.WaitClose()
+			}()
+		}
+	}
+	wgCh := make(chan struct{})
+	go func() {
+		wg.Wait()
+		wgCh <- struct{}{}
+	}()
+
+	select {
+	case <-ctx.Done():
+		timeoutRules := make([]string, 0)
+		m.Range(func(key, value any) bool {
+			timeoutRules = append(timeoutRules, key.(string))
+			return true
+		})
+		return fmt.Errorf("stop rules timeout, remain::%s", timeoutRules)
+	case <-wgCh:
+		return nil
+	}
+}
diff --git a/internal/topo/context/default.go b/internal/topo/context/default.go
index a45c4d34..fdafba69 100644
--- a/internal/topo/context/default.go
+++ b/internal/topo/context/default.go
@@ -32,8 +32,9 @@ import (
 )
 
 const (
-	LoggerKey    = "$$logger"
-	RuleStartKey = "$$ruleStart"
+	LoggerKey        = "$$logger"
+	RuleStartKey     = "$$ruleStart"
+	RuleWaitGroupKey = "$$ruleWaitGroup"
 )
 
 type DefaultContext struct {
@@ -49,6 +50,9 @@ type DefaultContext struct {
 	// cache
 	tpReg sync.Map
 	jpReg sync.Map
+
+	// ops waitGroup
+	opsWg *sync.WaitGroup
 }
 
 func Background() *DefaultContext {
@@ -63,6 +67,11 @@ func WithValue(parent *DefaultContext, key, val interface{}) *DefaultContext {
 	return parent
 }
 
+func WithWg(parent *DefaultContext, wg *sync.WaitGroup) *DefaultContext {
+	parent.opsWg = wg
+	return parent
+}
+
 // Deadline Implement context interface
 func (c *DefaultContext) Deadline() (deadline time.Time, ok bool) {
 	return c.ctx.Deadline()
@@ -183,6 +192,7 @@ func (c *DefaultContext) WithMeta(ruleId string, opId string, store api.Store) a
 		state:      s,
 		tpReg:      sync.Map{},
 		jpReg:      sync.Map{},
+		opsWg:      c.opsWg,
 	}
 }
 
@@ -193,6 +203,7 @@ func (c *DefaultContext) WithInstance(instanceId int) api.StreamContext {
 		opId:       c.opId,
 		ctx:        c.ctx,
 		state:      c.state,
+		opsWg:      c.opsWg,
 	}
 }
 
@@ -204,6 +215,7 @@ func (c *DefaultContext) WithCancel() (api.StreamContext, context.CancelFunc) {
 		instanceId: c.instanceId,
 		ctx:        ctx,
 		state:      c.state,
+		opsWg:      c.opsWg,
 	}, cancel
 }
 
diff --git a/internal/topo/node/batch_op.go b/internal/topo/node/batch_op.go
index f142f1e4..b5523b17 100644
--- a/internal/topo/node/batch_op.go
+++ b/internal/topo/node/batch_op.go
@@ -68,7 +68,10 @@ func (b *BatchOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 func (b *BatchOp) runWithTickerAndBatchSize(ctx api.StreamContext) {
 	ticker := timex.GetTicker(b.lingerInterval)
 	go func() {
-		defer ticker.Stop()
+		defer func() {
+			ticker.Stop()
+			b.Close()
+		}()
 		for {
 			select {
 			case <-ctx.Done():
@@ -122,6 +125,9 @@ func (b *BatchOp) send() {
 
 func (b *BatchOp) runWithBatchSize(ctx api.StreamContext) {
 	go func() {
+		defer func() {
+			b.Close()
+		}()
 		for {
 			select {
 			case <-ctx.Done():
@@ -136,7 +142,10 @@ func (b *BatchOp) runWithBatchSize(ctx api.StreamContext) {
 func (b *BatchOp) runWithTicker(ctx api.StreamContext) {
 	ticker := timex.GetTicker(b.lingerInterval)
 	go func() {
-		defer ticker.Stop()
+		defer func() {
+			ticker.Stop()
+			b.Close()
+		}()
 		for {
 			select {
 			case <-ctx.Done():
diff --git a/internal/topo/node/cache_op.go b/internal/topo/node/cache_op.go
index 8368a75f..74a33d24 100644
--- a/internal/topo/node/cache_op.go
+++ b/internal/topo/node/cache_op.go
@@ -67,6 +67,9 @@ func (s *CacheOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}
 	s.prepareExec(ctx, errCh, "op")
 	go func() {
+		defer func() {
+			s.Close()
+		}()
 		for {
 			select {
 			case <-ctx.Done():
diff --git a/internal/topo/node/compress_op.go b/internal/topo/node/compress_op.go
index 3bf48aac..8b98e63c 100644
--- a/internal/topo/node/compress_op.go
+++ b/internal/topo/node/compress_op.go
@@ -43,6 +43,9 @@ func NewCompressOp(name string, rOpt *def.RuleOption, compressMethod string) (*C
 func (o *CompressOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	o.prepareExec(ctx, errCh, "op")
 	go func() {
+		defer func() {
+			o.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			runWithOrder(ctx, o.defaultSinkNode, o.concurrency, o.Worker)
 			return nil
diff --git a/internal/topo/node/decode_op.go b/internal/topo/node/decode_op.go
index 0360bf99..8d663085 100644
--- a/internal/topo/node/decode_op.go
+++ b/internal/topo/node/decode_op.go
@@ -82,6 +82,9 @@ func NewDecodeOp(name, StreamName string, ruleId string, rOpt *def.RuleOption, o
 func (o *DecodeOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	o.prepareExec(ctx, errCh, "op")
 	go func() {
+		defer func() {
+			o.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			runWithOrder(ctx, o.defaultSinkNode, o.concurrency, o.Worker)
 			return nil
diff --git a/internal/topo/node/decompress_op.go b/internal/topo/node/decompress_op.go
index 0ee3788d..2e3ea970 100644
--- a/internal/topo/node/decompress_op.go
+++ b/internal/topo/node/decompress_op.go
@@ -44,6 +44,9 @@ func NewDecompressOp(name string, rOpt *def.RuleOption, compressMethod string) (
 func (o *DecompressOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	o.prepareExec(ctx, errCh, "op")
 	go func() {
+		defer func() {
+			o.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			runWithOrder(ctx, o.defaultSinkNode, o.concurrency, o.Worker)
 			return nil
diff --git a/internal/topo/node/dedup_trigger_op.go b/internal/topo/node/dedup_trigger_op.go
index 593fdbf4..39d8c634 100644
--- a/internal/topo/node/dedup_trigger_op.go
+++ b/internal/topo/node/dedup_trigger_op.go
@@ -70,6 +70,9 @@ func (w *DedupTriggerNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	w.prepareExec(ctx, errCh, "op")
 
 	go func() {
+		defer func() {
+			w.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			for {
 				select {
diff --git a/internal/topo/node/encode_op.go b/internal/topo/node/encode_op.go
index 33796592..d2dac42e 100644
--- a/internal/topo/node/encode_op.go
+++ b/internal/topo/node/encode_op.go
@@ -51,6 +51,9 @@ func NewEncodeOp(name string, rOpt *def.RuleOption, sc *SinkConf) (*EncodeOp, er
 func (o *EncodeOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	o.prepareExec(ctx, errCh, "op")
 	go func() {
+		defer func() {
+			o.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			runWithOrder(ctx, o.defaultSinkNode, o.concurrency, o.Worker)
 			return nil
diff --git a/internal/topo/node/encrypt_op.go b/internal/topo/node/encrypt_op.go
index 098de314..d55f6fd0 100644
--- a/internal/topo/node/encrypt_op.go
+++ b/internal/topo/node/encrypt_op.go
@@ -47,6 +47,9 @@ func NewEncryptOp(name string, rOpt *def.RuleOption, encryptMethod string) (*Enc
 func (o *EncryptNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	o.prepareExec(ctx, errCh, "op")
 	go func() {
+		defer func() {
+			o.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			runWithOrder(ctx, o.defaultSinkNode, o.concurrency, o.Worker)
 			return nil
diff --git a/internal/topo/node/join_align_node.go b/internal/topo/node/join_align_node.go
index eae816a5..b30b4574 100644
--- a/internal/topo/node/join_align_node.go
+++ b/internal/topo/node/join_align_node.go
@@ -49,6 +49,9 @@ func (n *JoinAlignNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	n.prepareExec(ctx, errCh, "op")
 	log := ctx.GetLogger()
 	go func() {
+		defer func() {
+			n.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			// restore batch state
 			if s, err := ctx.GetState(BatchKey); err == nil {
diff --git a/internal/topo/node/lookup_node.go b/internal/topo/node/lookup_node.go
index 6eb088f9..e592c0e2 100644
--- a/internal/topo/node/lookup_node.go
+++ b/internal/topo/node/lookup_node.go
@@ -79,6 +79,9 @@ func (n *LookupNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	log := ctx.GetLogger()
 	n.prepareExec(ctx, errCh, "op")
 	go func() {
+		defer func() {
+			n.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			ns, err := lookup.Attach(n.name)
 			if err != nil {
diff --git a/internal/topo/node/node.go b/internal/topo/node/node.go
index 09a15e23..a7512dd0 100644
--- a/internal/topo/node/node.go
+++ b/internal/topo/node/node.go
@@ -24,6 +24,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/util"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/checkpoint"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/node/metric"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
@@ -39,6 +40,7 @@ type defaultNode struct {
 	qos         def.Qos
 	outputMu    sync.RWMutex
 	outputs     map[string]chan<- any
+	opsWg       *sync.WaitGroup
 }
 
 func newDefaultNode(name string, options *def.RuleOption) *defaultNode {
@@ -169,9 +171,26 @@ func (o *defaultNode) prepareExec(ctx api.StreamContext, errCh chan<- error, opT
 	ctx.GetLogger().Infof("%s started", o.name)
 	o.statManager = metric.NewStatManager(ctx, opType)
 	o.ctx = ctx
+	wg := ctx.Value(context.RuleWaitGroupKey)
+	if wg != nil {
+		o.opsWg = wg.(*sync.WaitGroup)
+	}
+	if o.opsWg != nil {
+		o.opsWg.Add(1)
+	}
 	o.ctrlCh = errCh
 }
 
+func (o *defaultNode) finishExec() {
+	o.Close()
+}
+
+func (o *defaultNode) Close() {
+	if o.opsWg != nil {
+		o.opsWg.Done()
+	}
+}
+
 func (o *defaultSinkNode) preprocess(ctx api.StreamContext, item any) (any, bool) {
 	if o.qos >= def.AtLeastOnce {
 		b, ok := item.(*checkpoint.BufferOrEvent)
diff --git a/internal/topo/node/operations.go b/internal/topo/node/operations.go
index f18e1910..b898568f 100644
--- a/internal/topo/node/operations.go
+++ b/internal/topo/node/operations.go
@@ -60,6 +60,9 @@ func (o *UnaryOperator) Exec(ctx api.StreamContext, errCh chan<- error) {
 		o.concurrency = 1
 	}
 	go func() {
+		defer func() {
+			o.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			o.doOp(ctx.WithInstance(0), errCh)
 			return nil
diff --git a/internal/topo/node/sink_node.go b/internal/topo/node/sink_node.go
index c1f1f311..9719a895 100644
--- a/internal/topo/node/sink_node.go
+++ b/internal/topo/node/sink_node.go
@@ -71,7 +71,10 @@ func (s *SinkNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 		if err != nil {
 			infra.DrainError(ctx, err, errCh)
 		}
-		defer s.sink.Close(ctx)
+		defer func() {
+			s.sink.Close(ctx)
+			s.Close()
+		}()
 		s.currentEof = 0
 		for {
 			select {
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index 9ad55811..fb3fe4bc 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -127,7 +127,10 @@ func (m *SourceNode) ingestEof(ctx api.StreamContext) {
 
 // Run Subscribe could be a long-running function
 func (m *SourceNode) Run(ctx api.StreamContext, ctrlCh chan<- error) {
-	defer m.s.Close(ctx)
+	defer func() {
+		m.s.Close(ctx)
+		m.Close()
+	}()
 	poe := infra.SafeRun(func() error {
 		err := m.s.Connect(ctx)
 		if err != nil {
@@ -149,3 +152,7 @@ func (m *SourceNode) Run(ctx api.StreamContext, ctrlCh chan<- error) {
 	}
 	<-ctx.Done()
 }
+
+func (m *SourceNode) Close() {
+	m.defaultNode.Close()
+}
diff --git a/internal/topo/node/switch_node.go b/internal/topo/node/switch_node.go
index 50552d4f..2856fe1b 100644
--- a/internal/topo/node/switch_node.go
+++ b/internal/topo/node/switch_node.go
@@ -72,6 +72,9 @@ func (n *SwitchNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}
 	fv, afv := xsql.NewFunctionValuersForOp(ctx)
 	go func() {
+		defer func() {
+			n.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			for {
 				ctx.GetLogger().Debugf("Switch node %s is looping", n.name)
diff --git a/internal/topo/node/transform_op.go b/internal/topo/node/transform_op.go
index 61b4ee23..2921d394 100644
--- a/internal/topo/node/transform_op.go
+++ b/internal/topo/node/transform_op.go
@@ -79,6 +79,9 @@ func NewTransformOp(name string, rOpt *def.RuleOption, sc *SinkConf, templates [
 func (t *TransformOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	t.prepareExec(ctx, errCh, "op")
 	go func() {
+		defer func() {
+			t.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			runWithOrder(ctx, t.defaultSinkNode, t.concurrency, t.Worker)
 			return nil
diff --git a/internal/topo/node/watermark_op.go b/internal/topo/node/watermark_op.go
index 305f116e..7b2b83cb 100644
--- a/internal/topo/node/watermark_op.go
+++ b/internal/topo/node/watermark_op.go
@@ -98,6 +98,9 @@ func (w *WatermarkOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 
 	ctx.GetLogger().Infof("Start with state lastWatermarkTs: %d", w.lastWatermarkTs.UnixMilli())
 	go func() {
+		defer func() {
+			w.Close()
+		}()
 		err := infra.SafeRun(func() error {
 			for {
 				select {
diff --git a/internal/topo/node/window_op.go b/internal/topo/node/window_op.go
index f2a0b851..5e16d7f9 100644
--- a/internal/topo/node/window_op.go
+++ b/internal/topo/node/window_op.go
@@ -103,6 +103,10 @@ func NewWindowOp(name string, w WindowConfig, options *def.RuleOption) (*WindowO
 	return o, nil
 }
 
+func (o *WindowOperator) Close() {
+	o.defaultNode.Close()
+}
+
 // Exec is the entry point for the executor
 // input: *xsql.Tuple from preprocessor
 // output: xsql.WindowTuplesSet
@@ -146,6 +150,9 @@ func (o *WindowOperator) Exec(ctx api.StreamContext, errCh chan<- error) {
 	log.Infof("Start with window state triggerTime: %d, msgCount: %d", o.triggerTime.UnixMilli(), o.msgCount)
 	if o.isEventTime {
 		go func() {
+			defer func() {
+				o.Close()
+			}()
 			err := infra.SafeRun(func() error {
 				o.execEventWindow(ctx, inputs, errCh)
 				return nil
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index 35665d19..1f688624 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -53,6 +53,8 @@ type Topo struct {
 	topo        *def.PrintableTopo
 	mu          sync.Mutex
 	hasOpened   atomic.Bool
+
+	opsWg *sync.WaitGroup
 }
 
 func NewWithNameAndOptions(name string, options *def.RuleOption) (*Topo, error) {
@@ -63,6 +65,7 @@ func NewWithNameAndOptions(name string, options *def.RuleOption) (*Topo, error)
 			Sources: make([]string, 0),
 			Edges:   make(map[string][]interface{}),
 		},
+		opsWg: &sync.WaitGroup{},
 	}
 	tp.prepareContext() // ensure context is set
 	return tp, nil
@@ -96,8 +99,7 @@ func (s *Topo) Cancel() {
 	s.store = nil
 	s.coordinator = nil
 	for _, src := range s.sources {
-		switch rt := src.(type) {
-		case node.MergeableTopo:
+		if rt, ok := src.(node.MergeableTopo); ok {
 			rt.Close(s.ctx, s.name)
 		}
 	}
@@ -209,6 +211,7 @@ func (s *Topo) prepareContext() {
 		}
 		ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
 		ctx = kctx.WithValue(ctx, kctx.RuleStartKey, timex.GetNowInMilli())
+		ctx = kctx.WithWg(ctx, s.opsWg)
 		s.ctx, s.cancel = ctx.WithCancel()
 	}
 }
@@ -354,3 +357,11 @@ func (s *Topo) ResetStreamOffset(name string, input map[string]interface{}) erro
 	}
 	return fmt.Errorf("stream %v not found in topo", name)
 }
+
+func (s *Topo) WaitClose() {
+	// wait all operators close
+	if s.opsWg != nil {
+		s.opsWg.Wait()
+	}
+	conf.Log.Infof("rule %s stopped", s.ctx.GetRuleId())
+}


Test patch:
diff --git a/internal/io/file/source_test.go b/internal/io/file/source_test.go
new file mode 100644
index 00000000..76402b3e
--- /dev/null
+++ b/internal/io/file/source_test.go
@@ -0,0 +1,344 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package file
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/mock"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+func TestSourceProvision(t *testing.T) {
+	// Create and write temp file
+	tmpfile, err := os.CreateTemp("", "test.lines")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.Remove(tmpfile.Name())
+	if _, err = tmpfile.Write([]byte("content")); err != nil {
+		t.Fatal(err)
+	}
+	if err = tmpfile.Close(); err != nil {
+		t.Fatal(err)
+	}
+	path := filepath.Dir(tmpfile.Name())
+	name := filepath.Base(tmpfile.Name())
+	_, wrongMoveToErr := conf.GetLoc("ddd/dd")
+	relPath, err := conf.GetLoc("data")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx := mockContext.NewMockContext("testsourcep", "test")
+	m := &Source{}
+	tests := []struct {
+		name  string
+		props map[string]any
+		c     *SourceConfig
+		e     string
+	}{
+		{
+			name: "normal",
+			props: map[string]any{
+				"datasource": name,
+				"path":       path,
+			},
+			c: &SourceConfig{
+				FileName:  name,
+				Path:      path,
+				FileType:  JSON_TYPE,
+				Delimiter: ",",
+			},
+		},
+		{
+			name: "invalid format",
+			props: map[string]any{
+				"datasource":   name,
+				"path":         path,
+				"sendInterval": "ddd",
+			},
+			e: fmt.Sprintf("read properties map[datasource:%s path:%s sendInterval:ddd] fail with error: 1 error(s) decoding:\n\n* error decoding 'sendInterval': time: invalid duration \"ddd\"", name, path),
+		},
+		{
+			name: "missing file type",
+			props: map[string]any{
+				"datasource": name,
+				"path":       path,
+				"fileType":   "",
+			},
+			e: "missing or invalid property fileType, must be 'json'",
+		},
+		{
+			name: "invalid file type",
+			props: map[string]any{
+				"datasource": name,
+				"path":       path,
+				"fileType":   "aaa",
+			},
+			e: "invalid property fileType: aaa",
+		},
+		{
+			name: "missing path",
+			props: map[string]any{
+				"datasource": name,
+			},
+			e: "missing property Path",
+		},
+		{
+			name: "invalid path",
+			props: map[string]any{
+				"datasource": name,
+				"path":       "dddd/ddd",
+			},
+			e: "invalid path dddd/ddd",
+		},
+		{
+			name: "invalid file",
+			props: map[string]any{
+				"datasource": "notexist",
+				"path":       path,
+			},
+			e: fmt.Sprintf("file %s not exist", filepath.Join(path, "notexist")),
+		},
+		{
+			name: "dir and lines",
+			props: map[string]any{
+				"datasource":       "",
+				"path":             "data",
+				"ignoreStartLines": -2,
+				"ignoreEndLines":   -2,
+			},
+			c: &SourceConfig{
+				FileName:         "",
+				Path:             relPath,
+				FileType:         JSON_TYPE,
+				Delimiter:        ",",
+				IgnoreStartLines: 0,
+				IgnoreEndLines:   0,
+			},
+		},
+		{
+			name: "wrong action after read",
+			props: map[string]any{
+				"datasource":      name,
+				"path":            path,
+				"actionAfterRead": 4,
+			},
+			e: "invalid actionAfterRead: 4",
+		},
+		{
+			name: "missing move to",
+			props: map[string]any{
+				"datasource":      name,
+				"path":            path,
+				"actionAfterRead": 2,
+			},
+			e: "missing moveTo when actionAfterRead is 2",
+		},
+		{
+			name: "wrong move to",
+			props: map[string]any{
+				"datasource":      name,
+				"path":            path,
+				"actionAfterRead": 2,
+				"moveTo":          "ddd/dd",
+			},
+			e: fmt.Sprintf("invalid moveTo : %v", wrongMoveToErr),
+		},
+		{
+			name: "move to dir",
+			props: map[string]any{
+				"datasource":      name,
+				"path":            path,
+				"actionAfterRead": 2,
+				"moveTo":          filepath.Join(path, "ddd"),
+			},
+			c: &SourceConfig{
+				FileName:        name,
+				Path:            path,
+				FileType:        JSON_TYPE,
+				Delimiter:       ",",
+				ActionAfterRead: 2,
+				MoveTo:          filepath.Join(path, "ddd"),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := m.Provision(ctx, tt.props)
+			if tt.e == "" {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.c, m.config)
+			} else {
+				assert.Error(t, err)
+				assert.EqualError(t, err, tt.e)
+			}
+		})
+	}
+}
+
+func TestLines(t *testing.T) {
+	path, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	path = filepath.Join(path, "test")
+	func() {
+		src, err := os.Open(filepath.Join(path, "test.lines"))
+		assert.NoError(t, err)
+		defer src.Close()
+		dest, err := os.Create(filepath.Join(path, "test.lines.copy"))
+		assert.NoError(t, err)
+		defer dest.Close()
+		_, err = io.Copy(dest, src)
+		assert.NoError(t, err)
+	}()
+
+	meta := map[string]any{
+		"file": filepath.Join(path, "test.lines.copy"),
+	}
+	mc := timex.Clock
+	exp := []api.MessageTuple{
+		model.NewDefaultRawTuple([]byte("{\"id\": 1,\"name\": \"John Doe\"}"), meta, mc.Now()),
+		model.NewDefaultRawTuple([]byte("{\"id\": 2,\"name\": \"Jane Doe\"}"), meta, mc.Now()),
+		model.NewDefaultRawTuple([]byte("{\"id\": 3,\"name\": \"John Smith\"}"), meta, mc.Now()),
+		model.NewDefaultRawTuple([]byte("[{\"id\": 4,\"name\": \"John Smith\"},{\"id\": 5,\"name\": \"John Smith\"}]"), meta, mc.Now()),
+	}
+	r := GetSource().(api.BytesSource)
+	mock.TestSourceConnector(t, r, map[string]any{
+		"path":            path,
+		"fileType":        "lines",
+		"datasource":      "test.lines.copy",
+		"actionAfterRead": 1,
+	}, exp, func() {
+		// do nothing
+	})
+}
+
+func TestBatch(t *testing.T) {
+	path, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	path = filepath.Join(path, "test", "json")
+
+	meta := map[string]any{
+		"file": filepath.Join(path, "f1.json"),
+	}
+	mc := timex.Clock
+	exp := []api.MessageTuple{
+		model.NewDefaultRawTuple([]byte("[{\"id\": 1,\"name\": \"John Doe\",\"height\": 1.82},{\"id\": 2,\"name\": \"Jane Doe\",\"height\": 1.65}]"), meta, mc.Now()),
+	}
+	r := GetSource().(api.BytesSource)
+	mock.TestSourceConnector(t, r, map[string]any{
+		"path":       path,
+		"fileType":   "json",
+		"datasource": "f1.json",
+	}, exp, func() {
+		// do nothing
+	})
+}
+
+func TestIgnoreLines(t *testing.T) {
+	tmpfile, err := os.CreateTemp("", "test.ignore.lines")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.Remove(tmpfile.Name())
+	if _, err = tmpfile.Write([]byte("{\"id\": 1,\"name\": \"John Doe\"}\n{\"id\": 2,\"name\": \"Jane Doe\"}\n{\"id\": 3,\"name\": \"John Smith\"}\n[{\"id\": 4,\"name\": \"John Smith\"},{\"id\": 5,\"name\": \"John Smith\"}]")); err != nil {
+		t.Fatal(err)
+	}
+	path := filepath.Dir(tmpfile.Name())
+	name := filepath.Base(tmpfile.Name())
+	meta := map[string]any{
+		"file": tmpfile.Name(),
+	}
+	mc := timex.Clock
+	exp := []api.MessageTuple{
+		model.NewDefaultRawTuple([]byte("{\"id\": 2,\"name\": \"Jane Doe\"}"), meta, mc.Now()),
+		model.NewDefaultRawTuple([]byte("{\"id\": 3,\"name\": \"John Smith\"}"), meta, mc.Now()),
+	}
+	r := GetSource().(api.BytesSource)
+	r.(api.Bounded).SetEofIngest(func(ctx api.StreamContext) {
+		fmt.Printf("eof")
+	})
+	moved, err := os.MkdirTemp(path, "test")
+	defer os.RemoveAll(moved)
+	assert.NoError(t, err)
+	mock.TestSourceConnector(t, r, map[string]any{
+		"path":             path,
+		"fileType":         "lines",
+		"datasource":       name,
+		"actionAfterRead":  2,
+		"moveTo":           moved,
+		"ignoreStartLines": 1,
+		"ignoreEndLines":   1,
+	}, exp, func() {
+		// do nothing
+	})
+	// wait for file moved
+	time.Sleep(100 * time.Millisecond)
+	movedFile := filepath.Join(moved, name)
+	_, err = os.Stat(movedFile)
+	if os.IsNotExist(err) {
+		assert.Fail(t, "copy file is not existed")
+	} else {
+		os.Remove(movedFile)
+	}
+}
+
+func TestIntervalAndDir(t *testing.T) {
+	ts0 := timex.GetNow()
+	ts := ts0.Add(2 * time.Second)
+	path, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	path = filepath.Join(path, "test")
+
+	meta := map[string]any{
+		"file": filepath.Join(path, "test.lines"),
+	}
+	exp := []api.MessageTuple{
+		model.NewDefaultRawTuple([]byte("{\"id\": 2,\"name\": \"Jane Doe\"}"), meta, ts),
+		model.NewDefaultRawTuple([]byte("{\"id\": 3,\"name\": \"John Smith\"}"), meta, ts),
+		model.NewDefaultRawTuple([]byte("[{\"id\": 4,\"name\": \"John Smith\"},{\"id\": 5,\"name\": \"John Smith\"}]"), meta, ts),
+		model.NewDefaultRawTuple([]byte("{\"id\": 2,\"name\": \"Jane Doe\"}"), meta, ts),
+	}
+	r := GetSource().(api.BytesSource)
+	mock.TestSourceConnector(t, r, map[string]any{
+		"path":             path,
+		"fileType":         "lines",
+		"interval":         "1s",
+		"sendInterval":     "100ms",
+		"parallel":         true,
+		"ignoreStartLines": 1,
+	}, exp, func() {
+		fmt.Println("before add")
+		timex.Add(2 * time.Second)
+		fmt.Println("after add")
+	})
+}
diff --git a/internal/io/file/test/json/f1.json b/internal/io/file/test/json/f1.json
new file mode 100644
index 00000000..e19dbf41
--- /dev/null
+++ b/internal/io/file/test/json/f1.json
@@ -0,0 +1,1 @@
+[{"id": 1,"name": "John Doe","height": 1.82},{"id": 2,"name": "Jane Doe","height": 1.65}]
\ No newline at end of file
diff --git a/internal/io/file/test/test.lines b/internal/io/file/test/test.lines
new file mode 100644
index 00000000..a283f53e
--- /dev/null
+++ b/internal/io/file/test/test.lines
@@ -0,0 +1,4 @@
+{"id": 1,"name": "John Doe"}
+{"id": 2,"name": "Jane Doe"}
+{"id": 3,"name": "John Smith"}
+[{"id": 4,"name": "John Smith"},{"id": 5,"name": "John Smith"}]
\ No newline at end of file
diff --git a/internal/topo/topotest/rule_test.go b/internal/topo/topotest/rule_test.go
index cd543372..03bda7a5 100644
--- a/internal/topo/topotest/rule_test.go
+++ b/internal/topo/topotest/rule_test.go
@@ -18,7 +18,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
+
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/planner"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mocknode"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 )
@@ -1784,3 +1787,35 @@ func TestAliasSQL(t *testing.T) {
 		DoRuleTest(t, tests, opt, 0)
 	}
 }
+
+func TestRuleWaitGroup(t *testing.T) {
+	streamList := []string{"demo"}
+	HandleStream(false, streamList, t)
+	HandleStream(true, streamList, t)
+	id := "rule991"
+	sql := "select color,size from demo"
+	rule := &def.Rule{
+		Id:  id,
+		Sql: sql,
+		Actions: []map[string]any{
+			{
+				"memory": map[string]any{
+					"topic":      id,
+					"sendSingle": false,
+				},
+			},
+		},
+		Options: &def.RuleOption{
+			BufferLength:       100,
+			SendError:          true,
+			Qos:                def.AtMostOnce,
+			CheckpointInterval: cast.DurationConf(time.Second * 5),
+		},
+	}
+	tp, err := planner.Plan(rule)
+	require.NoError(t, err)
+	tp.Open()
+	time.Sleep(10 * time.Millisecond)
+	tp.Cancel()
+	tp.WaitClose()
+}
diff --git a/pkg/mock/test_source.go b/pkg/mock/test_source.go
index 34e296a7..1b22feb7 100644
--- a/pkg/mock/test_source.go
+++ b/pkg/mock/test_source.go
@@ -30,7 +30,7 @@ import (
 
 var count atomic.Value
 
-func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expected []api.MessageTuple, sender func()) {
+func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expected any, sender func()) {
 	// init
 	c := count.Load()
 	if c == nil {
@@ -45,9 +45,21 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 	// connect, subscribe and read data
 	err = r.Connect(ctx)
 	assert.NoError(t, err)
-
+	// Send data
+	go func() {
+		sender()
+	}()
+	time.Sleep(10 * time.Millisecond)
 	// Send and receive data
-	limit := len(expected)
+	limit := 0
+	switch et := expected.(type) {
+	case []api.MessageTuple:
+		limit = len(et)
+	case []api.RawTuple:
+		limit = len(et)
+	default:
+		t.Fatal("invalid expected type")
+	}
 	var (
 		wg     sync.WaitGroup
 		result []api.MessageTuple
@@ -59,7 +71,7 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 			err = ss.Subscribe(ctx, func(ctx api.StreamContext, payload []byte, meta map[string]any, ts time.Time) {
 				result = append(result, model.NewDefaultRawTuple(payload, meta, ts))
 				limit--
-				if limit <= 0 {
+				if limit == 0 {
 					wg.Done()
 				}
 			}, func(ctx api.StreamContext, err error) {
@@ -75,10 +87,6 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 		err = r.Close(ctx)
 		assert.NoError(t, err)
 	}()
-	// Send data
-	go func() {
-		sender()
-	}()
 
 	ticker := time.After(60 * time.Second)
 	finished := make(chan struct{})