fix: fix BatchAbleSink configure (#3515)

Signed-off-by: Song Gao <[email protected]>
lf-edge · Jan 16, 2025 · f3cb876 · f3cb876
1 parent 85a31af
commit f3cb876
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 3 deletions.
diff --git a/internal/topo/node/sink_node.go b/internal/topo/node/sink_node.go
@@ -17,6 +17,7 @@ package node
 import (
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/lf-edge/ekuiper/internal/binder/io"
 	"github.com/lf-edge/ekuiper/internal/conf"
@@ -554,6 +555,11 @@ func resendDataToSink(ctx api.StreamContext, sink api.Sink, outData interface{},
 	}
 }
 
+type batchConf struct {
+	BatchSize      int           `json:"batchSize"`
+	LingerInterval time.Duration `json:"lingerInterval"`
+}
+
 func getSink(name string, action map[string]interface{}) (api.Sink, error) {
 	var (
 		s   api.Sink
@@ -566,6 +572,11 @@ func getSink(name string, action map[string]interface{}) (api.Sink, error) {
 		if err != nil {
 			return nil, err
 		}
+		if bas, ok := s.(api.BatchAbleSink); ok {
+			bc := batchConf{}
+			cast.MapToStruct(newAction, &bc)
+			bas.ConfigureBatch(bc.BatchSize, bc.LingerInterval)
+		}
 		return s, nil
 	} else {
 		if err != nil {

diff --git a/internal/topo/planner/sink_planner.go b/internal/topo/planner/sink_planner.go
@@ -16,7 +16,6 @@ package planner
 
 import (
 	"fmt"
-	"time"
 
 	"github.com/lf-edge/ekuiper/internal/binder/io"
 	"github.com/lf-edge/ekuiper/internal/conf"
@@ -67,8 +66,7 @@ func fulfillProps(rule *api.Rule, props map[string]any) map[string]any {
 
 // Split sink node according to the sink configuration. Return the new input emitters.
 func splitSink(sink api.Sink, tp *topo.Topo, inputs []api.Emitter, sinkName string, options *api.RuleOption, sc *node.SinkConf) ([]api.Emitter, error) {
-	if bas, ok := sink.(api.BatchAbleSink); ok {
-		bas.ConfigureBatch(sc.BatchSize, time.Duration(sc.LingerInterval))
+	if _, ok := sink.(api.BatchAbleSink); ok {
 		return inputs, nil
 	}
 	index := 0