Repository: lf-edge/ekuiper
Pull request #2839 (2024-05-10 06:08:38)

fix: revise interval config


Patch:
diff --git a/deploy/chart/ekuiper/values.yaml b/deploy/chart/ekuiper/values.yaml
index 64213321..60a78e2b 100644
--- a/deploy/chart/ekuiper/values.yaml
+++ b/deploy/chart/ekuiper/values.yaml
@@ -159,8 +159,8 @@ kuiperConfig:
       # restored for unintended interrupt or planned restart of the rule. The performance may be affected
       # to enable the checkpoint mechanism
       qos: 0
-      # The interval in millisecond to run the checkpoint mechanism.
-      checkpointInterval: 300000
+      # The interval duration to run the checkpoint mechanism.
+      checkpointInterval: 300s
       # Whether to send errors to sinks
       sendError: true
 
@@ -177,8 +177,8 @@ kuiperConfig:
       # The number of messages for a buffer page which is the unit to read/write to disk batchly to prevent frequent IO
       bufferPageSize: 256
 
-      # The interval in millisecond to resend the cached messages
-      resendInterval: 0
+      # The interval duration to resend the cached messages
+      resendInterval: 0s
 
       # Whether to clean the cache when the rule stops
       cleanCacheAtStop: false
diff --git a/etc/kuiper.yaml b/etc/kuiper.yaml
index a0a057b3..e3a80491 100644
--- a/etc/kuiper.yaml
+++ b/etc/kuiper.yaml
@@ -71,8 +71,8 @@ rule:
   # restored for unintended interrupt or planned restart of the rule. The performance may be affected
   # to enable the checkpoint mechanism
   qos: 0
-  # The interval in millisecond to run the checkpoint mechanism.
-  checkpointInterval: 300000
+  # The interval duration to run the checkpoint mechanism.
+  checkpointInterval: 300s
   # Whether to send errors to sinks
   sendError: true
   # The strategy to retry for rule errors.
@@ -100,8 +100,8 @@ sink:
   # The number of messages for a buffer page which is the unit to read/write to disk in batch to prevent frequent IO
   bufferPageSize: 256
 
-  # The interval in millisecond to resend the cached messages
-  resendInterval: 0
+  # The interval duration to resend the cached messages
+  resendInterval: 0s
 
   # Whether to clean the cache when the rule stops
   cleanCacheAtStop: false
diff --git a/internal/conf/conf.go b/internal/conf/conf.go
index bfc071b6..3f7dddb9 100644
--- a/internal/conf/conf.go
+++ b/internal/conf/conf.go
@@ -62,7 +62,7 @@ type SinkConf struct {
 	MaxDiskCache         int    `json:"maxDiskCache" yaml:"maxDiskCache"`
 	BufferPageSize       int    `json:"bufferPageSize" yaml:"bufferPageSize"`
 	EnableCache          bool   `json:"enableCache" yaml:"enableCache"`
-	ResendInterval       int    `json:"resendInterval" yaml:"resendInterval"`
+	ResendInterval       string `json:"resendInterval" yaml:"resendInterval"`
 	CleanCacheAtStop     bool   `json:"cleanCacheAtStop" yaml:"cleanCacheAtStop"`
 	ResendAlterQueue     bool   `json:"resendAlterQueue" yaml:"resendAlterQueue"`
 	ResendPriority       int    `json:"resendPriority" yaml:"resendPriority"`
@@ -87,11 +87,14 @@ func (sc *SinkConf) Validate() error {
 		Log.Warnf("bufferPageSize is less than or equal to 0, set to 256")
 		errs = errors.Join(errs, errors.New("bufferPageSize:bufferPageSize must be positive"))
 	}
-	if sc.ResendInterval < 0 {
-		sc.ResendInterval = 0
-		Log.Warnf("resendInterval is less than 0, set to 0")
+	d, err := cast.ConvertDuration(sc.ResendInterval)
+	if err != nil {
+		errs = errors.Join(errs, err)
+	}
+	if d < 0 {
 		errs = errors.Join(errs, errors.New("resendInterval:resendInterval must be positive"))
 	}
+
 	if sc.BufferPageSize > sc.MemoryCacheThreshold {
 		sc.MemoryCacheThreshold = sc.BufferPageSize
 		Log.Warnf("memoryCacheThreshold is less than bufferPageSize, set to %d", sc.BufferPageSize)
@@ -302,7 +305,7 @@ func InitConf() {
 			LateTol:            1000,
 			Concurrency:        1,
 			BufferLength:       1024,
-			CheckpointInterval: 300000, // 5 minutes
+			CheckpointInterval: "300s", // 5 minutes
 			SendError:          true,
 			Restart: &def.RestartStrategy{
 				Attempts:     0,
@@ -414,10 +417,9 @@ func SetLogFormat(disableTimestamp bool) {
 
 func ValidateRuleOption(option *def.RuleOption) error {
 	var errs error
-	if option.CheckpointInterval < 0 {
-		option.CheckpointInterval = 0
-		Log.Warnf("checkpointInterval is negative, set to 0")
-		errs = errors.Join(errs, errors.New("invalidCheckpointInterval:checkpointInterval must be greater than 0"))
+	_, err := cast.ConvertDuration(option.CheckpointInterval)
+	if err != nil {
+		errs = errors.Join(errs, err)
 	}
 	if option.Concurrency < 0 {
 		option.Concurrency = 1
diff --git a/internal/pkg/def/rule.go b/internal/pkg/def/rule.go
index f265df20..59e51186 100644
--- a/internal/pkg/def/rule.go
+++ b/internal/pkg/def/rule.go
@@ -24,7 +24,7 @@ type RuleOption struct {
 	SendMetaToSink     bool             `json:"sendMetaToSink" yaml:"sendMetaToSink"`
 	SendError          bool             `json:"sendError" yaml:"sendError"`
 	Qos                Qos              `json:"qos" yaml:"qos"`
-	CheckpointInterval int              `json:"checkpointInterval" yaml:"checkpointInterval"`
+	CheckpointInterval string           `json:"checkpointInterval" yaml:"checkpointInterval"`
 	Restart            *RestartStrategy `json:"restartStrategy" yaml:"restartStrategy"`
 	Cron               string           `json:"cron" yaml:"cron"`
 	Duration           string           `json:"duration" yaml:"duration"`
@@ -109,7 +109,7 @@ func GetDefaultRule(name, sql string) *Rule {
 			SendMetaToSink:     false,
 			SendError:          true,
 			Qos:                AtMostOnce,
-			CheckpointInterval: 300000,
+			CheckpointInterval: "300s",
 			Restart: &RestartStrategy{
 				Attempts:     0,
 				Delay:        1000,
diff --git a/internal/topo/node/cache_op.go b/internal/topo/node/cache_op.go
index faa1c1d0..8fca6013 100644
--- a/internal/topo/node/cache_op.go
+++ b/internal/topo/node/cache_op.go
@@ -25,6 +25,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/checkpoint"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/node/cache"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
@@ -166,7 +167,8 @@ func (s *CacheOp) doBroadcast(val interface{}) {
 		if !s.hasCache {
 			s.ctx.GetLogger().Debugf("memory buffer full, start to save cache")
 			// Start the send interval
-			s.resendTicker = timex.GetTicker(int64(s.cacheConf.ResendInterval))
+			d, _ := cast.ConvertDuration(s.cacheConf.ResendInterval)
+			s.resendTicker = timex.Clock.Ticker(d)
 			s.resendTimerCh = s.resendTicker.C
 			s.hasCache = true
 		}
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index 18a704d3..830393e2 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -35,6 +35,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/topo/node"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/node/metric"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/state"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
@@ -214,7 +215,9 @@ func (s *Topo) Open() <-chan error {
 		if s.store, err = state.CreateStore(s.name, s.options.Qos); err != nil {
 			return fmt.Errorf("topo %s create store error %v", s.name, err)
 		}
-		s.enableCheckpoint(s.ctx)
+		if err := s.enableCheckpoint(s.ctx); err != nil {
+			return err
+		}
 		// open stream sink, after log sink is ready.
 		for _, snk := range s.sinks {
 			snk.Exec(s.ctx.WithMeta(s.name, snk.GetName(), s.store), s.drain)
@@ -244,7 +247,7 @@ func (s *Topo) HasOpen() bool {
 	return s.hasOpened.Load()
 }
 
-func (s *Topo) enableCheckpoint(ctx api.StreamContext) {
+func (s *Topo) enableCheckpoint(ctx api.StreamContext) error {
 	if s.options.Qos >= def.AtLeastOnce {
 		var (
 			sources []checkpoint.StreamTask
@@ -267,9 +270,14 @@ func (s *Topo) enableCheckpoint(ctx api.StreamContext) {
 		for _, r := range s.sinks {
 			sinks = append(sinks, r)
 		}
-		c := checkpoint.NewCoordinator(s.name, sources, ops, sinks, s.options.Qos, s.store, s.options.CheckpointInterval, s.ctx)
+		d, err := cast.ConvertDuration(s.options.CheckpointInterval)
+		if err != nil {
+			return err
+		}
+		c := checkpoint.NewCoordinator(s.name, sources, ops, sinks, s.options.Qos, s.store, int(d/time.Millisecond), s.ctx)
 		s.coordinator = c
 	}
+	return nil
 }
 
 func (s *Topo) GetCoordinator() *checkpoint.Coordinator {
diff --git a/pkg/cast/time.go b/pkg/cast/time.go
index 84a869b0..74d7ad08 100644
--- a/pkg/cast/time.go
+++ b/pkg/cast/time.go
@@ -16,6 +16,7 @@ package cast
 
 import (
 	"fmt"
+	"strconv"
 	"time"
 
 	"github.com/jinzhu/now"
@@ -364,3 +365,11 @@ func InterfaceToDuration(i interface{}) (time.Duration, error) {
 	}
 	return time.ParseDuration(duration)
 }
+
+func ConvertDuration(s string) (time.Duration, error) {
+	i, err := strconv.ParseInt(s, 10, 64)
+	if err == nil {
+		return time.Duration(i) * time.Millisecond, nil
+	}
+	return time.ParseDuration(s)
+}


Test patch:
diff --git a/internal/conf/conf_test.go b/internal/conf/conf_test.go
index b300f9e2..6ceb89b5 100644
--- a/internal/conf/conf_test.go
+++ b/internal/conf/conf_test.go
@@ -92,15 +92,19 @@ func TestRuleOptionValidate(t *testing.T) {
 		err string
 	}{
 		{
-			s: &def.RuleOption{},
-			e: &def.RuleOption{},
+			s: &def.RuleOption{
+				CheckpointInterval: "300s", // 5 minutes
+			},
+			e: &def.RuleOption{
+				CheckpointInterval: "300s", // 5 minutes
+			},
 		},
 		{
 			s: &def.RuleOption{
 				LateTol:            1000,
 				Concurrency:        1,
 				BufferLength:       1024,
-				CheckpointInterval: 300000, // 5 minutes
+				CheckpointInterval: "300s", // 5 minutes
 				SendError:          true,
 				Restart: &def.RestartStrategy{
 					Attempts:     0,
@@ -114,7 +118,7 @@ func TestRuleOptionValidate(t *testing.T) {
 				LateTol:            1000,
 				Concurrency:        1,
 				BufferLength:       1024,
-				CheckpointInterval: 300000, // 5 minutes
+				CheckpointInterval: "300s", // 5 minutes
 				SendError:          true,
 				Restart: &def.RestartStrategy{
 					Attempts:     0,
@@ -130,7 +134,7 @@ func TestRuleOptionValidate(t *testing.T) {
 				LateTol:            1000,
 				Concurrency:        1,
 				BufferLength:       1024,
-				CheckpointInterval: 300000, // 5 minutes
+				CheckpointInterval: "300s", // 5 minutes
 				SendError:          true,
 				Restart: &def.RestartStrategy{
 					Attempts:     3,
@@ -144,7 +148,7 @@ func TestRuleOptionValidate(t *testing.T) {
 				LateTol:            1000,
 				Concurrency:        1,
 				BufferLength:       1024,
-				CheckpointInterval: 300000, // 5 minutes
+				CheckpointInterval: "300s", // 5 minutes
 				SendError:          true,
 				Restart: &def.RestartStrategy{
 					Attempts:     3,
@@ -160,7 +164,7 @@ func TestRuleOptionValidate(t *testing.T) {
 				LateTol:            1000,
 				Concurrency:        1,
 				BufferLength:       1024,
-				CheckpointInterval: 300000, // 5 minutes
+				CheckpointInterval: "300s", // 5 minutes
 				SendError:          true,
 				Restart: &def.RestartStrategy{
 					Attempts:     3,
@@ -174,7 +178,7 @@ func TestRuleOptionValidate(t *testing.T) {
 				LateTol:            1000,
 				Concurrency:        1,
 				BufferLength:       1024,
-				CheckpointInterval: 300000, // 5 minutes
+				CheckpointInterval: "300s", // 5 minutes
 				SendError:          true,
 				Restart: &def.RestartStrategy{
 					Attempts:     3,
@@ -190,7 +194,7 @@ func TestRuleOptionValidate(t *testing.T) {
 				LateTol:            1000,
 				Concurrency:        1,
 				BufferLength:       1024,
-				CheckpointInterval: 300000, // 5 minutes
+				CheckpointInterval: "300s", // 5 minutes
 				SendError:          true,
 				Restart: &def.RestartStrategy{
 					Attempts:     -2,
@@ -204,7 +208,7 @@ func TestRuleOptionValidate(t *testing.T) {
 				LateTol:            1000,
 				Concurrency:        1,
 				BufferLength:       1024,
-				CheckpointInterval: 300000, // 5 minutes
+				CheckpointInterval: "300s", // 5 minutes
 				SendError:          true,
 				Restart: &def.RestartStrategy{
 					Attempts:     0,
@@ -242,7 +246,7 @@ func TestSinkConf_Validate(t *testing.T) {
 				MaxDiskCache:         1024000,
 				BufferPageSize:       256,
 				EnableCache:          true,
-				ResendInterval:       0,
+				ResendInterval:       "0s",
 				CleanCacheAtStop:     true,
 				ResendAlterQueue:     true,
 				ResendPriority:       0,
@@ -256,7 +260,7 @@ func TestSinkConf_Validate(t *testing.T) {
 				MaxDiskCache:         1024000,
 				BufferPageSize:       256,
 				EnableCache:          true,
-				ResendInterval:       0,
+				ResendInterval:       "0s",
 				CleanCacheAtStop:     true,
 				ResendAlterQueue:     true,
 				ResendPriority:       0,
@@ -270,7 +274,7 @@ func TestSinkConf_Validate(t *testing.T) {
 				MaxDiskCache:         -1,
 				BufferPageSize:       256,
 				EnableCache:          true,
-				ResendInterval:       0,
+				ResendInterval:       "0s",
 				CleanCacheAtStop:     true,
 				ResendAlterQueue:     true,
 				ResendPriority:       0,
@@ -284,7 +288,7 @@ func TestSinkConf_Validate(t *testing.T) {
 				MaxDiskCache:         1024000,
 				BufferPageSize:       0,
 				EnableCache:          true,
-				ResendInterval:       0,
+				ResendInterval:       "0s",
 				CleanCacheAtStop:     true,
 				ResendAlterQueue:     true,
 				ResendPriority:       0,
@@ -298,7 +302,7 @@ func TestSinkConf_Validate(t *testing.T) {
 				MaxDiskCache:         1024000,
 				BufferPageSize:       256,
 				EnableCache:          true,
-				ResendInterval:       -1,
+				ResendInterval:       "-1s",
 				CleanCacheAtStop:     true,
 				ResendAlterQueue:     true,
 				ResendPriority:       0,
@@ -312,7 +316,7 @@ func TestSinkConf_Validate(t *testing.T) {
 				MaxDiskCache:         1024000,
 				BufferPageSize:       256,
 				EnableCache:          true,
-				ResendInterval:       0,
+				ResendInterval:       "0s",
 				CleanCacheAtStop:     true,
 				ResendAlterQueue:     true,
 				ResendPriority:       0,
@@ -326,7 +330,7 @@ func TestSinkConf_Validate(t *testing.T) {
 				MaxDiskCache:         1024000,
 				BufferPageSize:       256,
 				EnableCache:          true,
-				ResendInterval:       0,
+				ResendInterval:       "0s",
 				CleanCacheAtStop:     true,
 				ResendAlterQueue:     true,
 				ResendPriority:       0,
@@ -340,7 +344,7 @@ func TestSinkConf_Validate(t *testing.T) {
 				MaxDiskCache:         128,
 				BufferPageSize:       256,
 				EnableCache:          true,
-				ResendInterval:       0,
+				ResendInterval:       "0s",
 				CleanCacheAtStop:     true,
 				ResendAlterQueue:     true,
 				ResendPriority:       0,
@@ -354,7 +358,7 @@ func TestSinkConf_Validate(t *testing.T) {
 				MaxDiskCache:         300,
 				BufferPageSize:       256,
 				EnableCache:          true,
-				ResendInterval:       0,
+				ResendInterval:       "0s",
 				CleanCacheAtStop:     true,
 				ResendAlterQueue:     true,
 				ResendPriority:       0,
@@ -368,7 +372,7 @@ func TestSinkConf_Validate(t *testing.T) {
 				MaxDiskCache:         1024000,
 				BufferPageSize:       256,
 				EnableCache:          true,
-				ResendInterval:       0,
+				ResendInterval:       "0s",
 				CleanCacheAtStop:     true,
 				ResendAlterQueue:     true,
 				ResendPriority:       2,
diff --git a/internal/processor/rule_test.go b/internal/processor/rule_test.go
index 7bbee194..4642db90 100644
--- a/internal/processor/rule_test.go
+++ b/internal/processor/rule_test.go
@@ -88,7 +88,7 @@ func TestRuleActionParse_Apply(t *testing.T) {
 					BufferLength:       1024,
 					SendMetaToSink:     false,
 					Qos:                def.AtMostOnce,
-					CheckpointInterval: 300000,
+					CheckpointInterval: "300s",
 					SendError:          true,
 					Restart: &def.RestartStrategy{
 						Attempts:     20,
@@ -123,7 +123,7 @@ func TestRuleActionParse_Apply(t *testing.T) {
 					"lateTolerance": 1000,
 					"bufferLength": 10240,
 					"qos": 2,
-					"checkpointInterval": 60000
+					"checkpointInterval": "60s"
 				}
 			}`,
 			result: &def.Rule{
@@ -151,7 +151,7 @@ func TestRuleActionParse_Apply(t *testing.T) {
 					BufferLength:       10240,
 					SendMetaToSink:     false,
 					Qos:                def.ExactlyOnce,
-					CheckpointInterval: 60000,
+					CheckpointInterval: "60s",
 					SendError:          true,
 					Restart: &def.RestartStrategy{
 						Attempts:     0,
@@ -187,7 +187,7 @@ func TestRuleActionParse_Apply(t *testing.T) {
 					BufferLength:       1024,
 					SendMetaToSink:     false,
 					Qos:                def.AtMostOnce,
-					CheckpointInterval: 300000,
+					CheckpointInterval: "300s",
 					SendError:          true,
 					Restart: &def.RestartStrategy{
 						Attempts:     0,
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 1d5a6dac..ca492222 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -365,7 +365,7 @@ func (suite *RestTestSuite) Test_rulesManageHandler() {
 	suite.r.ServeHTTP(w1, req1)
 
 	returnVal, _ = io.ReadAll(w1.Result().Body)
-	expect = `{"triggered":true,"id":"rule321","sql":"select * from alert","actions":[{"nop":{}}],"options":{"debug":false,"logFilename":"","isEventTime":false,"lateTolerance":1000,"concurrency":1,"bufferLength":1024,"sendMetaToSink":false,"sendError":true,"qos":0,"checkpointInterval":300000,"restartStrategy":{"attempts":0,"delay":1000,"multiplier":2,"maxDelay":30000,"jitter":0.1},"cron":"","duration":"","cronDatetimeRange":null}}`
+	expect = `{"triggered":true,"id":"rule321","sql":"select * from alert","actions":[{"nop":{}}],"options":{"debug":false,"logFilename":"","isEventTime":false,"lateTolerance":1000,"concurrency":1,"bufferLength":1024,"sendMetaToSink":false,"sendError":true,"qos":0,"checkpointInterval":"300s","restartStrategy":{"attempts":0,"delay":1000,"multiplier":2,"maxDelay":30000,"jitter":0.1},"cron":"","duration":"","cronDatetimeRange":null}}`
 	assert.Equal(suite.T(), expect, string(returnVal))
 
 	// delete rule
diff --git a/internal/topo/node/cache/sync_cache_test.go b/internal/topo/node/cache/sync_cache_test.go
index 1c8f10b0..8785e19b 100644
--- a/internal/topo/node/cache/sync_cache_test.go
+++ b/internal/topo/node/cache/sync_cache_test.go
@@ -124,7 +124,7 @@ func TestCache(t *testing.T) {
 		MaxDiskCache:     6,
 		BufferPageSize:   2,
 		EnableCache:      true,
-		ResendInterval:   0,
+		ResendInterval:   "0s",
 		CleanCacheAtStop: false,
 	})
 	assert.NoError(t, err)
@@ -229,7 +229,7 @@ func TestCacheCase2(t *testing.T) {
 		MaxDiskCache:         4,
 		BufferPageSize:       2,
 		EnableCache:          true,
-		ResendInterval:       10,
+		ResendInterval:       "10",
 	})
 	assert.NoError(t, err)
 	// prepare data
@@ -331,7 +331,7 @@ func TestCacheInit(t *testing.T) {
 		MaxDiskCache:         4,
 		BufferPageSize:       2,
 		EnableCache:          true,
-		ResendInterval:       0,
+		ResendInterval:       "0ms",
 		CleanCacheAtStop:     false,
 	})
 	assert.NoError(t, err)
@@ -355,7 +355,7 @@ func TestCacheInit(t *testing.T) {
 		MaxDiskCache:         4,
 		BufferPageSize:       2,
 		EnableCache:          true,
-		ResendInterval:       0,
+		ResendInterval:       "0ms",
 		CleanCacheAtStop:     false,
 	})
 	assert.NoError(t, err)
@@ -374,7 +374,7 @@ func TestCacheInit(t *testing.T) {
 		MaxDiskCache:         4,
 		BufferPageSize:       2,
 		EnableCache:          true,
-		ResendInterval:       0,
+		ResendInterval:       "0ms",
 		CleanCacheAtStop:     false,
 	})
 	assert.NoError(t, err)
@@ -394,7 +394,7 @@ func TestCacheInit(t *testing.T) {
 		MaxDiskCache:         2,
 		BufferPageSize:       2,
 		EnableCache:          true,
-		ResendInterval:       0,
+		ResendInterval:       "0ms",
 		CleanCacheAtStop:     true,
 	})
 	assert.NoError(t, err)
diff --git a/internal/topo/node/cache_op_test.go b/internal/topo/node/cache_op_test.go
index 823b445c..ae881692 100644
--- a/internal/topo/node/cache_op_test.go
+++ b/internal/topo/node/cache_op_test.go
@@ -108,7 +108,7 @@ func TestCacheRun(t *testing.T) {
 		MaxDiskCache:         4,
 		BufferPageSize:       2,
 		EnableCache:          true,
-		ResendInterval:       10,
+		ResendInterval:       "10",
 	})
 	assert.NoError(t, err)
 	// In sink_node planner, set this buffer length
@@ -159,7 +159,7 @@ func TestRunError(t *testing.T) {
 		MaxDiskCache:         4,
 		BufferPageSize:       2,
 		EnableCache:          true,
-		ResendInterval:       10,
+		ResendInterval:       "10",
 	})
 	assert.NoError(t, err)
 	err = op.AddOutput(make(chan any, 2), "output1")
diff --git a/internal/topo/node/watermark_op_test.go b/internal/topo/node/watermark_op_test.go
index 2a9c57cf..9df50d22 100644
--- a/internal/topo/node/watermark_op_test.go
+++ b/internal/topo/node/watermark_op_test.go
@@ -208,7 +208,7 @@ func TestSingleStreamWatermark(t *testing.T) {
 				SendMetaToSink:     false,
 				SendError:          false,
 				Qos:                0,
-				CheckpointInterval: 0,
+				CheckpointInterval: "0s",
 			})
 			errCh := make(chan error)
 			outputCh := make(chan interface{}, 50)
@@ -460,7 +460,7 @@ func TestMultiStreamWatermark(t *testing.T) {
 				SendMetaToSink:     false,
 				SendError:          false,
 				Qos:                0,
-				CheckpointInterval: 0,
+				CheckpointInterval: "0s",
 			})
 			errCh := make(chan error)
 			outputCh := make(chan interface{}, 50)
diff --git a/internal/topo/planner/analyzer_test.go b/internal/topo/planner/analyzer_test.go
index 9261a97b..2cc072b4 100644
--- a/internal/topo/planner/analyzer_test.go
+++ b/internal/topo/planner/analyzer_test.go
@@ -195,7 +195,7 @@ func TestCheckTopoSort(t *testing.T) {
 		BufferLength:       0,
 		SendMetaToSink:     false,
 		Qos:                0,
-		CheckpointInterval: 0,
+		CheckpointInterval: "0s",
 		SendError:          true,
 	}, store)
 	errWithCode, ok := err.(errorx.ErrorWithCode)
@@ -258,7 +258,7 @@ func Test_validation(t *testing.T) {
 			BufferLength:       0,
 			SendMetaToSink:     false,
 			Qos:                0,
-			CheckpointInterval: 0,
+			CheckpointInterval: "0s",
 			SendError:          true,
 		}, store)
 		assert.Equal(t, tt.r.err, testx.Errstring(err), tt.sql)
@@ -314,7 +314,7 @@ func Test_validationSchemaless(t *testing.T) {
 			BufferLength:       0,
 			SendMetaToSink:     false,
 			Qos:                0,
-			CheckpointInterval: 0,
+			CheckpointInterval: "0s",
 			SendError:          true,
 		}, store)
 		serr := tt.r.Serr()
diff --git a/internal/topo/planner/planner_alias_test.go b/internal/topo/planner/planner_alias_test.go
index 2e004aa2..52dca4f8 100644
--- a/internal/topo/planner/planner_alias_test.go
+++ b/internal/topo/planner/planner_alias_test.go
@@ -251,7 +251,7 @@ func TestPlannerAlias(t *testing.T) {
 			BufferLength:       0,
 			SendMetaToSink:     false,
 			Qos:                0,
-			CheckpointInterval: 0,
+			CheckpointInterval: "0s",
 			SendError:          true,
 		}, kv)
 		if !reflect.DeepEqual(tt.p, p) {
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index 824004eb..f704c475 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -51,7 +51,7 @@ var defaultOption = &def.RuleOption{
 	SendMetaToSink:     false,
 	SendError:          true,
 	Qos:                def.AtMostOnce,
-	CheckpointInterval: 300000,
+	CheckpointInterval: "300s",
 	Restart: &def.RestartStrategy{
 		Attempts:     0,
 		Delay:        1000,
@@ -2753,7 +2753,7 @@ func Test_createLogicalPlan(t *testing.T) {
 			BufferLength:       0,
 			SendMetaToSink:     false,
 			Qos:                0,
-			CheckpointInterval: 0,
+			CheckpointInterval: "0s",
 			SendError:          true,
 		}, kv)
 		if !reflect.DeepEqual(tt.err, testx.Errstring(err)) {
@@ -4156,7 +4156,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 			BufferLength:       0,
 			SendMetaToSink:     false,
 			Qos:                0,
-			CheckpointInterval: 0,
+			CheckpointInterval: "0s",
 			SendError:          true,
 		}, kv)
 		if !reflect.DeepEqual(tt.err, testx.Errstring(err)) {
@@ -4744,7 +4744,7 @@ func Test_createLogicalPlan4Lookup(t *testing.T) {
 				BufferLength:       0,
 				SendMetaToSink:     false,
 				Qos:                0,
-				CheckpointInterval: 0,
+				CheckpointInterval: "0s",
 				SendError:          true,
 			}, kv)
 			assert.Equal(t, tt.err, testx.Errstring(err))
diff --git a/internal/topo/rule/ruleState_test.go b/internal/topo/rule/ruleState_test.go
index 0587d3d0..85ba56aa 100644
--- a/internal/topo/rule/ruleState_test.go
+++ b/internal/topo/rule/ruleState_test.go
@@ -40,7 +40,7 @@ var defaultOption = &def.RuleOption{
 	SendMetaToSink:     false,
 	SendError:          true,
 	Qos:                def.AtMostOnce,
-	CheckpointInterval: 300000,
+	CheckpointInterval: "300s",
 	Restart: &def.RestartStrategy{
 		Attempts:     0,
 		Delay:        1000,
diff --git a/internal/topo/topotest/rule_test.go b/internal/topo/topotest/rule_test.go
index 762ee2af..6b9fda45 100644
--- a/internal/topo/topotest/rule_test.go
+++ b/internal/topo/topotest/rule_test.go
@@ -144,13 +144,13 @@ func TestWindowFuncSQL(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.AtLeastOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 		{
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.ExactlyOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 	}
 	for _, opt := range options {
@@ -391,13 +391,13 @@ func TestAccAggSQL(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.AtLeastOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 		{
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.ExactlyOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 	}
 	for _, opt := range options {
@@ -510,12 +510,12 @@ func TestSRFSQL(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.AtLeastOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		}, {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.ExactlyOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 	}
 	for _, opt := range options {
@@ -1169,13 +1169,13 @@ func TestSingleSQL(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.AtLeastOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 		{
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.ExactlyOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 	}
 	for _, opt := range options {
@@ -1310,14 +1310,14 @@ func TestSingleSQLWithEventTime(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.AtLeastOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 			IsEventTime:        true,
 			LateTol:            1000,
 		}, {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.ExactlyOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 			IsEventTime:        true,
 			LateTol:            1000,
 		},
@@ -1556,12 +1556,12 @@ func TestSingleSQLForBinary(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.AtLeastOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		}, {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.ExactlyOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 	}
 	for _, opt := range options {
@@ -1629,14 +1629,14 @@ func TestWindowSQL(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.AtLeastOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 			IsEventTime:        true,
 		},
 		{
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.ExactlyOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 			IsEventTime:        true,
 		},
 	}
@@ -1765,13 +1765,13 @@ func TestAliasSQL(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.AtLeastOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 		{
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.ExactlyOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 	}
 	for _, opt := range options {
diff --git a/internal/topo/topotest/window_rule_test.go b/internal/topo/topotest/window_rule_test.go
index 7bebcc88..0586f5ca 100644
--- a/internal/topo/topotest/window_rule_test.go
+++ b/internal/topo/topotest/window_rule_test.go
@@ -648,13 +648,13 @@ func TestWindow(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.AtLeastOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 		{
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.ExactlyOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 		},
 	}
 	for _, opt := range options {
@@ -1285,7 +1285,7 @@ func TestEventWindow(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.AtLeastOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 			IsEventTime:        true,
 			LateTol:            1000,
 		},
@@ -1293,7 +1293,7 @@ func TestEventWindow(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.ExactlyOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 			IsEventTime:        true,
 			LateTol:            1000,
 		},
@@ -1510,7 +1510,7 @@ func TestEventSlidingWindow(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.AtLeastOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 			IsEventTime:        true,
 			LateTol:            1000,
 		},
@@ -1518,7 +1518,7 @@ func TestEventSlidingWindow(t *testing.T) {
 			BufferLength:       100,
 			SendError:          true,
 			Qos:                def.ExactlyOnce,
-			CheckpointInterval: 5000,
+			CheckpointInterval: "5s",
 			IsEventTime:        true,
 			LateTol:            1000,
 		},
diff --git a/pkg/cast/time_test.go b/pkg/cast/time_test.go
index 8971969b..aca2743c 100644
--- a/pkg/cast/time_test.go
+++ b/pkg/cast/time_test.go
@@ -271,3 +271,13 @@ func TestInterfaceToUnixMilli(t *testing.T) {
 		assert.Equal(t, tt.want, got)
 	}
 }
+
+func TestConvertDuration(t *testing.T) {
+	t1, err := ConvertDuration("100")
+	require.NoError(t, err)
+	require.Equal(t, 100*time.Millisecond, t1)
+
+	t2, err := ConvertDuration("100s")
+	require.NoError(t, err)
+	require.Equal(t, 100*time.Second, t2)
+}