Merge pull request #1038 from Shopify/simplify-record-batch-encoding

Simplify record batch encoding a bit

record_batch.go

@@ -208,21 +208,15 @@ func (b *RecordBatch) decode(pd packetDecoder) (err error) {
 
 func (b *RecordBatch) encodeRecords(pe packetEncoder) error {
 	var raw []byte
-	if b.Codec != CompressionNone {
-		var err error
-		if raw, err = encode(recordsArray(b.Records), nil); err != nil {
-			return err
-		}
-		b.recordsLen = len(raw)
+	var err error
+	if raw, err = encode(recordsArray(b.Records), pe.metricRegistry()); err != nil {
+		return err
 	}
+	b.recordsLen = len(raw)
 
 	switch b.Codec {
 	case CompressionNone:
-		offset := pe.offset()
-		if err := recordsArray(b.Records).encode(pe); err != nil {
-			return err
-		}
-		b.recordsLen = pe.offset() - offset
+		b.compressedRecords = raw
 	case CompressionGZIP:
 		var buf bytes.Buffer
 		writer := gzip.NewWriter(&buf)