kgo: do not reuse decompression buffer for uncompressed batches (#8)

ref: twmb#823 (comment)

Signed-off-by: Miguel Ángel Ortuño <ortuman@gmail.com>

pkg/kgo/compression.go

@@ -349,6 +349,9 @@ func (d *decompressor) getDecodedBuffer(src []byte, compCodec codecType, pool *p
 }
 
 func (d *decompressor) copyDecodedBuffer(decoded []byte, compCodec codecType, pool *pool.BucketedPool[byte]) []byte {
+	if pool == nil {
+		return append([]byte(nil), decoded...)
+	}
 	if compCodec == codecSnappy {
 		// We already know the actual size of the decoded buffer before decompression,
 		// so there's no need to copy the buffer.

pkg/kgo/record_and_fetch.go

@@ -178,9 +178,13 @@ type Record struct {
 // Once this method has been called, any reference to the passed record should be considered invalid by the caller,
 // as it may be reused as a result of future calls to the PollFetches/PollRecords method.
 func (r *Record) Reuse() {
-	if r.recordsPool != nil {
+	if r.rcRawRecordsBuffer != nil {
 		r.rcRawRecordsBuffer.release()
+	}
+	if r.rcBatchBuffer != nil {
 		r.rcBatchBuffer.release()
+	}
+	if r.recordsPool != nil {
 		r.recordsPool.put(r)
 	}
 }

pkg/kgo/source.go

@@ -1583,7 +1583,9 @@ func processRecordBatch(
 	}
 
 	rawRecords := batch.Records
-	if compression := byte(batch.Attributes & 0x0007); compression != 0 {
+
+	compression := byte(batch.Attributes & 0x0007)
+	if compression != 0 {
 		var err error
 		if rawRecords, err = decompressor.decompress(rawRecords, compression, o.DecompressBufferPool); err != nil {
 			return 0, 0 // truncated batch
@@ -1616,7 +1618,7 @@ func processRecordBatch(
 		rcBatchBuff      *rcBuffer[byte]
 		rcRawRecordsBuff *rcBuffer[kmsg.Record]
 	)
-	if o.recordPool != nil {
+	if o.recordPool != nil && codecType(compression) != codecNone {
 		rcBatchBuff = newRCBuffer(rawRecords, o.DecompressBufferPool)
 		rcRawRecordsBuff = newRCBuffer(krecords, rawRecordsPool)
 	}