fix Kafka record size calculation

Author: marle3003

kafka/record.go

@@ -81,6 +81,7 @@ func (r *Record) Size(baseOffSet int64, base time.Time) int {
 	if r.Value != nil {
 		valueLength = r.Value.Size()
 	}
+
 	size := 1 + // attribute
 		sizeVarInt(deltaTimestamp) +
 		sizeVarInt(deltaOffset) +
@@ -229,8 +230,8 @@ func sizeVarInt(x int64) int {
 		ux = ^ux
 	}
 	i := 0
-	for x >= 0x80 {
-		x >>= 7
+	for ux >= 0x80 {
+		ux >>= 7
 		i++
 	}
 	return i + 1

kafka/record_test.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"bytes"
 	"github.com/stretchr/testify/require"
+	"strings"
 	"testing"
 	"time"
 )
@@ -346,3 +347,19 @@ func TestRecordBatch_WriteTo_Bytes_Compare(t *testing.T) {
 
 	require.Equal(t, expected, b)
 }
+
+func TestSize_LongValue(t *testing.T) {
+	// 80 because it must be lower than 128 but the value * 2 must be >= 128 (x<<1 is equal to *2)
+	val := []byte(strings.Repeat("b", 80))
+	records := RecordBatch{Records: []*Record{
+		{
+			Offset:  0,
+			Time:    ToTime(1657010762684),
+			Key:     NewBytes([]byte("foo")),
+			Value:   NewBytes(val),
+			Headers: nil,
+		},
+	}}
+
+	require.Equal(t, 150, records.Size())
+}