fix: enforce disableBufferFullDiscard when qos>=1

internal/conf/conf.go

@@ -27,7 +27,7 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
-	"github.com/yisaer/file-rotatelogs"
+	rotatelogs "github.com/yisaer/file-rotatelogs"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf/logger"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
@@ -290,6 +290,15 @@ func ValidateRuleOption(option *def.RuleOption) error {
 	if err := schedule.ValidateRanges(option.CronDatetimeRange); err != nil {
 		errs = errors.Join(errs, fmt.Errorf("validate cronDatetimeRange failed, err:%v", err))
 	}
+	if option.Qos >= def.AtLeastOnce {
+		if option.DisableBufferFullDiscard == nil {
+			t := true
+			option.DisableBufferFullDiscard = &t
+		} else if !*option.DisableBufferFullDiscard {
+			Log.Warnf("QoS is %d but disableBufferFullDiscard is explicitly set to false, this may lead to silent data loss during congestion.", option.Qos)
+			errs = errors.Join(errs, fmt.Errorf("invalidDisableBufferFullDiscard:disableBufferFullDiscard must be true when qos is %d or higher", def.AtLeastOnce))
+		}
+	}
 	return errs
 }
 

internal/conf/conf_test.go

@@ -107,6 +107,26 @@ func TestRuleOptionValidate(t *testing.T) {
 				},
 			},
 		},
+		{
+			s: &def.RuleOption{
+				Qos: def.AtLeastOnce,
+			},
+			e: &def.RuleOption{
+				Qos:                      def.AtLeastOnce,
+				DisableBufferFullDiscard: func() *bool { b := true; return &b }(),
+			},
+		},
+		{
+			s: &def.RuleOption{
+				Qos:                      def.AtLeastOnce,
+				DisableBufferFullDiscard: func() *bool { b := false; return &b }(),
+			},
+			e: &def.RuleOption{
+				Qos:                      def.AtLeastOnce,
+				DisableBufferFullDiscard: func() *bool { b := false; return &b }(),
+			},
+			err: "invalidDisableBufferFullDiscard:disableBufferFullDiscard must be true when qos is 1 or higher",
+		},
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 	for i, tt := range tests {

internal/pkg/def/rule.go

@@ -42,7 +42,7 @@ type RuleOption struct {
 	CronDatetimeRange         []schedule.DatetimeRange `json:"cronDatetimeRange,omitempty" yaml:"cronDatetimeRange,omitempty"`
 	PlanOptimizeStrategy      *PlanOptimizeStrategy    `json:"planOptimizeStrategy,omitempty" yaml:"planOptimizeStrategy,omitempty"`
 	NotifySub                 bool                     `json:"notifySub,omitempty" yaml:"notifySub,omitempty"`
-	DisableBufferFullDiscard  bool                     `json:"disableBufferFullDiscard,omitempty" yaml:"disableBufferFullDiscard,omitempty"`
+	DisableBufferFullDiscard  *bool                    `json:"disableBufferFullDiscard,omitempty" yaml:"disableBufferFullDiscard,omitempty"`
 	EnableSaveStateBeforeStop bool                     `json:"enableSaveStateBeforeStop,omitempty" yaml:"enableSaveStateBeforeStop,omitempty"`
 	ForceExitTimeout          cast.DurationConf        `json:"forceExitTimeout,omitempty" yaml:"forceExitTimeout,omitempty"`
 	Experiment                *ExpOpts                 `json:"experiment,omitempty" yaml:"experiment,omitempty"`

internal/processor/rule_test.go

@@ -143,14 +143,15 @@ func TestRuleActionParse_Apply(t *testing.T) {
 					},
 				},
 				Options: &def.RuleOption{
-					IsEventTime:        true,
-					LateTol:            cast.DurationConf(time.Second),
-					Concurrency:        1,
-					BufferLength:       10240,
-					SendMetaToSink:     false,
-					Qos:                def.ExactlyOnce,
-					CheckpointInterval: cast.DurationConf(time.Minute),
-					SendError:          false,
+					IsEventTime:              true,
+					LateTol:                  cast.DurationConf(time.Second),
+					Concurrency:              1,
+					BufferLength:             10240,
+					SendMetaToSink:           false,
+					Qos:                      def.ExactlyOnce,
+					CheckpointInterval:       cast.DurationConf(time.Minute),
+					SendError:                false,
+					DisableBufferFullDiscard: boolPtr(true),
 					RestartStrategy: &def.RestartStrategy{
 						Attempts: 0,
 					},
@@ -257,3 +258,7 @@ func TestAllRules(t *testing.T) {
 	}
 	assert.Equal(t, expected, all)
 }
+
+func boolPtr(b bool) *bool {
+	return &b
+}

internal/topo/node/node.go

@@ -66,7 +66,7 @@ func newDefaultNode(name string, options *def.RuleOption) *defaultNode {
 		outputs:                  make(map[string]chan any),
 		concurrency:              c,
 		sendError:                options.SendError,
-		disableBufferFullDiscard: options.DisableBufferFullDiscard,
+		disableBufferFullDiscard: options.DisableBufferFullDiscard != nil && *options.DisableBufferFullDiscard,
 	}
 }
 
@@ -98,6 +98,9 @@ func (o *defaultNode) GetName() string {
 
 func (o *defaultNode) SetQos(qos def.Qos) {
 	o.qos = qos
+	if qos >= def.AtLeastOnce {
+		o.disableBufferFullDiscard = true
+	}
 }
 
 func (o *defaultNode) GetMetrics() []any {

internal/topo/node/node_test.go

@@ -113,3 +113,19 @@ func TestMultipleOutputsBroadcast(t *testing.T) {
 		})
 	}
 }
+
+func TestSetQos(t *testing.T) {
+	tests := []struct {
+		qos                              def.Qos
+		expectedDisableBufferFullDiscard bool
+	}{
+		{qos: def.AtMostOnce, expectedDisableBufferFullDiscard: false},
+		{qos: def.AtLeastOnce, expectedDisableBufferFullDiscard: true},
+		{qos: def.ExactlyOnce, expectedDisableBufferFullDiscard: true},
+	}
+	for _, tt := range tests {
+		n := newDefaultNode("test", &def.RuleOption{})
+		n.SetQos(tt.qos)
+		assert.Equal(t, tt.expectedDisableBufferFullDiscard, n.disableBufferFullDiscard)
+	}
+}

internal/topo/planner/planner.go

@@ -506,7 +506,7 @@ func CreateLogicalPlan(stmt *ast.SelectStatement, opt *def.RuleOption, store kv.
 }
 
 func checkSharedSourceOption(streams []*streamInfo, opt *def.RuleOption) error {
-	if !opt.DisableBufferFullDiscard {
+	if opt.DisableBufferFullDiscard == nil || !*opt.DisableBufferFullDiscard {
 		return nil
 	}
 	for _, stream := range streams {

internal/topo/planner/planner_test.go

@@ -4672,8 +4672,9 @@ func TestCheckSharedSourceOption(t *testing.T) {
 			},
 		},
 	}
+	b := true
 	r1 := &def.RuleOption{
-		DisableBufferFullDiscard: true,
+		DisableBufferFullDiscard: &b,
 	}
 	require.Error(t, checkSharedSourceOption(s1, r1))
 }

internal/topo/topotest/rule_test.go

@@ -1245,14 +1245,15 @@ func TestSingleSQL(t *testing.T) {
 		},
 	}
 	HandleStream(true, streamList, t)
+	bTrue := true
 	options := []*def.RuleOption{
 		{
 			BufferLength: 100,
 			SendError:    true,
 			PlanOptimizeStrategy: &def.PlanOptimizeStrategy{
 				EnableIncrementalWindow: true,
 			},
-			DisableBufferFullDiscard: true,
+			DisableBufferFullDiscard: &bTrue,
 		},
 		{
 			BufferLength:       100,
@@ -1262,7 +1263,7 @@ func TestSingleSQL(t *testing.T) {
 			PlanOptimizeStrategy: &def.PlanOptimizeStrategy{
 				EnableIncrementalWindow: true,
 			},
-			DisableBufferFullDiscard: true,
+			DisableBufferFullDiscard: &bTrue,
 		},
 		{
 			BufferLength:       100,
@@ -1272,7 +1273,7 @@ func TestSingleSQL(t *testing.T) {
 			PlanOptimizeStrategy: &def.PlanOptimizeStrategy{
 				EnableIncrementalWindow: true,
 			},
-			DisableBufferFullDiscard: true,
+			DisableBufferFullDiscard: &bTrue,
 		},
 	}
 	for _, opt := range options {