Merge pull request #169 from practo/max-buffer-fix

Max buffer should be equal to the concurrency set
practo · Mar 19, 2021 · 0326795 · 0326795
2 parents 3b2451b + f00766f
commit 0326795
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 3 deletions.
diff --git a/pkg/redshiftbatcher/batch_processor.go b/pkg/redshiftbatcher/batch_processor.go
@@ -411,7 +411,7 @@ func (b *batchProcessor) Process(
 		msgBufs := [][]*serializer.Message{}
 
 		klog.V(2).Infof(
-			"%s: buffchan:%v msgs",
+			"%s: processChan:%v",
 			b.topic,
 			len(processChan),
 		)

diff --git a/pkg/redshiftbatcher/batcher_handler.go b/pkg/redshiftbatcher/batcher_handler.go
@@ -134,7 +134,7 @@ func (h *batcherHandler) ConsumeClaim(
 	)
 
 	var lastSchemaId *int
-	processChan := make(chan []*serializer.Message, 1000)
+	processChan := make(chan []*serializer.Message, h.maxConcurrency)
 	errChan := make(chan error)
 	processor := newBatchProcessor(
 		h.consumerGroupID,

diff --git a/pkg/serializer/message.go b/pkg/serializer/message.go
@@ -60,7 +60,7 @@ func (b *MessageAsyncBatch) Flush() {
 		b.processChan <- b.msgBuf
 		b.msgBuf = make([]*Message, 0, b.maxSize)
 		klog.V(4).Infof(
-			"%s: flushed:%d, buffchan:%v msgs",
+			"%s: flushed:%d, processChan:%v",
 			b.topic,
 			size,
 			len(b.processChan),