Method fetchPollLoop

producer.go:609–626

Loops every FetchPollInterval to check for jobs. This is meant as a back up in case something with listen/notify didn't work, or the fetch limiter was limited so there's still jobs to pick up, and it's also important in poll-only mode.

(ctx context.Context, wg *sync.WaitGroup)

// limited so there's still jobs to pick up, and it's also important in
// poll-only mode.
func (p *producer) fetchPollLoop(ctx context.Context, wg *sync.WaitGroup) {
	defer wg.Done()

	fetchPollTimer := time.NewTimer(p.jitteredFetchPollInterval())
	for {
		select {
		case <-ctx.Done():
			// Stop fetch timer so no more fetches are triggered.
			if !fetchPollTimer.Stop() {
				<-fetchPollTimer.C
			}
			return
		case <-fetchPollTimer.C:
			p.fetchLimiter.Call()
			fetchPollTimer.Reset(p.jitteredFetchPollInterval())
		}
	}
}

// jitteredFetchPollInterval returns FetchPollInterval with random jitter in
// [0, 10% of FetchPollInterval) added (minimum 10ms). This prevents multiple