M

Signed-off-by: Miguel Ángel Ortuño <[email protected]>
grafana · Oct 1, 2024 · 9d508f2 · 9d508f2
1 parent 95bcb25
commit 9d508f2
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/pkg/kgo/compression.go b/pkg/kgo/compression.go
@@ -277,7 +277,6 @@ func (d *decompressor) decompress(src []byte, codec byte, pool *pool.BucketedPoo
 	var out *bytes.Buffer
 
 	if pool != nil {
-		// Assume the worst case scenario here, where decompressed buffer size is FetchMaxBytes.
 		outBuf := pool.Get(pool.MaxSize())[:0]
 		defer func() {
 			pool.Put(outBuf)

diff --git a/pkg/kgo/config.go b/pkg/kgo/config.go
@@ -395,7 +395,10 @@ func (cfg *cfg) validate() error {
 	cfg.hooks = processedHooks
 
 	if cfg.recordsPool != nil {
-		cfg.decompressBufferPool = pool.NewBucketedPool[byte](4096, int(cfg.maxBytes.load()), 2, func(sz int) []byte {
+		// Assume a 2x compression ratio.
+		maxDecompressBatchSize := int(cfg.maxBytes.load()) * 2
+
+		cfg.decompressBufferPool = pool.NewBucketedPool[byte](4096, maxDecompressBatchSize, 2, func(sz int) []byte {
 			return make([]byte, sz)
 		})
 	}