fix(decoder): use configurable limit for max number of records in a record batch (#3120)

Replace the `tmp > 2*math.MaxUint16` arbitrary number safety blanket with `tmp > MaxResponseSize` as we already have that constant (that a user can change via sarama.MaxResponseSize = ...) which is what we use for MaxBytes in a FetchRequest.

Also added a test for large record counts.

Fixes: #3119

Author: rmb938

real_decoder.go

@@ -115,7 +115,7 @@ func (rd *realDecoder) getArrayLength() (int, error) {
 	if tmp > rd.remaining() {
 		rd.off = len(rd.raw)
 		return -1, ErrInsufficientData
-	} else if tmp > 2*math.MaxUint16 {
+	} else if tmp > int(MaxResponseSize) {
 		return -1, errInvalidArrayLength
 	}
 	return tmp, nil

record_test.go

@@ -3,6 +3,8 @@
 package sarama
 
 import (
+	"encoding/binary"
+	"math"
 	"reflect"
 	"testing"
 	"time"
@@ -254,3 +256,50 @@ func TestRecordBatchDecoding(t *testing.T) {
 		}
 	}
 }
+
+func TestRecordBatchLargeNumRecords(t *testing.T) {
+	numOfRecords := 10 + (2 * math.MaxUint16)
+	numofRecordsBytes := make([]byte, 4)
+	binary.BigEndian.PutUint32(numofRecordsBytes, uint32(numOfRecords))
+
+	encodedBatch := []byte{
+		0, 0, 0, 0, 0, 0, 0, 0, // First Offset
+		0, 42, 0, 250, // Length
+		0, 0, 0, 0, // Partition Leader Epoch
+		2,                 // Version
+		103, 68, 166, 213, // CRC
+		0, 0, // Attributes
+		0, 0, 0, 0, // Last Offset Delta
+		0, 0, 1, 88, 141, 205, 89, 56, // First Timestamp
+		0, 0, 0, 0, 0, 0, 0, 0, // Max Timestamp
+		0, 0, 0, 0, 0, 0, 0, 0, // Producer ID
+		0, 0, // Producer Epoch
+		0, 0, 0, 0, // First Sequence
+	}
+
+	encodedBatch = append(encodedBatch, numofRecordsBytes...)
+
+	for range numOfRecords {
+		encodedBatch = append(encodedBatch, []byte{
+			40, // Record Length
+			0,  // Attributes
+			10, // Timestamp Delta
+			0,  // Offset Delta
+			8,  // Key Length
+			1, 2, 3, 4,
+			6, // Value Length
+			5, 6, 7,
+			2,        // Number of Headers
+			6,        // Header Key Length
+			8, 9, 10, // Header Key
+			4,      // Header Value Length
+			11, 12, // Header Value
+		}...)
+	}
+
+	var batch RecordBatch
+	err := decode(encodedBatch, &batch, nil)
+	if err != nil {
+		t.Fatal("received error while decoding record batch", err)
+	}
+}