/* Registers a new consumer with Consumerid id and TopicCount subscription that is a part of consumer group Groupid in this ConsumerCoordinator. Returns an error if registration failed, nil otherwise. */
340056
func (this *ZookeeperCoordinator) RegisterConsumer(Consumerid string, Groupid string, TopicCount TopicsToNumStreams) (err error) {
	backoffMultiplier := 1
	this.ensureZkPathsExist(Groupid)
	for i := 0; i <= this.config.MaxRequestRetries; i++ {
		err = this.tryRegisterConsumer(Consumerid, Groupid, TopicCount)
		if err == nil {
			return
		}
		Tracef(this, "Registering consumer %s in group %s failed after %d-th retry", Consumerid, Groupid, i)
		time.Sleep(this.config.RequestBackoff * time.Duration(backoffMultiplier))
		backoffMultiplier++
	}
	return
}
769035
func (e *Login) SubConsumers() []libkb.UIConsumer {
	return []libkb.UIConsumer{
		&LoginProvisionedDevice{},
		&loginLoadUser{},
		&loginProvision{},
		&AccountReset{},
	}
}
324223
func (this *ZookeeperCoordinator) GetConsumersInGroup(Groupid string) (consumers []string, err error) {
	backoffMultiplier := 1
	for i := 0; i <= this.config.MaxRequestRetries; i++ {
		consumers, err = this.tryGetConsumersInGroup(Groupid)
		if err == nil {
			return
		}
		Tracef(this, "GetConsumersInGroup failed for group %s after %d-th retry", Groupid, i)
		time.Sleep(this.config.RequestBackoff * time.Duration(backoffMultiplier))
		backoffMultiplier++
	}
	return
}
924740
func (cg *Consumergroup) PartitionOwner(topic string, partition int32) (*ConsumergroupInstance, error) {
	node := fmt.Sprintf("%s/consumers/%s/owners/%s/%d", cg.kz.conf.Chroot, cg.Name, topic, partition)
	val, _, err := cg.kz.conn.Get(node)

	// If the node does not exists, nobody has claimed it.
	switch err {
	case nil:
		return &ConsumergroupInstance{cg: cg, ID: string(val)}, nil
	case zk.ErrNoNode:
		return nil, nil
	default:
		return nil, err
	}
}
427755
func NewConsumerConf(topic string, partition int32) ConsumerConf {
	return ConsumerConf{
		Topic:          topic,
		Partition:      partition,
		RequestTimeout: time.Millisecond * 50,
		RetryLimit:     -1,
		RetryWait:      time.Millisecond * 50,
		RetryErrLimit:  10,
		RetryErrWait:   time.Millisecond * 500,
		MinFetchSize:   1,
		MaxFetchSize:   2000000,
		StartOffset:    StartOffsetOldest,
		Logger:         nil,
	}
}
87459
func (c *consumerGroup) Consume(ctx context.Context, topics []string, handler ConsumerGroupHandler) error {
	// Ensure group is not closed
	select {
	case <-c.closed:
		return ErrClosedConsumerGroup
	default:
	}

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

	// Quick exit when no topics are provided
	if len(topics) == 0 {
		return fmt.Errorf("no topics provided")
	}

	// Refresh metadata for requested topics
	if err := c.client.RefreshMetadata(topics...); err != nil {
		return err
	}

	// Init session
	sess, err := c.newSession(ctx, topics, handler, c.config.Consumer.Group.Rebalance.Retry.Max)
	if err == ErrClosedClient {
		return ErrClosedConsumerGroup
	} else if err != nil {
		return err
	}

	// Wait for session exit signal
	<-sess.ctx.Done()

	// Gracefully release session claims
	return sess.release(true)
}
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)
	}
}
830373
func RegisterAlgorithmProvider(name string, predicateKeys, priorityKeys sets.String) string {
	schedulerFactoryMutex.Lock()
	defer schedulerFactoryMutex.Unlock()
	validateAlgorithmNameOrDie(name)
	algorithmProviderMap[name] = AlgorithmProviderConfig{
		FitPredicateKeys:     predicateKeys,
		PriorityFunctionKeys: priorityKeys,
	}
	return name
}
772952
func newConsumer(brokers []string, topic string, group string, offset int64, sync bool) (*ConsumerGroup, error) {

	config := cluster.NewConfig()
	config.Group.Return.Notifications = true
	config.Consumer.Return.Errors = true
	config.Consumer.MaxWaitTime = 50 * time.Millisecond
	config.Consumer.Offsets.Initial = offset
	config.Consumer.Offsets.Retention = 0 // indefinite retention

	logData := log.Data{"topic": topic, "group": group, "config": config}

	consumer, err := cluster.NewConsumer(brokers, group, []string{topic}, config)
	if err != nil {
		log.ErrorC("newConsumer failed", err, logData)
		return nil, err
	}

	var upstream chan Message
	if sync {
		// make the upstream channel buffered, so we can send-and-wait for upstreamDone
		upstream = make(chan Message, 1)
	} else {
		upstream = make(chan Message)
	}

	cg := &ConsumerGroup{
		consumer:     consumer,
		incoming:     upstream,
		closer:       make(chan struct{}),
		closed:       make(chan struct{}),
		errors:       make(chan error),
		topic:        topic,
		group:        group,
		sync:         sync,
		upstreamDone: make(chan bool, 1),
	}

	// listener goroutine - listen to consumer.Messages() and upstream them
	// if this blocks while upstreaming a message, we can shutdown consumer via the following goroutine
	go func() {
		logData := log.Data{"topic": topic, "group": group, "config": config}

		log.Info("Started kafka consumer listener", logData)
		defer close(cg.closed)
		for looping := true; looping; {
			select {
			case <-cg.closer:
				looping = false
			case msg := <-cg.consumer.Messages():
				cg.Incoming() <- SaramaMessage{msg, cg.consumer}
				if cg.sync {
					// wait for msg-processed or close-consumer triggers
					for loopingForSync := true; looping && loopingForSync; {
						select {
						case <-cg.upstreamDone:
							loopingForSync = false
						case <-cg.closer:
							// XXX if we read closer here, this means that the release/upstreamDone blocks unless it is buffered
							looping = false
						}
					}
				}
			}
		}
		cg.consumer.CommitOffsets()
		log.Info("Closed kafka consumer listener", logData)
	}()

	// control goroutine - allows us to close consumer even if blocked while upstreaming a message (above)
	go func() {
		logData := log.Data{"topic": topic, "group": group}

		hasBalanced := false // avoid CommitOffsets() being called before we have balanced (otherwise causes a panic)
		for looping := true; looping; {
			select {
			case <-cg.closer:
				log.Info("Closing kafka consumer controller", logData)
				<-cg.closed
				looping = false
			case err := <-cg.consumer.Errors():
				log.Error(err, nil)
				cg.Errors() <- err
			case <-time.After(tick):
				if hasBalanced {
					cg.consumer.CommitOffsets()
				}
			case n, more := <-cg.consumer.Notifications():
				if more {
					hasBalanced = true
					log.Trace("Rebalancing group", log.Data{"topic": cg.topic, "group": cg.group, "partitions": n.Current[cg.topic]})
				}
			}
		}
		log.Info("Closed kafka consumer controller", logData)
	}()

	return cg, nil
}
98745
func (cg *Consumergroup) Create() error {
	return cg.kz.mkdirRecursive(fmt.Sprintf("%s/consumers/%s", cg.kz.conf.Chroot, cg.Name))
}
255359
func (sc *conn) subscribe(subject, qgroup string, cb MsgHandler, options ...SubscriptionOption) (Subscription, error) {
	sub := &subscription{subject: subject, qgroup: qgroup, inbox: nats.NewInbox(), cb: cb, sc: sc, opts: DefaultSubscriptionOptions}
	for _, opt := range options {
		if err := opt(&sub.opts); err != nil {
			return nil, err
		}
	}
	sc.Lock()
	if sc.nc == nil {
		sc.Unlock()
		return nil, ErrConnectionClosed
	}

	// Register subscription.
	sc.subMap[sub.inbox] = sub
	nc := sc.nc
	sc.Unlock()

	// Hold lock throughout.
	sub.Lock()
	defer sub.Unlock()

	// Listen for actual messages.
	nsub, err := nc.Subscribe(sub.inbox, sc.processMsg)
	if err != nil {
		return nil, err
	}
	sub.inboxSub = nsub

	// Create a subscription request
	// FIXME(dlc) add others.
	sr := &pb.SubscriptionRequest{
		ClientID:      sc.clientID,
		Subject:       subject,
		QGroup:        qgroup,
		Inbox:         sub.inbox,
		MaxInFlight:   int32(sub.opts.MaxInflight),
		AckWaitInSecs: int32(sub.opts.AckWait / time.Second),
		StartPosition: sub.opts.StartAt,
		DurableName:   sub.opts.DurableName,
	}

	// Conditionals
	switch sr.StartPosition {
	case pb.StartPosition_TimeDeltaStart:
		sr.StartTimeDelta = time.Now().UnixNano() - sub.opts.StartTime.UnixNano()
	case pb.StartPosition_SequenceStart:
		sr.StartSequence = sub.opts.StartSequence
	}

	b, _ := sr.Marshal()
	reply, err := nc.Request(sc.subRequests, b, sc.opts.ConnectTimeout)
	if err != nil {
		sub.inboxSub.Unsubscribe()
		if err == nats.ErrTimeout {
			err = ErrSubReqTimeout
		}
		return nil, err
	}
	r := &pb.SubscriptionResponse{}
	if err := r.Unmarshal(reply.Data); err != nil {
		sub.inboxSub.Unsubscribe()
		return nil, err
	}
	if r.Error != "" {
		sub.inboxSub.Unsubscribe()
		return nil, errors.New(r.Error)
	}
	sub.ackInbox = r.AckInbox

	return sub, nil
}
965353
func (l *MockLedger) Register(consumer Consumer) {
	l.Lock()
	defer l.Unlock()
	l.consumers = append(l.consumers, consumer)
}
645725
func (cgi *ConsumergroupInstance) RegisterWithSubscription(subscriptionJSON []byte) error {
	node := fmt.Sprintf("%s/consumers/%s/ids/%s", cgi.cg.kz.conf.Chroot, cgi.cg.Name, cgi.ID)
	err := cgi.cg.kz.create(node, subscriptionJSON, true)
	if err == zk.ErrNodeExists {
		return ErrInstanceAlreadyRegistered
	}
	return err
}
1185568
func (g *groupRegistry) mustRegisterCounter(desc *desc) ID {
	desc.mt = counterMetricType
	g.mustRegister(desc)

	desc.id = newID(len(g.group.counters), g.desc.id)
	g.group.counters = append(g.group.counters, Counter{desc: desc})

	return desc.id
}
524999
func (this *ZookeeperCoordinator) GetConsumerInfo(Consumerid string, Groupid string) (info *ConsumerInfo, err error) {
	backoffMultiplier := 1
	for i := 0; i <= this.config.MaxRequestRetries; i++ {
		info, err = this.tryGetConsumerInfo(Consumerid, Groupid)
		if err == nil {
			return
		}
		Tracef(this, "GetConsumerInfo failed for consumer %s in group %s after %d-th retry", Consumerid, Groupid, i)
		time.Sleep(this.config.RequestBackoff * time.Duration(backoffMultiplier))
		backoffMultiplier++
	}
	return
}
489678
func (cgi *ConsumergroupInstance) Registered() (bool, error) {
	node := fmt.Sprintf("%s/consumers/%s/ids/%s", cgi.cg.kz.conf.Chroot, cgi.cg.Name, cgi.ID)
	return cgi.cg.kz.exists(node)
}
789019
func (s *NatsSubscriber) Subscribe(subject, group string, handler nats.Handler) error {
	ch, err := s.server.initSubject(subject)
	if err != nil {
		return err
	}

	go func(s *NatsSubscriber, subject string, handler nats.Handler) {
		for r := range ch {
			handler.Process(&natsSubscription{subject: subject},
				&natsMessage{
					r: r,
				},
			)
		}
	}(s, subject, handler)
	return nil
}