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

packet_decoder.go

@@ -15,6 +15,7 @@ type packetDecoder interface {
 	getUVarint() (uint64, error)
 	getFloat64() (float64, error)
 	getArrayLength() (int, error)
+	getArrayLengthNoLimit() (int, error)
 	getCompactArrayLength() (int, error)
 	getBool() (bool, error)
 	getEmptyTaggedFieldArray() (int, error)

real_decoder.go

@@ -121,6 +121,20 @@ func (rd *realDecoder) getArrayLength() (int, error) {
 	return tmp, nil
 }
 
+func (rd *realDecoder) getArrayLengthNoLimit() (int, error) {
+	if rd.remaining() < 4 {
+		rd.off = len(rd.raw)
+		return -1, ErrInsufficientData
+	}
+	tmp := int(int32(binary.BigEndian.Uint32(rd.raw[rd.off:])))
+	rd.off += 4
+	if tmp > rd.remaining() {
+		rd.off = len(rd.raw)
+		return -1, ErrInsufficientData
+	}
+	return tmp, nil
+}
+
 func (rd *realDecoder) getCompactArrayLength() (int, error) {
 	n, err := rd.getUVarint()
 	if err != nil {

record_batch.go

@@ -157,7 +157,11 @@ func (b *RecordBatch) decode(pd packetDecoder) (err error) {
 		return err
 	}
 
-	numRecs, err := pd.getArrayLength()
+	// Using NoLimit because a single record batch could contain
+	// more then 2*math.MaxUint16 records. The packet decoder will
+	// check to make sure the array is not greater than the
+	// remaining bytes.
+	numRecs, err := pd.getArrayLengthNoLimit()
 	if err != nil {
 		return err
 	}

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)
+	}
+}