fix(sink): sink bufferLength is not applied (#3608)

Signed-off-by: Song Gao <disxiaofei@163.com>

Author: Yisaer

internal/topo/node/sink_node.go

@@ -22,7 +22,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"
 	kctx "github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
@@ -50,15 +49,19 @@ type SinkNode struct {
 // 1. Set cache settings to enable diskCache
 // 2. Set resendInterval and bufferLength will use bufferLength as the memory cache
 // 3. By default, drop if it cannot sends out.
-func newSinkNode(ctx api.StreamContext, name string, rOpt def.RuleOption, eoflimit int, sc *conf.SinkConf, isRetry bool) *SinkNode {
+func newSinkNode(ctx api.StreamContext, name string, rOpt def.RuleOption, eoflimit int, sc *SinkConf, isRetry bool) *SinkNode {
 	// set collect retry according to cache setting
 	retry := time.Duration(sc.ResendInterval)
 	if (sc.EnableCache || isRetry) && retry <= 0 {
 		// default retry interval to 100ms
 		retry = 100 * time.Millisecond
 	}
 	// Sink input channel as buffer
-	rOpt.BufferLength = sc.MemoryCacheThreshold
+	if isRetry || (sc.EnableCache && !sc.ResendAlterQueue) {
+		rOpt.BufferLength = sc.MemoryCacheThreshold
+	} else {
+		rOpt.BufferLength = sc.BufferLength
+	}
 	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),
@@ -191,7 +194,7 @@ 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, sc *conf.SinkConf, isRetry bool) (*SinkNode, error) {
+func NewBytesSinkNode(ctx api.StreamContext, name string, sink api.BytesCollector, rOpt def.RuleOption, eoflimit int, sc *SinkConf, isRetry bool) (*SinkNode, error) {
 	ctx.GetLogger().Infof("create bytes sink node %s", name)
 	n := newSinkNode(ctx, name, rOpt, eoflimit, sc, isRetry)
 	n.sink = sink
@@ -215,7 +218,7 @@ 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, sc *conf.SinkConf, isRetry bool) (*SinkNode, error) {
+func NewTupleSinkNode(ctx api.StreamContext, name string, sink api.TupleCollector, rOpt def.RuleOption, eoflimit int, sc *SinkConf, isRetry bool) (*SinkNode, error) {
 	ctx.GetLogger().Infof("create message sink node %s", name)
 	n := newSinkNode(ctx, name, rOpt, eoflimit, sc, isRetry)
 	n.sink = sink

internal/topo/node/sink_node_test.go

@@ -91,7 +91,10 @@ func TestNewSinkNode(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			n := newSinkNode(ctx, "test", def.RuleOption{
 				BufferLength: 1024,
-			}, 1, tt.sc, tt.isRetry)
+			}, 1, &SinkConf{
+				SinkConf:     *tt.sc,
+				BufferLength: 1024,
+			}, tt.isRetry)
 			assert.Equal(t, tt.resendInterval, n.resendInterval, "resend interval")
 			assert.Equal(t, tt.bufferLength, cap(n.input))
 		})
@@ -103,10 +106,12 @@ func TestRetry(t *testing.T) {
 	s := &mockResendSink{failTimes: 2}
 	n, err := NewBytesSinkNode(ctx, "resendout_sink", s, def.RuleOption{
 		BufferLength: 1024,
-	}, 1, &conf.SinkConf{
-		ResendInterval:       cast.DurationConf(100 * time.Millisecond),
-		EnableCache:          true,
-		MemoryCacheThreshold: 10,
+	}, 1, &SinkConf{
+		SinkConf: conf.SinkConf{
+			ResendInterval:       cast.DurationConf(100 * time.Millisecond),
+			EnableCache:          true,
+			MemoryCacheThreshold: 10,
+		},
 	}, true)
 	assert.NoError(t, err)
 	data := &xsql.RawTuple{
@@ -133,11 +138,13 @@ func TestResendOut(t *testing.T) {
 	s := &mockResendSink{failTimes: 10}
 	n, err := NewBytesSinkNode(ctx, "resendout_sink", s, def.RuleOption{
 		BufferLength: 1024,
-	}, 1, &conf.SinkConf{
-		ResendInterval:       cast.DurationConf(100 * time.Millisecond),
-		EnableCache:          true,
-		MemoryCacheThreshold: 10,
-		ResendAlterQueue:     true,
+	}, 1, &SinkConf{
+		SinkConf: conf.SinkConf{
+			ResendInterval:       cast.DurationConf(100 * time.Millisecond),
+			EnableCache:          true,
+			MemoryCacheThreshold: 10,
+			ResendAlterQueue:     true,
+		},
 	}, true)
 	assert.NoError(t, err)
 	alertCh := make(chan any, 10)

internal/topo/planner/planner_sink.go

@@ -102,9 +102,9 @@ func SinkToComp(tp *topo.Topo, sinkType string, sinkName string, props map[strin
 	var snk node.DataSinkNode
 	switch ss := s.(type) {
 	case api.BytesCollector:
-		snk, err = node.NewBytesSinkNode(tp.GetContext(), sinkName, ss, *rule.Options, streamCount, &commonConf.SinkConf, false)
+		snk, err = node.NewBytesSinkNode(tp.GetContext(), sinkName, ss, *rule.Options, streamCount, commonConf, false)
 	case api.TupleCollector:
-		snk, err = node.NewTupleSinkNode(tp.GetContext(), sinkName, ss, *rule.Options, streamCount, &commonConf.SinkConf, false)
+		snk, err = node.NewTupleSinkNode(tp.GetContext(), sinkName, ss, *rule.Options, streamCount, commonConf, false)
 	default:
 		err = fmt.Errorf("sink type %s does not implement any collector", sinkType)
 	}
@@ -135,9 +135,9 @@ func SinkToComp(tp *topo.Topo, sinkType string, sinkName string, props map[strin
 		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)
+			snk, err = node.NewBytesSinkNode(tp.GetContext(), sinkName, ss, *rule.Options, streamCount, commonConf, true)
 		case api.TupleCollector:
-			snk, err = node.NewTupleSinkNode(tp.GetContext(), sinkName, ss, *rule.Options, streamCount, &commonConf.SinkConf, true)
+			snk, err = node.NewTupleSinkNode(tp.GetContext(), sinkName, ss, *rule.Options, streamCount, commonConf, true)
 		default:
 			err = fmt.Errorf("sink type %s does not implement any collector", sinkType)
 		}