fix

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

Hide file tree

Showing 3 changed files with 22 additions and 11 deletions.
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/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 {