Skip to content

Commit

Permalink
fix(protocol): tidyup DescribeGroupsResponse
Browse files Browse the repository at this point in the history 
As we currently expose this directly in the response from admin.go it
makes sense to document the struct fields using the text included in the
kafka json protocol description. Whilst checking the protocol
implementation against the specs, I worked these in.
  • Loading branch information
@dnwe
dnwe committed Jun 7, 2022
1 parent 9bf344f commit 41bea2e
Show file tree
Hide file tree
Showing 2 changed files with 111 additions and 93 deletions.
156 changes: 90 additions & 66 deletions describe_groups_response.go
View file
Original file line number Diff line number Diff line change
@@ -1,22 +1,26 @@
package sarama

type DescribeGroupsResponse struct {
Version int16
// Version defines the protocol version to use for encode and decode
Version int16
// ThrottleTimeMs contains the duration in milliseconds for which the
// request was throttled due to a quota violation, or zero if the request
// did not violate any quota.
ThrottleTimeMs int32
Groups []*GroupDescription
// Groups contains each described group.
Groups []*GroupDescription
}

func (r *DescribeGroupsResponse) encode(pe packetEncoder) error {
func (r *DescribeGroupsResponse) encode(pe packetEncoder) (err error) {
if r.Version >= 1 {
pe.putInt32(r.ThrottleTimeMs)
}
if err := pe.putArrayLength(len(r.Groups)); err != nil {
return err
}

for _, groupDescription := range r.Groups {
groupDescription.Version = r.Version
if err := groupDescription.encode(pe); err != nil {
for _, block := range r.Groups {
if err := block.encode(pe, r.Version); err != nil {
return err
}
}
Expand All @@ -31,17 +35,16 @@ func (r *DescribeGroupsResponse) decode(pd packetDecoder, version int16) (err er
return err
}
}
n, err := pd.getArrayLength()
if err != nil {
if numGroups, err := pd.getArrayLength(); err != nil {
return err
}

r.Groups = make([]*GroupDescription, n)
for i := 0; i < n; i++ {
r.Groups[i] = new(GroupDescription)
r.Groups[i].Version = r.Version
if err := r.Groups[i].decode(pd); err != nil {
return err
} else if numGroups > 0 {
r.Groups = make([]*GroupDescription, numGroups)
for i := 0; i < numGroups; i++ {
block := &GroupDescription{}
if err := block.decode(pd, r.Version); err != nil {
return err
}
r.Groups[i] = block
}
}

Expand All @@ -68,20 +71,32 @@ func (r *DescribeGroupsResponse) requiredVersion() KafkaVersion {
return V0_9_0_0
}

// GroupDescription contains each described group.
type GroupDescription struct {
// Version defines the protocol version to use for encode and decode
Version int16

Err KError
GroupId string
State string
ProtocolType string
Protocol string
Members map[string]*GroupMemberDescription
// Err contains the describe error as the KError type.
Err KError
// ErrorCode contains the describe error, or 0 if there was no error.
ErrorCode int16
// GroupId contains the group ID string.
GroupId string
// State contains the group state string, or the empty string.
State string
// ProtocolType contains the group protocol type, or the empty string.
ProtocolType string
// Protocol contains the group protocol data, or the empty string.
Protocol string
// Members contains the group members.
Members map[string]*GroupMemberDescription
// AuthorizedOperations contains a 32-bit bitfield to represent authorized
// operations for this group.
AuthorizedOperations int32
}

func (gd *GroupDescription) encode(pe packetEncoder) error {
pe.putInt16(int16(gd.Err))
func (gd *GroupDescription) encode(pe packetEncoder, version int16) (err error) {
gd.Version = version
pe.putInt16(gd.ErrorCode)

if err := pe.putString(gd.GroupId); err != nil {
return err
Expand All @@ -100,13 +115,8 @@ func (gd *GroupDescription) encode(pe packetEncoder) error {
return err
}

for memberId, groupMemberDescription := range gd.Members {
if err := pe.putString(memberId); err != nil {
return err
}
// encode with version
groupMemberDescription.Version = gd.Version
if err := groupMemberDescription.encode(pe); err != nil {
for _, block := range gd.Members {
if err := block.encode(pe, gd.Version); err != nil {
return err
}
}
Expand All @@ -118,44 +128,38 @@ func (gd *GroupDescription) encode(pe packetEncoder) error {
return nil
}

func (gd *GroupDescription) decode(pd packetDecoder) (err error) {
kerr, err := pd.getInt16()
if err != nil {
func (gd *GroupDescription) decode(pd packetDecoder, version int16) (err error) {
gd.Version = version
if gd.ErrorCode, err = pd.getInt16(); err != nil {
return err
}

gd.Err = KError(kerr)
gd.Err = KError(gd.ErrorCode)

if gd.GroupId, err = pd.getString(); err != nil {
return
return err
}
if gd.State, err = pd.getString(); err != nil {
return
return err
}
if gd.ProtocolType, err = pd.getString(); err != nil {
return
return err
}
if gd.Protocol, err = pd.getString(); err != nil {
return
}

n, err := pd.getArrayLength()
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 {
if numMembers, err := pd.getArrayLength(); err != nil {
return err
} else if numMembers > 0 {
gd.Members = make(map[string]*GroupMemberDescription, numMembers)
for i := 0; i < numMembers; i++ {
block := &GroupMemberDescription{}
if err := block.decode(pd, gd.Version); 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
if block != nil {
gd.Members[block.MemberId] = block
}
}
}
Expand All @@ -169,17 +173,33 @@ func (gd *GroupDescription) decode(pd packetDecoder) (err error) {
return nil
}

// GroupMemberDescription contains the group members.
type GroupMemberDescription struct {
// Version defines the protocol version to use for encode and decode
Version int16

GroupInstanceId *string
ClientId string
ClientHost string
MemberMetadata []byte
// MemberId contains the member ID assigned by the group coordinator.
MemberId string
// GroupInstanceId contains the unique identifier of the consumer instance
// provided by end user.
GroupInstanceId *string
// ClientId contains the client ID used in the member's latest join group
// request.
ClientId string
// ClientHost contains the client host.
ClientHost string
// MemberMetadata contains the metadata corresponding to the current group
// protocol in use.
MemberMetadata []byte
// MemberAssignment contains the current assignment provided by the group
// leader.
MemberAssignment []byte
}

func (gmd *GroupMemberDescription) encode(pe packetEncoder) error {
func (gmd *GroupMemberDescription) encode(pe packetEncoder, version int16) (err error) {
gmd.Version = version
if err := pe.putString(gmd.MemberId); err != nil {
return err
}
if gmd.Version >= 4 {
if err := pe.putNullableString(gmd.GroupInstanceId); err != nil {
return err
Expand All @@ -201,23 +221,27 @@ func (gmd *GroupMemberDescription) encode(pe packetEncoder) error {
return nil
}

func (gmd *GroupMemberDescription) decode(pd packetDecoder) (err error) {
func (gmd *GroupMemberDescription) decode(pd packetDecoder, version int16) (err error) {
gmd.Version = version
if gmd.MemberId, err = pd.getString(); err != nil {
return err
}
if gmd.Version >= 4 {
if gmd.GroupInstanceId, err = pd.getNullableString(); err != nil {
return
return err
}
}
if gmd.ClientId, err = pd.getString(); err != nil {
return
return err
}
if gmd.ClientHost, err = pd.getString(); err != nil {
return
return err
}
if gmd.MemberMetadata, err = pd.getBytes(); err != nil {
return
return err
}
if gmd.MemberAssignment, err = pd.getBytes(); err != nil {
return
return err
}

return nil
Expand Down
48 changes: 21 additions & 27 deletions describe_groups_response_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@ import (
"errors"
"reflect"
"testing"

"github.com/stretchr/testify/assert"
)

var (
Expand Down Expand Up @@ -161,7 +163,7 @@ var (
func TestDescribeGroupsResponseV1plus(t *testing.T) {
groupInstanceId := "gid"
tests := []struct {
CaseName string
Name string
Version int16
MessageBytes []byte
Message *DescribeGroupsResponse
Expand All @@ -171,9 +173,7 @@ func TestDescribeGroupsResponseV1plus(t *testing.T) {
3,
describeGroupsResponseEmptyV3,
&DescribeGroupsResponse{
Version: 3,
ThrottleTimeMs: int32(0),
Groups: []*GroupDescription{},
Version: 3,
},
},
{
Expand All @@ -194,6 +194,7 @@ func TestDescribeGroupsResponseV1plus(t *testing.T) {
Members: map[string]*GroupMemberDescription{
"id": {
Version: 3,
MemberId: "id",
ClientId: "sarama",
ClientHost: "localhost",
MemberMetadata: []byte{1, 2, 3},
Expand All @@ -202,13 +203,9 @@ func TestDescribeGroupsResponseV1plus(t *testing.T) {
},
},
{
Version: 3,
Err: KError(30),
GroupId: "",
State: "",
ProtocolType: "",
Protocol: "",
Members: nil,
Version: 3,
Err: KError(30),
ErrorCode: 30,
},
},
},
Expand All @@ -218,9 +215,7 @@ func TestDescribeGroupsResponseV1plus(t *testing.T) {
4,
describeGroupsResponseEmptyV4,
&DescribeGroupsResponse{
Version: 4,
ThrottleTimeMs: int32(0),
Groups: []*GroupDescription{},
Version: 4,
},
},
{
Expand All @@ -241,6 +236,7 @@ func TestDescribeGroupsResponseV1plus(t *testing.T) {
Members: map[string]*GroupMemberDescription{
"id": {
Version: 4,
MemberId: "id",
GroupInstanceId: &groupInstanceId,
ClientId: "sarama",
ClientHost: "localhost",
Expand All @@ -250,25 +246,23 @@ func TestDescribeGroupsResponseV1plus(t *testing.T) {
},
},
{
Version: 4,
Err: KError(30),
GroupId: "",
State: "",
ProtocolType: "",
Protocol: "",
Members: nil,
Version: 4,
Err: KError(30),
ErrorCode: 30,
},
},
},
},
}

for _, c := range tests {
response := new(DescribeGroupsResponse)
testVersionDecodable(t, c.CaseName, response, c.MessageBytes, c.Version)
if !reflect.DeepEqual(c.Message, response) {
t.Errorf("case %s decode failed, expected:%+v got %+v", c.CaseName, c.Message, response)
}
testEncodable(t, c.CaseName, c.Message, c.MessageBytes)
t.Run(c.Name, func(t *testing.T) {
response := new(DescribeGroupsResponse)
testVersionDecodable(t, c.Name, response, c.MessageBytes, c.Version)
if !assert.Equal(t, c.Message, response) {
t.Errorf("case %s decode failed, expected:%+v got %+v", c.Name, c.Message, response)
}
testEncodable(t, c.Name, c.Message, c.MessageBytes)
})
}
}

0 comments on commit 41bea2e

Please sign in to comment.