Merge pull request #1898 from mmaslankaprv/fixed-parsing-control-reco…

…rd-value

Parsing only known control batches value

control_record.go

@@ -23,16 +23,6 @@ type ControlRecord struct {
 
 func (cr *ControlRecord) decode(key, value packetDecoder) error {
 	var err error
-	cr.Version, err = value.getInt16()
-	if err != nil {
-		return err
-	}
-
-	cr.CoordinatorEpoch, err = value.getInt32()
-	if err != nil {
-		return err
-	}
-
 	// There a version for the value part AND the key part. And I have no idea if they are supposed to match or not
 	// Either way, all these version can only be 0 for now
 	cr.Version, err = key.getInt16()
@@ -55,6 +45,18 @@ func (cr *ControlRecord) decode(key, value packetDecoder) error {
 		// UNKNOWN is used to indicate a control type which the client is not aware of and should be ignored
 		cr.Type = ControlRecordUnknown
 	}
+	// we want to parse value only if we are decoding control record of known type
+	if cr.Type != ControlRecordUnknown {
+		cr.Version, err = value.getInt16()
+		if err != nil {
+			return err
+		}
+
+		cr.CoordinatorEpoch, err = value.getInt32()
+		if err != nil {
+			return err
+		}
+	}
 	return nil
 }
 

control_record_test.go

@@ -1 +1,67 @@
 package sarama
+
+import (
+	"testing"
+)
+
+var (
+	abortTxCtrlRecKey = []byte{
+		0, 0, // version
+		0, 0, // TX_ABORT = 0
+	}
+	abortTxCtrlRecValue = []byte{
+		0, 0, // version
+		0, 0, 0, 10, // coordinator epoch
+	}
+	commitTxCtrlRecKey = []byte{
+		0, 0, // version
+		0, 1, // TX_COMMIT = 1
+	}
+	commitTxCtrlRecValue = []byte{
+		0, 0, // version
+		0, 0, 0, 15, // coordinator epoch
+	}
+	unknownCtrlRecKey = []byte{
+		0, 0, // version
+		0, 128, // UNKNOWN = -1
+	}
+	// empty value for unknown record
+	unknownCtrlRecValue = []byte{}
+)
+
+func testDecode(t *testing.T, tp string, key []byte, value []byte) ControlRecord {
+	controlRecord := ControlRecord{}
+	err := controlRecord.decode(&realDecoder{raw: key}, &realDecoder{raw: value})
+	if err != nil {
+		t.Error("Decoding control record of type " + tp + " failed")
+		return ControlRecord{}
+	}
+	return controlRecord
+}
+
+func assertRecordType(t *testing.T, r *ControlRecord, expected ControlRecordType) {
+	if r.Type != expected {
+		t.Errorf("control record type mismatch, expected: %v, have %v", expected, r.Type)
+	}
+}
+
+func TestDecodingControlRecords(t *testing.T) {
+	abortTx := testDecode(t, "abort transaction", abortTxCtrlRecKey, abortTxCtrlRecValue)
+
+	assertRecordType(t, &abortTx, ControlRecordAbort)
+
+	if abortTx.CoordinatorEpoch != 10 {
+		t.Errorf("abort tx control record coordinator epoch mismatch")
+	}
+
+	commitTx := testDecode(t, "commit transaction", commitTxCtrlRecKey, commitTxCtrlRecValue)
+
+	if commitTx.CoordinatorEpoch != 15 {
+		t.Errorf("commit tx control record coordinator epoch mismatch")
+	}
+	assertRecordType(t, &commitTx, ControlRecordCommit)
+
+	unknown := testDecode(t, "unknown", unknownCtrlRecKey, unknownCtrlRecValue)
+
+	assertRecordType(t, &unknown, ControlRecordUnknown)
+}