Method joinGroup

conn.go:371–397

joinGroup attempts to join a consumer group See http://kafka.apache.org/protocol.html#The_Messages_JoinGroup

(request joinGroupRequest)

//
// See http://kafka.apache.org/protocol.html#The_Messages_JoinGroup
func (c *Conn) joinGroup(request joinGroupRequest) (joinGroupResponse, error) {
	version, err := c.negotiateVersion(joinGroup, v1, v2)
	if err != nil {
		return joinGroupResponse{}, err
	}

	response := joinGroupResponse{v: version}

	err = c.writeOperation(
		func(deadline time.Time, id int32) error {
			return c.writeRequest(joinGroup, version, id, request)
		},
		func(deadline time.Time, size int) error {
			return expectZeroSize(func() (remain int, err error) {
				return (&response).readFrom(&c.rbuf, size)
			}())
		},
	)
	if err != nil {
		return joinGroupResponse{}, err
	}
	if response.ErrorCode != 0 {
		return joinGroupResponse{}, Error(response.ErrorCode)
	}

	return response, nil
}

// leaveGroup leaves the consumer from the consumer group
//