Repository: lf-edge/ekuiper
Pull request #2855 (2024-05-17 04:11:25)

feat(cache): resend alter queue support

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2855?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 `67.28972%` with `35 lines` in your changes are missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@f89aa77`). [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/2855?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/topo/node/sink\_node.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2855?src=pr&el=tree&filepath=internal%2Ftopo%2Fnode%2Fsink_node.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ub2RlL3Npbmtfbm9kZS5nbw==) | 59.09% | [17 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2855?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/topo/planner/planner\_sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2855?src=pr&el=tree&filepath=internal%2Ftopo%2Fplanner%2Fplanner_sink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9wbGFubmVyL3BsYW5uZXJfc2luay5nbw==) | 55.56% | [8 Missing and 4 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2855?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/mqtt/connection.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2855?src=pr&el=tree&filepath=internal%2Fio%2Fmqtt%2Fconnection.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vbXF0dC9jb25uZWN0aW9uLmdv) | 58.33% | [2 Missing and 3 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2855?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    #2855   +/-   ##
============================================
  Coverage             ?   68.66%           
============================================
  Files                ?      272           
  Lines                ?    25783           
  Branches             ?        0           
============================================
  Hits                 ?    17703           
  Misses               ?     6493           
  Partials             ?     1587           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2855?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/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..5f30fc94 100644
--- a/internal/conf/conf.go
+++ b/internal/conf/conf.go
@@ -62,11 +62,12 @@ 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"`
 	ResendIndicatorField string `json:"resendIndicatorField" yaml:"resendIndicatorField"`
+	ResendDestination    string `json:"resendDestination" yaml:"resendDestination"`
 }
 
 // Validate the configuration and reset to the default value for invalid values.
@@ -87,11 +88,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 +306,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 +418,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/io/mqtt/connection.go b/internal/io/mqtt/connection.go
index 3e6e0cad..cfd12bf0 100644
--- a/internal/io/mqtt/connection.go
+++ b/internal/io/mqtt/connection.go
@@ -57,6 +57,14 @@ type SubscriptionInfo struct {
 	ErrHandler func(error)
 }
 
+func (conn *Connection) Publish(topic string, qos byte, retained bool, payload any) error {
+	if !conn.connected.Load() {
+		return errorx.NewIOErr("mqtt client is not connected")
+	}
+	token := conn.Client.Publish(topic, qos, retained, payload)
+	return handleToken(token)
+}
+
 func (conn *Connection) onConnect(_ pahoMqtt.Client) {
 	conn.connected.Store(true)
 	conn.logger.Infof("The connection to mqtt broker is established")
@@ -142,7 +150,7 @@ func CreateClient(ctx api.StreamContext, selId string, props map[string]any) (*C
 	if err != nil {
 		return nil, err
 	}
-	opts := pahoMqtt.NewClientOptions().AddBroker(c.Server).SetProtocolVersion(c.pversion)
+	opts := pahoMqtt.NewClientOptions().AddBroker(c.Server).SetProtocolVersion(c.pversion).SetAutoReconnect(true).SetMaxReconnectInterval(time.Minute)
 
 	opts = opts.SetTLSConfig(c.tls)
 
@@ -176,6 +184,15 @@ func CreateClient(ctx api.StreamContext, selId string, props map[string]any) (*C
 	return con, nil
 }
 
+func handleToken(token pahoMqtt.Token) error {
+	if !token.WaitTimeout(5 * time.Second) {
+		return errorx.NewIOErr("timeout")
+	} else if token.Error() != nil {
+		return errorx.NewIOErr(token.Error().Error())
+	}
+	return nil
+}
+
 func validateConfig(props map[string]any) (*ConnectionConfig, error) {
 	c := &ConnectionConfig{PVersion: "3.1.1"}
 	err := cast.MapToStruct(props, c)
@@ -207,15 +224,6 @@ func validateConfig(props map[string]any) (*ConnectionConfig, error) {
 	return c, nil
 }
 
-func handleToken(token pahoMqtt.Token) error {
-	if !token.WaitTimeout(5 * time.Second) {
-		return errorx.NewIOErr("timeout")
-	} else if token.Error() != nil {
-		return errorx.NewIOErr(token.Error().Error())
-	}
-	return nil
-}
-
 func (conn *Connection) GetClientId() string {
 	return conn.selId
 }
diff --git a/internal/io/mqtt/sink.go b/internal/io/mqtt/sink.go
index 55aba596..791afe4f 100644
--- a/internal/io/mqtt/sink.go
+++ b/internal/io/mqtt/sink.go
@@ -20,16 +20,14 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
-	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 )
 
 // AdConf is the advanced configuration for the mqtt sink
 type AdConf struct {
-	Tpc         string `json:"topic"`
-	Qos         byte   `json:"qos"`
-	Retained    bool   `json:"retained"`
-	SelId       string `json:"connectionSelector"`
-	ResendTopic string `json:"resendDestination"`
+	Tpc      string `json:"topic"`
+	Qos      byte   `json:"qos"`
+	Retained bool   `json:"retained"`
+	SelId    string `json:"connectionSelector"`
 }
 
 type Sink struct {
@@ -58,9 +56,6 @@ func (ms *Sink) Provision(_ api.StreamContext, ps map[string]any) error {
 		return fmt.Errorf("invalid qos value %v, the value could be only int 0 or 1 or 2", adconf.Qos)
 	}
 	ms.config = ps
-	if adconf.ResendTopic == "" {
-		adconf.ResendTopic = adconf.Tpc
-	}
 	ms.adconf = adconf
 	return nil
 }
@@ -89,12 +84,7 @@ func (ms *Sink) Collect(ctx api.StreamContext, item api.RawTuple) error {
 		}
 	}
 	ctx.GetLogger().Debugf("publishing to topic %s", tpc)
-	token := ms.cli.Publish(tpc, ms.adconf.Qos, ms.adconf.Retained, item.Raw())
-	err := handleToken(token)
-	if err != nil {
-		return errorx.NewIOErr(fmt.Sprintf("found error when publishing to topic %s: %s", ms.adconf.Tpc, err))
-	}
-	return nil
+	return ms.cli.Publish(tpc, ms.adconf.Qos, ms.adconf.Retained, item.Raw())
 }
 
 func (ms *Sink) Close(ctx api.StreamContext) error {
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..364d296c 100644
--- a/internal/topo/node/cache_op.go
+++ b/internal/topo/node/cache_op.go
@@ -23,8 +23,8 @@ import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"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"
 )
@@ -100,11 +100,21 @@ func (s *CacheOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 
 				s.statManager.ProcessTimeEnd()
 				s.statManager.IncTotalMessagesProcessed(1)
+				l := int64(len(s.input) + s.cache.CacheLength)
+				if s.currItem != nil {
+					l += 1
+				}
+				s.statManager.SetBufferLength(l)
 			case <-s.resendTimerCh:
 				ctx.GetLogger().Debugf("ticker is triggered")
 				s.statManager.ProcessTimeStart()
 				s.send()
 				s.statManager.ProcessTimeEnd()
+				l := int64(len(s.input) + s.cache.CacheLength)
+				if s.currItem != nil {
+					l += 1
+				}
+				s.statManager.SetBufferLength(l)
 			}
 		}
 	}()
@@ -130,23 +140,7 @@ func (s *CacheOp) send() {
 		}
 	}
 	// Send by custom broadcast, if successful, reset currItem to nil
-	s.Broadcast(s.currItem)
-}
-
-func (s *CacheOp) Broadcast(val interface{}) {
-	if _, ok := val.(error); ok && !s.sendError {
-		return
-	}
-	if s.qos >= def.AtLeastOnce {
-		boe := &checkpoint.BufferOrEvent{
-			Data:    val,
-			Channel: s.name,
-		}
-		s.doBroadcast(boe)
-		return
-	}
-	s.doBroadcast(val)
-	return
+	s.BroadcastCustomized(s.currItem, s.doBroadcast)
 }
 
 func (s *CacheOp) doBroadcast(val interface{}) {
@@ -166,7 +160,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/node/node.go b/internal/topo/node/node.go
index e9902fcd..09a15e23 100644
--- a/internal/topo/node/node.go
+++ b/internal/topo/node/node.go
@@ -83,6 +83,10 @@ func (o *defaultNode) RemoveMetrics(ruleId string) {
 }
 
 func (o *defaultNode) Broadcast(val interface{}) {
+	o.BroadcastCustomized(val, o.doBroadcast)
+}
+
+func (o *defaultNode) BroadcastCustomized(val interface{}, broadcastFunc func(val any)) {
 	if _, ok := val.(error); ok && !o.sendError {
 		return
 	}
@@ -91,10 +95,10 @@ func (o *defaultNode) Broadcast(val interface{}) {
 			Data:    val,
 			Channel: o.name,
 		}
-		o.doBroadcast(boe)
+		broadcastFunc(boe)
 		return
 	}
-	o.doBroadcast(val)
+	broadcastFunc(val)
 	return
 }
 
diff --git a/internal/topo/node/sink_node.go b/internal/topo/node/sink_node.go
index 7ec5d4f4..985c70a3 100644
--- a/internal/topo/node/sink_node.go
+++ b/internal/topo/node/sink_node.go
@@ -17,10 +17,13 @@ package node
 import (
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 	"github.com/lf-edge/ekuiper/v2/pkg/model"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
@@ -31,10 +34,34 @@ import (
 // This node is the skeleton. It will refer to a sink instance to do the real work.
 type SinkNode struct {
 	*defaultSinkNode
-	sink       api.Sink
-	eoflimit   int
-	currentEof int
-	doCollect  func(ctx api.StreamContext, sink api.Sink, data any) error
+	sink           api.Sink
+	eoflimit       int
+	currentEof     int
+	resendInterval int
+	doCollect      func(ctx api.StreamContext, sink api.Sink, data any) error
+	// channel for resend
+	resendOut chan<- any
+}
+
+func newSinkNode(ctx api.StreamContext, name string, rOpt def.RuleOption, eoflimit int, sc *conf.SinkConf, isRetry bool) *SinkNode {
+	// set collect retry according to cache setting
+	retry := sc.ResendInterval
+	if !sc.EnableCache && !isRetry {
+		retry = 0
+	} else if retry <= 0 {
+		// default retry interval to 100ms
+		retry = 100
+	}
+	// Sink input channel as buffer
+	if isRetry || (sc.EnableCache && !sc.ResendAlterQueue) {
+		rOpt.BufferLength = sc.MemoryCacheThreshold
+	}
+	ctx.GetLogger().Infof("create sink node %s with isRetry %v, resendInterval %d, bufferLength %d", name, isRetry, retry, rOpt.BufferLength)
+	return &SinkNode{
+		defaultSinkNode: newDefaultSinkNode(name, &rOpt),
+		eoflimit:        eoflimit,
+		resendInterval:  retry,
+	}
 }
 
 func (s *SinkNode) Exec(ctx api.StreamContext, errCh chan<- error) {
@@ -59,19 +86,48 @@ func (s *SinkNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 				s.statManager.IncTotalRecordsIn()
 				s.statManager.ProcessTimeStart()
 				err = s.doCollect(ctx, s.sink, data)
-				if err != nil {
+				if err != nil { // resend handling when enabling cache. Two cases: 1. send to alter queue with resendOUt. 2. retry (blocking) until success or unrecoverable error if resendInterval is set
 					ctx.GetLogger().Error(err)
 					s.statManager.IncTotalExceptions(err.Error())
+					if s.resendOut != nil {
+						s.BroadcastCustomized(data, func(val any) {
+							select {
+							case s.resendOut <- val:
+								// do nothing
+							case <-ctx.Done():
+								// rule stop so stop waiting
+							default:
+								s.statManager.IncTotalExceptions(fmt.Sprintf("buffer full, drop message from %s to resend sink", s.name))
+							}
+						})
+					} else if s.resendInterval > 0 {
+						for err != nil && errorx.IsIOError(err) {
+							timex.Sleep(time.Duration(s.resendInterval) * time.Millisecond)
+							err = s.doCollect(ctx, s.sink, data)
+							ctx.GetLogger().Debugf("resending, got err? %v", err)
+							s.statManager.SetBufferLength(int64(len(s.input)))
+						}
+						if err == nil {
+							s.statManager.IncTotalRecordsOut()
+						} else {
+							ctx.GetLogger().Errorf("resend error %v", err)
+						}
+					}
 				} else {
 					s.statManager.IncTotalRecordsOut()
 				}
 				s.statManager.ProcessTimeEnd()
 				s.statManager.IncTotalMessagesProcessed(1)
+				s.statManager.SetBufferLength(int64(len(s.input)))
 			}
 		}
 	}()
 }
 
+func (s *SinkNode) SetResendOutput(output chan<- any) {
+	s.resendOut = output
+}
+
 func (s *SinkNode) ingest(ctx api.StreamContext, item any) (any, bool) {
 	ctx.GetLogger().Debugf("%s_%d receive %v", ctx.GetOpId(), ctx.GetInstanceId(), item)
 	item, processed := s.preprocess(ctx, item)
@@ -99,14 +155,12 @@ func (s *SinkNode) ingest(ctx api.StreamContext, item any) (any, bool) {
 }
 
 // NewBytesSinkNode creates a sink node that collects data from the stream. Do some static validation
-func NewBytesSinkNode(ctx api.StreamContext, name string, sink api.BytesCollector, rOpt *def.RuleOption, eoflimit int) (*SinkNode, error) {
+func NewBytesSinkNode(ctx api.StreamContext, name string, sink api.BytesCollector, rOpt def.RuleOption, eoflimit int, sc *conf.SinkConf, isRetry bool) (*SinkNode, error) {
 	ctx.GetLogger().Infof("create bytes sink node %s", name)
-	return &SinkNode{
-		defaultSinkNode: newDefaultSinkNode(name, rOpt),
-		sink:            sink,
-		doCollect:       bytesCollect,
-		eoflimit:        eoflimit,
-	}, nil
+	n := newSinkNode(ctx, name, rOpt, eoflimit, sc, isRetry)
+	n.sink = sink
+	n.doCollect = bytesCollect
+	return n, nil
 }
 
 func bytesCollect(ctx api.StreamContext, sink api.Sink, data any) (err error) {
@@ -125,14 +179,12 @@ func bytesCollect(ctx api.StreamContext, sink api.Sink, data any) (err error) {
 }
 
 // NewTupleSinkNode creates a sink node that collects data from the stream. Do some static validation
-func NewTupleSinkNode(ctx api.StreamContext, name string, sink api.TupleCollector, rOpt *def.RuleOption, eoflimit int) (*SinkNode, error) {
+func NewTupleSinkNode(ctx api.StreamContext, name string, sink api.TupleCollector, rOpt def.RuleOption, eoflimit int, sc *conf.SinkConf, isRetry bool) (*SinkNode, error) {
 	ctx.GetLogger().Infof("create message sink node %s", name)
-	return &SinkNode{
-		defaultSinkNode: newDefaultSinkNode(name, rOpt),
-		sink:            sink,
-		doCollect:       tupleCollect,
-		eoflimit:        eoflimit,
-	}, nil
+	n := newSinkNode(ctx, name, rOpt, eoflimit, sc, isRetry)
+	n.sink = sink
+	n.doCollect = tupleCollect
+	return n, nil
 }
 
 // return error that cannot be sent
diff --git a/internal/topo/planner/planner_sink.go b/internal/topo/planner/planner_sink.go
index ec7c277e..f49b90f1 100644
--- a/internal/topo/planner/planner_sink.go
+++ b/internal/topo/planner/planner_sink.go
@@ -55,12 +55,13 @@ func buildActions(tp *topo.Topo, rule *def.Rule, inputs []node.Emitter, streamCo
 				return err
 			}
 			tp.GetContext().GetLogger().Infof("provision sink %s with props %+v", sinkName, props)
+
 			var snk node.DataSinkNode
 			switch ss := s.(type) {
 			case api.BytesCollector:
-				snk, err = node.NewBytesSinkNode(tp.GetContext(), sinkName, ss, rule.Options, streamCount)
+				snk, err = node.NewBytesSinkNode(tp.GetContext(), sinkName, ss, *rule.Options, streamCount, &commonConf.SinkConf, false)
 			case api.TupleCollector:
-				snk, err = node.NewTupleSinkNode(tp.GetContext(), sinkName, ss, rule.Options, streamCount)
+				snk, err = node.NewTupleSinkNode(tp.GetContext(), sinkName, ss, *rule.Options, streamCount, &commonConf.SinkConf, false)
 			default:
 				err = fmt.Errorf("sink type %s does not implement any collector", name)
 			}
@@ -68,6 +69,41 @@ func buildActions(tp *topo.Topo, rule *def.Rule, inputs []node.Emitter, streamCo
 				return err
 			}
 			tp.AddSink(newInputs, snk)
+			// Cache in alter queue, the topo becomes sink (fail) -> cache -> resendSink
+			// If no alter queue, the topo is cache -> sink
+			if commonConf.EnableCache && commonConf.ResendAlterQueue {
+				s, _ := io.Sink(name)
+				// TODO currently, the destination prop must be named topic
+				if commonConf.ResendDestination != "" {
+					props["topic"] = commonConf.ResendDestination
+				}
+				if err = s.Provision(tp.GetContext(), props); err != nil {
+					return err
+				}
+				tp.GetContext().GetLogger().Infof("provision sink %s with props %+v", sinkName, props)
+
+				cacheOp, err := node.NewCacheOp(tp.GetContext(), fmt.Sprintf("%s_cache", sinkName), rule.Options, &commonConf.SinkConf)
+				if err != nil {
+					return err
+				}
+				tp.AddSinkAlterOperator(snk.(*node.SinkNode), cacheOp)
+				newInputs = []node.Emitter{cacheOp}
+
+				sinkName := fmt.Sprintf("%s_resend_%d", name, i)
+				var snk node.DataSinkNode
+				switch ss := s.(type) {
+				case api.BytesCollector:
+					snk, err = node.NewBytesSinkNode(tp.GetContext(), sinkName, ss, *rule.Options, streamCount, &commonConf.SinkConf, true)
+				case api.TupleCollector:
+					snk, err = node.NewTupleSinkNode(tp.GetContext(), sinkName, ss, *rule.Options, streamCount, &commonConf.SinkConf, true)
+				default:
+					err = fmt.Errorf("sink type %s does not implement any collector", name)
+				}
+				if err != nil {
+					return err
+				}
+				tp.AddSink(newInputs, snk)
+			}
 		}
 	}
 	return nil
@@ -144,7 +180,7 @@ func splitSink(tp *topo.Topo, inputs []node.Emitter, s api.Sink, sinkName string
 		}
 	}
 	// Caching
-	if sc.EnableCache {
+	if sc.EnableCache && !sc.ResendAlterQueue {
 		cacheOp, err := node.NewCacheOp(tp.GetContext(), fmt.Sprintf("%s_%d_cache", sinkName, index), options, &sc.SinkConf)
 		if err != nil {
 			return nil, err
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index 18a704d3..6cccd69d 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"
 )
@@ -124,10 +125,19 @@ func (s *Topo) AddSink(inputs []node.Emitter, snk node.DataSinkNode) *Topo {
 	return s
 }
 
+func (s *Topo) AddSinkAlterOperator(sink *node.SinkNode, operator node.OperatorNode) *Topo {
+	ch, _ := operator.GetInput()
+	sink.SetResendOutput(ch)
+	operator.AddInputCount()
+	s.addEdge(sink, operator, "op")
+	s.ops = append(s.ops, operator)
+	return s
+}
+
 func (s *Topo) AddOperator(inputs []node.Emitter, operator node.OperatorNode) *Topo {
+	ch, opName := operator.GetInput()
 	for _, input := range inputs {
 		// add rule id to make operator name unique
-		ch, opName := operator.GetInput()
 		_ = input.AddOutput(ch, fmt.Sprintf("%s_%s", s.name, opName))
 		operator.AddInputCount()
 		switch rt := input.(type) {
@@ -145,6 +155,8 @@ func (s *Topo) addEdge(from node.TopNode, to node.TopNode, toType string) {
 	fromType := "op"
 	if _, ok := from.(node.DataSourceNode); ok {
 		fromType = "source"
+	} else if _, ok := from.(*node.SinkNode); ok {
+		fromType = "sink"
 	}
 	f := fmt.Sprintf("%s_%s", fromType, from.GetName())
 	t := fmt.Sprintf("%s_%s", toType, to.GetName())
@@ -214,7 +226,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 +258,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 +281,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)
+}
diff --git a/pkg/timex/time.go b/pkg/timex/time.go
index 0bddadd7..5884a08c 100644
--- a/pkg/timex/time.go
+++ b/pkg/timex/time.go
@@ -54,6 +54,10 @@ func GetTimer(duration int64) *clock.Timer {
 	return Clock.Timer(time.Duration(duration) * time.Millisecond)
 }
 
+func Sleep(duration time.Duration) {
+	Clock.Sleep(duration)
+}
+
 func GetTimerByTime(t time.Time) *clock.Timer {
 	if IsTesting {
 		return Clock.Timer(time.Duration(t.UnixMilli()-GetNowInMilli()) * time.Millisecond)


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/io/mqtt/sink_test.go b/internal/io/mqtt/sink_test.go
index da9a579c..903efd3e 100644
--- a/internal/io/mqtt/sink_test.go
+++ b/internal/io/mqtt/sink_test.go
@@ -95,11 +95,10 @@ func TestSinkConfigure(t *testing.T) {
 				"connectionSelector": "mqtt.local",
 			},
 			expectedAdConf: &AdConf{
-				Tpc:         "testTopic3",
-				Qos:         0,
-				Retained:    false,
-				ResendTopic: "testTopic3",
-				SelId:       "mqtt.local",
+				Tpc:      "testTopic3",
+				Qos:      0,
+				Retained: false,
+				SelId:    "mqtt.local",
 			},
 		},
 		{
@@ -112,10 +111,9 @@ func TestSinkConfigure(t *testing.T) {
 				"server":      "123",
 			},
 			expectedAdConf: &AdConf{
-				Tpc:         "testTopic4",
-				Qos:         1,
-				Retained:    false,
-				ResendTopic: "testTopic4",
+				Tpc:      "testTopic4",
+				Qos:      1,
+				Retained: false,
 			},
 		},
 	}
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/sink_node_test.go b/internal/topo/node/sink_node_test.go
new file mode 100644
index 00000000..95ba0f6d
--- /dev/null
+++ b/internal/topo/node/sink_node_test.go
@@ -0,0 +1,190 @@
+// 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 node
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+func TestNewSinkNode(t *testing.T) {
+	ctx := mockContext.NewMockContext("testSink", "sink")
+	tests := []struct {
+		name           string
+		sc             *conf.SinkConf
+		isRetry        bool
+		resendInterval int
+		bufferLength   int
+	}{
+		{
+			name: "normal sink",
+			sc: &conf.SinkConf{
+				ResendInterval: 100,
+			},
+			isRetry:        false,
+			resendInterval: 0,
+			bufferLength:   1024,
+		},
+		{
+			name: "linear cache sink",
+			sc: &conf.SinkConf{
+				ResendInterval:       100,
+				EnableCache:          true,
+				MemoryCacheThreshold: 10,
+			},
+			isRetry:        false,
+			resendInterval: 100,
+			bufferLength:   10,
+		},
+		{
+			name: "retry cache normal sink",
+			sc: &conf.SinkConf{
+				ResendInterval:       100,
+				EnableCache:          true,
+				MemoryCacheThreshold: 10,
+				ResendAlterQueue:     true,
+			},
+			isRetry: false,
+			// resend interval is set but no use
+			resendInterval: 100,
+			bufferLength:   1024,
+		},
+		{
+			name: "retry cache resend sink",
+			sc: &conf.SinkConf{
+				ResendInterval:       100,
+				EnableCache:          true,
+				MemoryCacheThreshold: 10,
+				ResendAlterQueue:     true,
+			},
+			isRetry: true,
+			// resend interval is set but no use
+			resendInterval: 100,
+			bufferLength:   10,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			n := newSinkNode(ctx, "test", def.RuleOption{
+				BufferLength: 1024,
+			}, 1, tt.sc, tt.isRetry)
+			assert.Equal(t, tt.resendInterval, n.resendInterval, "resend interval")
+			assert.Equal(t, tt.bufferLength, cap(n.input))
+		})
+	}
+}
+
+func TestRetry(t *testing.T) {
+	ctx, cancel := mockContext.NewMockContext("resendout", "sink").WithCancel()
+	s := &mockResendSink{failTimes: 2}
+	n, err := NewBytesSinkNode(ctx, "resendout_sink", s, def.RuleOption{
+		BufferLength: 1024,
+	}, 1, &conf.SinkConf{
+		ResendInterval:       100,
+		EnableCache:          true,
+		MemoryCacheThreshold: 10,
+	}, true)
+	assert.NoError(t, err)
+	data := &xsql.RawTuple{
+		Emitter:   "",
+		Timestamp: 1,
+	}
+	errCh := make(chan error, 1)
+	n.Exec(ctx, errCh)
+	n.input <- data
+	for {
+		timex.Add(50 * time.Millisecond)
+		processed := n.statManager.GetMetrics()[2]
+		if processed == int64(1) {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+	assert.Equal(t, s.val, data)
+}
+
+func TestResendOut(t *testing.T) {
+	ctx, cancel := mockContext.NewMockContext("resendout", "sink").WithCancel()
+	s := &mockResendSink{failTimes: 10}
+	n, err := NewBytesSinkNode(ctx, "resendout_sink", s, def.RuleOption{
+		BufferLength: 1024,
+	}, 1, &conf.SinkConf{
+		ResendInterval:       100,
+		EnableCache:          true,
+		MemoryCacheThreshold: 10,
+		ResendAlterQueue:     true,
+	}, true)
+	assert.NoError(t, err)
+	alertCh := make(chan any, 10)
+	n.SetResendOutput(alertCh)
+	data := &xsql.RawTuple{
+		Emitter:   "",
+		Timestamp: 1,
+	}
+	errCh := make(chan error, 1)
+	n.Exec(ctx, errCh)
+	go func() {
+		n.input <- data
+	}()
+	got := false
+	select {
+	case d := <-alertCh:
+		assert.Equal(t, data, d)
+		got = true
+	case e := <-errCh:
+		assert.NoError(t, e)
+	}
+	cancel()
+	assert.True(t, got)
+}
+
+type mockResendSink struct {
+	failTimes int
+	val       any
+}
+
+func (m *mockResendSink) Provision(ctx api.StreamContext, configs map[string]any) error {
+	return nil
+}
+
+func (m *mockResendSink) Close(ctx api.StreamContext) error {
+	return nil
+}
+
+func (m *mockResendSink) Connect(ctx api.StreamContext) error {
+	return nil
+}
+
+func (m *mockResendSink) Collect(ctx api.StreamContext, item api.RawTuple) error {
+	if m.failTimes > 0 {
+		m.failTimes--
+		return errorx.NewIOErr("fake error")
+	}
+	m.val = item
+	return nil
+}
+
+var _ api.BytesCollector = &mockResendSink{}
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_sink_test.go b/internal/topo/planner/planner_sink_test.go
index b5b96817..5d6b0bb1 100644
--- a/internal/topo/planner/planner_sink_test.go
+++ b/internal/topo/planner/planner_sink_test.go
@@ -212,6 +212,40 @@ func TestSinkPlan(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "resend sink plan",
+			rule: &def.Rule{
+				Actions: []map[string]any{
+					{
+						"log": map[string]any{
+							"enableCache":      true,
+							"resendAlterQueue": true,
+						},
+					},
+				},
+				Options: defaultOption,
+			},
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_src1"},
+				Edges: map[string][]any{
+					"source_src1": {
+						"op_log_0_0_transform",
+					},
+					"op_log_0_0_transform": {
+						"op_log_0_1_encode",
+					},
+					"op_log_0_1_encode": {
+						"sink_log_0",
+					},
+					"sink_log_0": {
+						"op_log_0_cache",
+					},
+					"op_log_0_cache": {
+						"sink_log_resend_0",
+					},
+				},
+			},
+		},
 	}
 	for _, c := range tc {
 		t.Run(c.name, func(t *testing.T) {
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)
+}