fix: revise kafka batch sink (#3520)

Signed-off-by: Song Gao <[email protected]>
lf-edge · Jan 17, 2025 · a936944 · a936944
1 parent d5ab84d
commit a936944
Show file tree

Hide file tree

Showing 5 changed files with 24 additions and 13 deletions.
diff --git a/extensions/sinks/kafka/ext/kafka.go b/extensions/sinks/kafka/ext/kafka.go
@@ -328,7 +328,7 @@ func getDefaultKafkaConf() *kafkaConf {
 		BatchSize: 5000,
 		// send batch ASAP
 		BatchTimeout: time.Microsecond,
-		BatchBytes:   1048576 * 10, // 10MB
+		BatchBytes:   10 * 1048576, // 1 MB
 	}
 	return c
 }

diff --git a/internal/topo/node/batch_op.go b/internal/topo/node/batch_op.go
@@ -138,6 +138,7 @@ func (b *BatchOp) send() {
 		Content: make([]xsql.Row, 0, b.batchSize),
 	}
 	b.currIndex = 0
+	b.statManager.SetBufferLength(int64(len(b.input) + b.currIndex))
 }
 
 func (b *BatchOp) runWithBatchSize(ctx api.StreamContext) {

diff --git a/internal/topo/node/metric/prometheus.go b/internal/topo/node/metric/prometheus.go
@@ -81,7 +81,7 @@ func newPrometheusMetrics() *PrometheusMetrics {
 		processLatencyHist := prometheus.NewHistogramVec(prometheus.HistogramOpts{
 			Name:    prefix + "_" + ProcessLatencyUsHist,
 			Help:    "Histograms of process latency in millisecond of " + prefix,
-			Buckets: prometheus.ExponentialBuckets(10, 2, 20), // 10us ~ 5s
+			Buckets: prometheus.ExponentialBuckets(10, 2, 25), // 10us ~ 160s
 		}, labelNames)
 		bufferLength := prometheus.NewGaugeVec(prometheus.GaugeOpts{
 			Name: prefix + "_" + BufferLength,

diff --git a/internal/topo/node/sink_node.go b/internal/topo/node/sink_node.go
@@ -287,10 +287,13 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 					if resendCh == nil { // no resend strategy
 						for {
 							select {
-							case data := <-m.input:
-								receiveQ(data)
 							case data := <-dataOutCh:
 								normalQ(data)
+							default:
+							}
+							select {
+							case data := <-m.input:
+								receiveQ(data)
 							case <-ctx.Done():
 								doneQ()
 								return nil
@@ -300,10 +303,13 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 						if sconf.ResendPriority == 0 {
 							for {
 								select {
-								case data := <-m.input:
-									receiveQ(data)
 								case data := <-dataOutCh:
 									normalQ(data)
+								default:
+								}
+								select {
+								case data := <-m.input:
+									receiveQ(data)
 								case data := <-rq.Out:
 									resendQ(data)
 								case <-ctx.Done():
@@ -314,16 +320,17 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 						} else if sconf.ResendPriority < 0 { // normal queue has higher priority
 							for {
 								select {
-								case data := <-m.input:
-									receiveQ(data)
 								case data := <-dataOutCh:
 									normalQ(data)
+								default:
+								}
+								select {
+								case data := <-m.input:
+									receiveQ(data)
 								default:
 									select {
 									case data := <-m.input:
 										receiveQ(data)
-									case data := <-dataOutCh:
-										normalQ(data)
 									case data := <-rq.Out:
 										resendQ(data)
 									case <-ctx.Done():
@@ -334,6 +341,11 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 							}
 						} else {
 							for {
+								select {
+								case data := <-dataOutCh:
+									normalQ(data)
+								default:
+								}
 								select {
 								case data := <-m.input:
 									receiveQ(data)
@@ -343,8 +355,6 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 									select {
 									case data := <-m.input:
 										receiveQ(data)
-									case data := <-dataOutCh:
-										normalQ(data)
 									case data := <-rq.Out:
 										resendQ(data)
 									case <-ctx.Done():

diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
@@ -90,7 +90,7 @@ func (m *SourceNode) Open(ctx api.StreamContext, errCh chan<- error) {
 				}
 			}
 			bl := 102400
-			if c, ok := props["bufferLength"]; ok {
+			if c, ok := props["bufferlength"]; ok {
 				if t, err := cast.ToInt(c, cast.STRICT); err != nil || t <= 0 {
 					logger.Warnf("invalid type for bufferLength property, should be positive integer but found %t", c)
 				} else {