Repository: lf-edge/ekuiper
Pull request #2861 (2024-05-20 06:53:58)

fix: let config compatible


Patch:
diff --git a/internal/conf/conf.go b/internal/conf/conf.go
index 3f7dddb9..de25a93b 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       string `json:"resendInterval" yaml:"resendInterval"`
+	ResendInterval       any    `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.
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 59e51186..7283c782 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 string           `json:"checkpointInterval" yaml:"checkpointInterval"`
+	CheckpointInterval any              `json:"checkpointInterval" yaml:"checkpointInterval"`
 	Restart            *RestartStrategy `json:"restartStrategy" yaml:"restartStrategy"`
 	Cron               string           `json:"cron" yaml:"cron"`
 	Duration           string           `json:"duration" yaml:"duration"`
diff --git a/internal/topo/node/cache_op.go b/internal/topo/node/cache_op.go
index 8fca6013..364d296c 100644
--- a/internal/topo/node/cache_op.go
+++ b/internal/topo/node/cache_op.go
@@ -23,7 +23,6 @@ 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"
@@ -101,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)
 			}
 		}
 	}()
@@ -131,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{}) {
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 830393e2..6cccd69d 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -125,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) {
@@ -146,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())
diff --git a/pkg/cast/time.go b/pkg/cast/time.go
index 74d7ad08..46a2b913 100644
--- a/pkg/cast/time.go
+++ b/pkg/cast/time.go
@@ -366,10 +366,16 @@ 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
+func ConvertDuration(s any) (time.Duration, error) {
+	switch x := s.(type) {
+	case string:
+		i, err := strconv.ParseInt(x, 10, 64)
+		if err == nil {
+			return time.Duration(i) * time.Millisecond, nil
+		}
+		return time.ParseDuration(x)
+	case int:
+		return time.Duration(x) * time.Millisecond, nil
 	}
-	return time.ParseDuration(s)
+	return 0, fmt.Errorf("unsupported type:%t", 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/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/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/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/pkg/cast/time_test.go b/pkg/cast/time_test.go
index aca2743c..2bd7f633 100644
--- a/pkg/cast/time_test.go
+++ b/pkg/cast/time_test.go
@@ -277,6 +277,10 @@ func TestConvertDuration(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, 100*time.Millisecond, t1)
 
+	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)