Method leave

consumer_group.go:687–743

Leaves the cluster, called by Close.

()

// Leaves the cluster, called by Close.
func (c *consumerGroup) leave() error {
	c.lock.Lock()
	defer c.lock.Unlock()
	if c.memberID == "" {
		return nil
	}

	coordinator, err := c.client.Coordinator(c.groupID)
	if err != nil {
		return err
	}

	// as per KIP-345 if groupInstanceId is set, i.e. static membership is in action, then do not leave group when consumer closed, just clear memberID
	if c.groupInstanceId != nil {
		c.memberID = ""
		return nil
	}
	req := &LeaveGroupRequest{
		GroupId:  c.groupID,
		MemberId: c.memberID,
	}
	if c.config.Version.IsAtLeast(V3_2_0_0) {
		req.Version = 5
	} else if c.config.Version.IsAtLeast(V2_4_0_0) {
		req.Version = 4
	} else if c.config.Version.IsAtLeast(V2_0_0_0) {
		req.Version = 2
	} else if c.config.Version.IsAtLeast(V0_11_0_0) {
		req.Version = 1
	}
	if req.Version >= 3 {
		member := MemberIdentity{
			MemberId: c.memberID,
		}
		if req.Version >= 5 {
			reason := "the consumer is being closed"
			member.Reason = &reason
		}
		req.Members = append(req.Members, member)
	}

	resp, err := coordinator.LeaveGroup(req)
	if err != nil {
		_ = coordinator.Close()
		return err
	}

	// clear the memberID
	c.memberID = ""

	switch resp.Err {
	case ErrRebalanceInProgress, ErrUnknownMemberId, ErrNoError:
		return nil
	default:
		return resp.Err
	}
}