fix(protocol): handle ElectLeaders V1 non-flexible headers

elect_leaders_request.go

@@ -96,7 +96,10 @@ func (r *ElectLeadersRequest) version() int16 {
 }
 
 func (r *ElectLeadersRequest) headerVersion() int16 {
-	return 2
+	if r.isFlexible() {
+		return 2
+	}
+	return 1
 }
 
 func (r *ElectLeadersRequest) isValidVersion() bool {

elect_leaders_request_test.go

@@ -2,7 +2,10 @@
 
 package sarama
 
-import "testing"
+import (
+	"bytes"
+	"testing"
+)
 
 var (
 	electLeadersRequestOneTopicV1 = []byte{
@@ -40,3 +43,58 @@ func TestElectLeadersRequest(t *testing.T) {
 	request.Version = 2
 	testRequest(t, "one topic V2", request, electLeadersRequestOneTopicV2)
 }
+
+func TestElectLeadersRequestHeaderVersion(t *testing.T) {
+	reqBody := &ElectLeadersRequest{
+		TimeoutMs: int32(10000),
+		Version:   int16(1),
+		TopicPartitions: map[string][]int32{
+			"topic": {0},
+		},
+		Type: PreferredElection,
+	}
+
+	packet, err := encode(&request{
+		correlationID: 123,
+		clientID:      "foo",
+		body:          reqBody,
+	}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	nonFlexibleHeaderSize := 14 + len("foo")
+	if !bytes.Equal(packet[nonFlexibleHeaderSize:], electLeadersRequestOneTopicV1) {
+		t.Fatalf("expected V1 body to start immediately after request header")
+	}
+
+	reqBody.Version = 2
+	packet, err = encode(&request{
+		correlationID: 123,
+		clientID:      "foo",
+		body:          reqBody,
+	}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	flexibleHeaderSize := nonFlexibleHeaderSize + 1
+	if packet[nonFlexibleHeaderSize] != 0 {
+		t.Fatalf("expected V2 flexible request header to include empty tagged fields, got byte %d", packet[nonFlexibleHeaderSize])
+	}
+	if !bytes.Equal(packet[flexibleHeaderSize:], electLeadersRequestOneTopicV2) {
+		t.Fatalf("expected V2 body to start after flexible request header")
+	}
+}
+
+func TestElectLeadersResponseHeaderVersion(t *testing.T) {
+	response := &ElectLeadersResponse{Version: 1}
+	if response.headerVersion() != 0 {
+		t.Fatalf("expected V1 response header version 0, got %d", response.headerVersion())
+	}
+
+	response.Version = 2
+	if response.headerVersion() != 1 {
+		t.Fatalf("expected V2 response header version 1, got %d", response.headerVersion())
+	}
+}

elect_leaders_response.go

@@ -128,7 +128,10 @@ func (r *ElectLeadersResponse) version() int16 {
 }
 
 func (r *ElectLeadersResponse) headerVersion() int16 {
-	return 1
+	if r.isFlexible() {
+		return 1
+	}
+	return 0
 }
 
 func (r *ElectLeadersResponse) isValidVersion() bool {

functional_admin_test.go

@@ -432,6 +432,51 @@ func TestFuncAdminDeleteTopic(t *testing.T) {
 	}
 }
 
+// TestFuncAdminElectLeadersV1 covers low-version client compatibility against a
+// newer broker for the ElectLeaders V1 non-flexible path. See
+// https://github.com/IBM/sarama/pull/3312.
+func TestFuncAdminElectLeadersV1(t *testing.T) {
+	checkKafkaVersion(t, "2.3.0.0")
+	setupFunctionalTest(t)
+	defer teardownFunctionalTest(t)
+
+	config := NewFunctionalTestConfig()
+	// Force ElectLeaders onto request/response version 1 so the broker emits
+	// the non-flexible wire format that previously regressed.
+	config.Version = V2_3_0_0
+
+	adminClient, err := NewClusterAdmin(FunctionalTestEnv.KafkaBrokerAddrs, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer safeClose(t, adminClient)
+
+	results, err := adminClient.ElectLeaders(PreferredElection, map[string][]int32{"test.1": {0}})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	topicResults, ok := results["test.1"]
+	if !ok {
+		t.Fatalf("topic test.1 missing in response: %#v", results)
+	}
+
+	partitionResult, ok := topicResults[0]
+	if !ok {
+		t.Fatalf("partition 0 missing in response: %#v", topicResults)
+	}
+
+	if partitionResult == nil {
+		t.Fatal("partition 0 returned nil result")
+	}
+
+	switch partitionResult.ErrorCode {
+	case ErrNoError, ErrElectionNotNeeded:
+	default:
+		t.Fatalf("expected partition 0 to return a decodable ElectLeaders result, got %v (%v)", partitionResult.ErrorCode, partitionResult.ErrorMessage)
+	}
+}
+
 func TestFuncAdminIncrementalAlterConfigs(t *testing.T) {
 	checkKafkaVersion(t, "2.3.0.0")
 	setupFunctionalTest(t)