refactor: remove getEncodedLength methods and use utils.GetEncodedLength for encoding length calculation

Author: ryan-gang

protocol/api/record_batch.go

@@ -8,19 +8,12 @@ import (
 	"github.com/codecrafters-io/kafka-tester/protocol/decoder"
 	"github.com/codecrafters-io/kafka-tester/protocol/encoder"
 	"github.com/codecrafters-io/kafka-tester/protocol/errors"
+	"github.com/codecrafters-io/kafka-tester/protocol/utils"
 	"github.com/codecrafters-io/tester-utils/logger"
 )
 
 type RecordBatches []RecordBatch
 
-func (rbs RecordBatches) getEncodedLength() int {
-	encodedLength := 0
-	for _, rb := range rbs {
-		encodedLength += rb.getEncodedLength()
-	}
-	return encodedLength
-}
-
 func (rbs RecordBatches) Encode(pe *encoder.Encoder) {
 	for _, rb := range rbs {
 		rb.Encode(pe)
@@ -76,13 +69,6 @@ func (rb RecordBatch) Encode(pe *encoder.Encoder) {
 	pe.PutInt32At(int32(computedChecksum), crcStartOffset, 4)
 }
 
-func (rb RecordBatch) getEncodedLength() int {
-	encoder := encoder.Encoder{}
-	encoder.Init(make([]byte, 1024))
-	rb.Encode(&encoder)
-	return encoder.Offset()
-}
-
 func (rb *RecordBatch) Decode(pd *decoder.Decoder, logger *logger.Logger, indentation int) (err error) {
 	if rb.BaseOffset, err = pd.GetInt64(); err != nil {
 		if decodingErr, ok := err.(*errors.PacketDecodingError); ok {
@@ -234,7 +220,7 @@ type Record struct {
 }
 
 func (r Record) Encode(pe *encoder.Encoder) {
-	pe.PutVarint(int64(r.getEncodedLength())) // Length placeholder
+	pe.PutVarint(int64(utils.GetEncodedLength(r))) // Length placeholder
 	// As this is variable length, we can't use placeholders and update later reliably.
 	// We need to have a value, close to the actual value, such that it takes the same space
 	// This is an approx value, the actual value will be computed at the end
@@ -254,28 +240,6 @@ func (r Record) Encode(pe *encoder.Encoder) {
 	}
 }
 
-func (r Record) getEncodedLength() int {
-	encoder := encoder.Encoder{}
-	encoder.Init(make([]byte, 1024))
-
-	encoder.PutInt8(r.Attributes)
-	encoder.PutVarint(r.TimestampDelta)
-	encoder.PutVarint(int64(r.OffsetDelta))
-	if string(r.Key) == "null" {
-		encoder.PutCompactBytes([]byte{})
-	} else {
-		encoder.PutCompactBytes(r.Key)
-	}
-	encoder.PutVarint(int64(len(r.Value)))
-	encoder.PutRawBytes(r.Value)
-	encoder.PutVarint(int64(len(r.Headers)))
-	for _, header := range r.Headers {
-		header.Encode(&encoder)
-	}
-
-	return encoder.Offset()
-}
-
 func (r *Record) Decode(pd *decoder.Decoder, logger *logger.Logger, indentation int) (err error) {
 	length, err := pd.GetSignedVarint()
 	if err != nil {

protocol/serializer/utils.go

@@ -6,20 +6,9 @@ import (
 	"os"
 	"strings"
 
-	"github.com/codecrafters-io/kafka-tester/protocol/encoder"
-	kafka_interface "github.com/codecrafters-io/kafka-tester/protocol/interface"
 	"github.com/google/uuid"
 )
 
-func GetEncodedBytes(encodableObject kafka_interface.Encodable) []byte {
-	encoder := encoder.Encoder{}
-	encoder.Init(make([]byte, 1024))
-	encodableObject.Encode(&encoder)
-	encodedBytes := encoder.Bytes()[:encoder.Offset()]
-
-	return encodedBytes
-}
-
 func generateDirectories(paths []string) error {
 	for _, path := range paths {
 		err := generateDirectory(path)