proper test for large record batch length

Signed-off-by: Ryan Belgrave <[email protected]>

Author: rmb938

record_test.go

@@ -3,7 +3,8 @@
 package sarama
 
 import (
-	"fmt"
+	"encoding/binary"
+	"math"
 	"reflect"
 	"testing"
 	"time"
@@ -256,39 +257,49 @@ func TestRecordBatchDecoding(t *testing.T) {
 	}
 }
 
-func TestRecordBatchInvalidNumRecords(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, 0, 0, 70, // Length
+		0, 42, 0, 250, // Length
 		0, 0, 0, 0, // Partition Leader Epoch
-		2,               // Version
-		91, 48, 202, 99, // CRC
+		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
-		0, 1, 255, 255, // Number of Records - 1 + 2*math.MaxUint16
-		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
 	}
 
-	batch := RecordBatch{}
+	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 != ErrInsufficientData {
-		t.Fatal(fmt.Errorf("was suppose to get ErrInsufficientData, instead got: %w", err))
+	if err != nil {
+		t.Fatal("received error while decoding record batch", err)
 	}
 }