Merge pull request #2247 from aiquestion/fix_describe_group

fix(protocol): move AuthorizedOperations into GroupDescription of DescribeGroupsResponse

Author: dnwe

describe_groups_response.go

@@ -1,10 +1,9 @@
 package sarama
 
 type DescribeGroupsResponse struct {
-	Version              int16
-	ThrottleTimeMs       int32
-	Groups               []*GroupDescription
-	AuthorizedOperations int32
+	Version        int16
+	ThrottleTimeMs int32
+	Groups         []*GroupDescription
 }
 
 func (r *DescribeGroupsResponse) encode(pe packetEncoder) error {
@@ -21,9 +20,6 @@ func (r *DescribeGroupsResponse) encode(pe packetEncoder) error {
 			return err
 		}
 	}
-	if r.Version >= 3 {
-		pe.putInt32(r.AuthorizedOperations)
-	}
 
 	return nil
 }
@@ -48,11 +44,6 @@ func (r *DescribeGroupsResponse) decode(pd packetDecoder, version int16) (err er
 			return err
 		}
 	}
-	if r.Version >= 3 {
-		if r.AuthorizedOperations, err = pd.getInt32(); err != nil {
-			return err
-		}
-	}
 
 	return nil
 }
@@ -80,12 +71,13 @@ func (r *DescribeGroupsResponse) requiredVersion() KafkaVersion {
 type GroupDescription struct {
 	Version int16
 
-	Err          KError
-	GroupId      string
-	State        string
-	ProtocolType string
-	Protocol     string
-	Members      map[string]*GroupMemberDescription
+	Err                  KError
+	GroupId              string
+	State                string
+	ProtocolType         string
+	Protocol             string
+	Members              map[string]*GroupMemberDescription
+	AuthorizedOperations int32
 }
 
 func (gd *GroupDescription) encode(pe packetEncoder) error {
@@ -119,6 +111,10 @@ func (gd *GroupDescription) encode(pe packetEncoder) error {
 		}
 	}
 
+	if gd.Version >= 3 {
+		pe.putInt32(gd.AuthorizedOperations)
+	}
+
 	return nil
 }
 
@@ -147,20 +143,25 @@ func (gd *GroupDescription) decode(pd packetDecoder) (err error) {
 	if err != nil {
 		return err
 	}
-	if n == 0 {
-		return nil
-	}
 
-	gd.Members = make(map[string]*GroupMemberDescription)
-	for i := 0; i < n; i++ {
-		memberId, err := pd.getString()
-		if err != nil {
-			return err
+	if n > 0 {
+		gd.Members = make(map[string]*GroupMemberDescription)
+		for i := 0; i < n; i++ {
+			memberId, err := pd.getString()
+			if err != nil {
+				return err
+			}
+
+			gd.Members[memberId] = new(GroupMemberDescription)
+			gd.Members[memberId].Version = gd.Version
+			if err := gd.Members[memberId].decode(pd); err != nil {
+				return err
+			}
 		}
+	}
 
-		gd.Members[memberId] = new(GroupMemberDescription)
-		gd.Members[memberId].Version = gd.Version
-		if err := gd.Members[memberId].decode(pd); err != nil {
+	if gd.Version >= 3 {
+		if gd.AuthorizedOperations, err = pd.getInt32(); err != nil {
 			return err
 		}
 	}

describe_groups_response_test.go

@@ -95,7 +95,6 @@ var (
 	describeGroupsResponseEmptyV3 = []byte{
 		0, 0, 0, 0, // throttle time 0
 		0, 0, 0, 0, // no groups
-		0, 0, 0, 0, // authorizedOperations 0
 	}
 
 	describeGroupsResponsePopulatedV3 = []byte{
@@ -113,21 +112,21 @@ var (
 		0, 9, 'l', 'o', 'c', 'a', 'l', 'h', 'o', 's', 't', // Client Host
 		0, 0, 0, 3, 0x01, 0x02, 0x03, // MemberMetadata
 		0, 0, 0, 3, 0x04, 0x05, 0x06, // MemberAssignment
+		0, 0, 0, 0, // authorizedOperations 0
 
 		0, 30, // ErrGroupAuthorizationFailed
 		0, 0,
 		0, 0,
 		0, 0,
 		0, 0,
 		0, 0, 0, 0,
-
 		0, 0, 0, 0, // authorizedOperations 0
+
 	}
 
 	describeGroupsResponseEmptyV4 = []byte{
 		0, 0, 0, 0, // throttle time 0
 		0, 0, 0, 0, // no groups
-		0, 0, 0, 0, // authorizedOperations 0
 	}
 
 	describeGroupsResponsePopulatedV4 = []byte{
@@ -146,15 +145,16 @@ var (
 		0, 9, 'l', 'o', 'c', 'a', 'l', 'h', 'o', 's', 't', // Client Host
 		0, 0, 0, 3, 0x01, 0x02, 0x03, // MemberMetadata
 		0, 0, 0, 3, 0x04, 0x05, 0x06, // MemberAssignment
+		0, 0, 0, 0, // authorizedOperations 0
 
 		0, 30, // ErrGroupAuthorizationFailed
 		0, 0,
 		0, 0,
 		0, 0,
 		0, 0,
 		0, 0, 0, 0,
-
 		0, 0, 0, 0, // authorizedOperations 0
+
 	}
 )
 
@@ -171,10 +171,9 @@ func TestDescribeGroupsResponseV1plus(t *testing.T) {
 			3,
 			describeGroupsResponseEmptyV3,
 			&DescribeGroupsResponse{
-				Version:              3,
-				ThrottleTimeMs:       int32(0),
-				Groups:               []*GroupDescription{},
-				AuthorizedOperations: int32(0),
+				Version:        3,
+				ThrottleTimeMs: int32(0),
+				Groups:         []*GroupDescription{},
 			},
 		},
 		{
@@ -212,18 +211,16 @@ func TestDescribeGroupsResponseV1plus(t *testing.T) {
 						Members:      nil,
 					},
 				},
-				AuthorizedOperations: int32(0),
 			},
 		},
 		{
 			"empty",
 			4,
 			describeGroupsResponseEmptyV4,
 			&DescribeGroupsResponse{
-				Version:              4,
-				ThrottleTimeMs:       int32(0),
-				Groups:               []*GroupDescription{},
-				AuthorizedOperations: int32(0),
+				Version:        4,
+				ThrottleTimeMs: int32(0),
+				Groups:         []*GroupDescription{},
 			},
 		},
 		{
@@ -262,7 +259,6 @@ func TestDescribeGroupsResponseV1plus(t *testing.T) {
 						Members:      nil,
 					},
 				},
-				AuthorizedOperations: int32(0),
 			},
 		},
 	}

functional_admin_test.go

@@ -123,3 +123,59 @@ func TestFuncAdminQuotas(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestFuncAdminDescribeGroups(t *testing.T) {
+	checkKafkaVersion(t, "2.3.0.0")
+	setupFunctionalTest(t)
+	defer teardownFunctionalTest(t)
+
+	group1 := testFuncConsumerGroupID(t)
+	group2 := testFuncConsumerGroupID(t)
+
+	kafkaVersion, err := ParseKafkaVersion(FunctionalTestEnv.KafkaVersion)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	config := NewTestConfig()
+	config.Version = kafkaVersion
+	adminClient, err := NewClusterAdmin(FunctionalTestEnv.KafkaBrokerAddrs, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	config1 := NewTestConfig()
+	config1.ClientID = "M1"
+	config1.Version = V2_3_0_0
+	config1.Consumer.Offsets.Initial = OffsetNewest
+	m1 := runTestFuncConsumerGroupMemberWithConfig(t, config1, group1, 100, nil, "test.4")
+	defer m1.Close()
+
+	config2 := NewTestConfig()
+	config2.ClientID = "M2"
+	config2.Version = V2_3_0_0
+	config2.Consumer.Offsets.Initial = OffsetNewest
+	config2.Consumer.Group.InstanceId = "Instance2"
+	m2 := runTestFuncConsumerGroupMemberWithConfig(t, config2, group2, 100, nil, "test.4")
+	defer m2.Close()
+
+	m1.WaitForState(2)
+	m2.WaitForState(2)
+
+	res, err := adminClient.DescribeConsumerGroups([]string{group1, group2})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(res) != 2 {
+		t.Errorf("group description should be 2, got %v\n", len(res))
+	}
+	if len(res[0].Members) != 1 {
+		t.Errorf("should have 1 members in group , got %v\n", len(res[0].Members))
+	}
+	if len(res[1].Members) != 1 {
+		t.Errorf("should have 1 members in group , got %v\n", len(res[1].Members))
+	}
+
+	m1.AssertCleanShutdown()
+	m2.AssertCleanShutdown()
+}