// joinGroup attempts to join a consumer group
//
// See http://kafka.apache.org/protocol.html#The_Messages_JoinGroup
185964
func (c *Conn) joinGroup(request joinGroupRequestV2) (joinGroupResponseV2, error) {
	var response joinGroupResponseV2

	err := c.writeOperation(
		func(deadline time.Time, id int32) error {
			return c.writeRequest(joinGroupRequest, v2, 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 joinGroupResponseV2{}, err
	}
	if response.ErrorCode != 0 {
		return joinGroupResponseV2{}, Error(response.ErrorCode)
	}

	return response, nil
}
773402
func (api *Client) OpenGroup(group string) (bool, bool, error) {
	return api.OpenGroupContext(context.Background(), group)
}
478295
func (w *Watcher) Join(client Client, recursive bool) *probe.Error {
	wo, err := client.Watch(watchParams{
		recursive: recursive,
		events:    []string{"put", "delete"},
	})
	if err != nil {
		return err
	}

	w.o = append(w.o, wo)

	// join monitoring waitgroup
	w.wg.Add(1)

	// wait for events and errors of individual client watchers
	// and sent then to eventsChan and errorsChan
	go func() {
		defer w.wg.Done()

		for {
			select {
			case event, ok := <-wo.Events():
				if !ok {
					return
				}

				w.eventInfoChan <- event
			case err, ok := <-wo.Errors():
				if !ok {
					return
				}

				w.errorChan <- err
			}
		}
	}()

	return nil
}
1123898
func (groups *Groups) AddGroup(newGroup *Group) {
	if newGroup.GroupID != "" {
		for key, group := range groups.Groups.Items {
			if group.GroupID == newGroup.GroupID {
				groups.Groups.Items[key] = newGroup
				return
			}
		}
	}

	newGroup.parent = groups

	groups.Groups.Items = append(groups.Groups.Items, newGroup)
}
1089695
func (c *Client) Group(id, action, target int32) (*GroupNode, error) {
	msg := osc.Message{
		Address: groupNewAddress,
		Arguments: osc.Arguments{
			osc.Int(id),
			osc.Int(action),
			osc.Int(target),
		},
	}
	if err := c.oscConn.Send(msg); err != nil {
		return nil, err
	}
	return newGroup(c, id), nil
}
1032568
func (b *Broker) SyncGroup(request *SyncGroupRequest) (*SyncGroupResponse, error) {
	response := new(SyncGroupResponse)

	err := b.sendAndReceive(request, response)
	if err != nil {
		return nil, err
	}

	return response, nil
}
929527
func (r *RoleBindingBuilder) Groups(groups ...string) *RoleBindingBuilder {
	for _, group := range groups {
		r.RoleBinding.Subjects = append(r.RoleBinding.Subjects, rbacv1.Subject{Kind: rbacv1.GroupKind, APIGroup: GroupName, Name: group})
	}
	return r
}
206544
func (c *MemoryGroupService) GroupMembers(ctx context.Context, groupName string) ([]string, error) {
	memoryGroupsMu.Lock()
	defer memoryGroupsMu.Unlock()

	mg, ok := memoryGroups[groupName]
	if !ok {
		return nil, constants.ErrGroupNotFound
	}

	uids := make([]string, len(mg.Uids))
	copy(uids, mg.Uids)

	return uids, nil
}
1240126
func (res *GroupsListAPIResponse) Groups() ([]*Group, error) {
	var groups []*Group
	err := json.Unmarshal(res.RawGroups, &groups)
	if err != nil {
		return nil, err
	}
	return groups, nil
}
156539
func (r *Relay) Groups(group string) *GroupOperations {
	return &GroupOperations{
		group: group,
		e:     r.exchange,
		relay: r,
	}
}
431575
func (c *Client) parseGroup(msg osc.Message) (*GroupNode, error) {
	n, _, err := c.parseNodeFrom(msg, 0)
	if err != nil {
		return nil, err
	}
	gn, ok := n.(*GroupNode)
	if !ok {
		return nil, errors.Wrap(err, "type assertion from Node to *GroupNode")
	}
	return gn, nil
}
32153
func (j *joinSender) SelectGroup(nodesJoined []string) []string {
	var group []string
	// if fully exhausted or first round, initialize this round's nodes
	if len(j.roundPreferredNodes) == 0 && len(j.roundNonPreferredNodes) == 0 {
		j.Init(nodesJoined)
	}

	numNodesLeft := j.size - len(nodesJoined)

	cont := func() bool {
		if len(group) == numNodesLeft*j.parallelismFactor {
			return false
		}

		nodesAvailable := len(j.roundPreferredNodes) + len(j.roundNonPreferredNodes)
		if nodesAvailable == 0 {
			return false
		}

		return true
	}

	for cont() {
		if len(j.roundPreferredNodes) > 0 {
			group = append(group, util.TakeNode(&j.roundPreferredNodes, -1))
		} else if len(j.roundNonPreferredNodes) > 0 {
			group = append(group, util.TakeNode(&j.roundNonPreferredNodes, -1))
		}
	}

	return group
}
1072166
func (api *Client) InviteUserToGroupContext(ctx context.Context, group, user string) (*Group, bool, error) {
	values := url.Values{
		"token":   {api.token},
		"channel": {group},
		"user":    {user},
	}

	response, err := api.groupRequest(ctx, "groups.invite", values)
	if err != nil {
		return nil, false, err
	}
	return &response.Group, response.AlreadyInGroup, nil
}
98745
func (cg *Consumergroup) Create() error {
	return cg.kz.mkdirRecursive(fmt.Sprintf("%s/consumers/%s", cg.kz.conf.Chroot, cg.Name))
}
1134497
func (g *Group) AddGroup(shortDescription string, longDescription string, data interface{}) (*Group, error) {
	group := newGroup(shortDescription, longDescription, data)

	group.parent = g

	if err := group.scan(); err != nil {
		return nil, err
	}

	g.groups = append(g.groups, group)
	return group, nil
}
334047
func (cg *ConsumerGroup) partitionConsumer(topic string, partition int32, messages chan<- *sarama.ConsumerMessage, errors chan<- error, wg *sync.WaitGroup, stopper <-chan struct{}) {
	defer wg.Done()

	// Since ProcessingTimeout is the amount of time we'll wait for the final batch
	// of messages to be processed before releasing a partition, we need to wait slightly
	// longer than that before timing out here to ensure that another consumer has had
	// enough time to release the partition. Hence, +2 seconds.
	maxRetries := int(cg.config.Offsets.ProcessingTimeout/time.Second) + 2
partitionClaimLoop:
	for tries := 0; tries < maxRetries; tries++ {
		select {
		case <-stopper:
			return
		case <-time.After(1 * time.Second):
			if err := cg.instance.ClaimPartition(topic, partition); err == nil {
				break partitionClaimLoop
			} else if tries+1 < maxRetries {
				if err == kazoo.ErrPartitionClaimedByOther {
					// Another consumer still owns this partition. We should wait longer for it to release it.
				} else {
					// An unexpected error occurred. Log it and continue trying until we hit the timeout.
					cg.Logf("%s/%d :: FAILED to claim partition on attempt %v of %v; retrying in 1 second. Error: %v", topic, partition, tries+1, maxRetries, err)
				}
			} else {
				cg.Logf("%s/%d :: FAILED to claim the partition: %s\n", topic, partition, err)
				cg.errors <- &sarama.ConsumerError{
					Topic:     topic,
					Partition: partition,
					Err:       err,
				}
				return
			}
		}
	}

	defer func() {
		err := cg.instance.ReleasePartition(topic, partition)
		if err != nil {
			cg.Logf("%s/%d :: FAILED to release partition: %s\n", topic, partition, err)
			cg.errors <- &sarama.ConsumerError{
				Topic:     topic,
				Partition: partition,
				Err:       err,
			}
		}
	}()

	nextOffset, err := cg.offsetManager.InitializePartition(topic, partition)
	if err != nil {
		cg.Logf("%s/%d :: FAILED to determine initial offset: %s\n", topic, partition, err)
		return
	}

	if nextOffset >= 0 {
		cg.Logf("%s/%d :: Partition consumer starting at offset %d.\n", topic, partition, nextOffset)
	} else {
		nextOffset = cg.config.Offsets.Initial
		if nextOffset == sarama.OffsetOldest {
			cg.Logf("%s/%d :: Partition consumer starting at the oldest available offset.\n", topic, partition)
		} else if nextOffset == sarama.OffsetNewest {
			cg.Logf("%s/%d :: Partition consumer listening for new messages only.\n", topic, partition)
		}
	}

	consumer, err := cg.consumePartition(topic, partition, nextOffset)

	if err != nil {
		cg.Logf("%s/%d :: FAILED to start partition consumer: %s\n", topic, partition, err)
		return
	}

	defer consumer.Close()

	err = nil
	var lastOffset int64 = -1 // aka unknown
partitionConsumerLoop:
	for {
		select {
		case <-stopper:
			break partitionConsumerLoop

		case err := <-consumer.Errors():
			if err == nil {
				cg.Logf("%s/%d :: Consumer encountered an invalid state: re-establishing consumption of partition.\n", topic, partition)

				// Errors encountered (if any) are logged in the consumerPartition function
				var cErr error
				consumer, cErr = cg.consumePartition(topic, partition, lastOffset)
				if cErr != nil {
					break partitionConsumerLoop
				}
				continue partitionConsumerLoop
			}

			for {
				select {
				case errors <- err:
					continue partitionConsumerLoop

				case <-stopper:
					break partitionConsumerLoop
				}
			}

		case message := <-consumer.Messages():
			if message == nil {
				cg.Logf("%s/%d :: Consumer encountered an invalid state: re-establishing consumption of partition.\n", topic, partition)

				// Errors encountered (if any) are logged in the consumerPartition function
				var cErr error
				consumer, cErr = cg.consumePartition(topic, partition, lastOffset)
				if cErr != nil {
					break partitionConsumerLoop
				}
				continue partitionConsumerLoop

			}

			for {
				select {
				case <-stopper:
					break partitionConsumerLoop

				case messages <- message:
					lastOffset = message.Offset
					continue partitionConsumerLoop
				}
			}
		}
	}

	cg.Logf("%s/%d :: Stopping partition consumer at offset %d\n", topic, partition, lastOffset)
	if err := cg.offsetManager.FinalizePartition(topic, partition, lastOffset, cg.config.Offsets.ProcessingTimeout); err != nil {
		cg.Logf("%s/%d :: %s\n", topic, partition, err)
	}
}
122390
func (c *consumerGroup) Close() (err error) {
	c.closeOnce.Do(func() {
		close(c.closed)

		c.lock.Lock()
		defer c.lock.Unlock()

		// leave group
		if e := c.leave(); e != nil {
			err = e
		}

		// drain errors
		go func() {
			close(c.errors)
		}()
		for e := range c.errors {
			err = e
		}

		if e := c.client.Close(); e != nil {
			err = e
		}
	})
	return
}