Max buffer should be equal to the concurrency set

Constant buffer size was causing the batcher to die out of memory. We do not need the buffer to be greater than the concurrency. This should greatly solve the scaling issue of batcher and make it perform in most optimum way. #167
practo · Mar 19, 2021 · f00766f · f00766f
1 parent 3b2451b
commit f00766f
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),