Fix describegroups in 2.3.1 case

Author: yolken-segment

describegroups.go

@@ -316,6 +316,9 @@ func (t *describeGroupsResponseV0) readFrom(r *bufio.Reader, sz int) (remain int
 
 // decodeMemberMetadata converts raw metadata bytes to a
 // DescribeGroupsResponseMemberMetadata struct.
+//
+// See https://github.com/apache/kafka/blob/2.4/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java#L49
+// for protocol details.
 func decodeMemberMetadata(rawMetadata []byte) (DescribeGroupsResponseMemberMetadata, error) {
 	mm := DescribeGroupsResponseMemberMetadata{}
 

describegroups_test.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"os"
 	"reflect"
 	"sort"
 	"testing"
@@ -54,6 +55,14 @@ func TestDescribeGroupsResponseV0(t *testing.T) {
 }
 
 func TestClientDescribeGroups(t *testing.T) {
+	if os.Getenv("KAFKA_VERSION") == "2.3.1" {
+		// There's a bug in 2.3.1 that causes the MemberMetadata to be in the wrong format and thus
+		// leads to an error when decoding the DescribeGroupsResponse.
+		//
+		// See https://issues.apache.org/jira/browse/KAFKA-9150 for details.
+		t.Skip("Skipping because kafka version is 2.3.1")
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
@@ -76,6 +85,7 @@ func TestClientDescribeGroups(t *testing.T) {
 			Value: []byte("value"),
 		},
 	)
+
 	if err != nil {
 		t.Fatal(err)
 	}