Fix memory leak in bulk indexer

elastic#701

Author: mhmtszr

esutil/bulk_indexer.go

@@ -279,6 +279,10 @@ type bulkIndexerStats struct {
 	flushedBytes uint64
 }
 
+var bufPool = &sync.Pool{
+	New: func() any { return new(bytes.Buffer) },
+}
+
 // NewBulkIndexer creates a new bulk indexer.
 func NewBulkIndexer(cfg BulkIndexerConfig) (BulkIndexer, error) {
 	if cfg.Client == nil {
@@ -351,6 +355,11 @@ func (bi *bulkIndexer) Close(ctx context.Context) error {
 		bi.wg.Wait()
 	}
 
+	for _, w := range bi.workers {
+		w.buf.Reset()
+		bufPool.Put(w.buf)
+	}
+
 	return nil
 }
 
@@ -379,7 +388,7 @@ func (bi *bulkIndexer) init() {
 			id:     i,
 			ch:     bi.queue,
 			bi:     bi,
-			buf:    bytes.NewBuffer(make([]byte, 0, bi.config.FlushBytes)),
+			buf:    bufPool.Get().(*bytes.Buffer),
 			ticker: time.NewTicker(bi.config.FlushInterval),
 		}
 		w.run()