Method dispatcher

consumer.go:536–586

()

}

func (child *partitionConsumer) dispatcher() {
	defer func() {
		if child.broker != nil {
			child.consumer.unrefBrokerConsumer(child.broker)
		}
		child.consumer.removeChild(child)
		close(child.feeder)
	}()

	var backoff <-chan time.Time
	for {
		select {
		case <-child.dispatcherStop:
			return
		case <-child.dying:
			child.waitForBrokerHandover()
			return
		case <-child.trigger:
			if max := child.conf.Consumer.Retry.Max; max > 0 && int(child.retries.Load()) >= max {
				Logger.Printf("consumer/%s/%d giving up after %d consecutive failures\n",
					child.topic, child.partition, child.retries.Load())
				child.sendError(ErrConsumerRetriesExhausted)
				child.AsyncClose()
				child.waitForBrokerHandover()
				return
			}
			// only set the timer when none is pending, so retries increments
			// once per dispatch attempt rather than once per trigger
			if backoff == nil {
				backoff = time.After(child.computeBackoff())
			}
		case <-backoff:
			backoff = nil
			if child.broker != nil {
				child.consumer.unrefBrokerConsumer(child.broker)
				child.broker = nil
			}
			if err := child.dispatch(); err != nil {
				select {
				case <-child.dispatcherStop:
					return
				case <-child.dying:
					return
				default:
					child.sendError(err)
					child.triggerRedispatch()
				}
			}
		}
	}
}

// waitForBrokerHandover blocks until the brokerConsumer releases the current
// subscription, so the deferred close(feeder) cannot race an in-flight