Method run

reader.go:291–348

run provides the main consumer group management loop. Each iteration performs the handshake to join the Reader to the consumer group. This function is responsible for closing the consumer group upon exit.

(cg *ConsumerGroup)

//
// This function is responsible for closing the consumer group upon exit.
func (r *Reader) run(cg *ConsumerGroup) {
	defer close(r.done)
	defer cg.Close()

	r.withLogger(func(l Logger) {
		l.Printf("entering loop for consumer group, %v\n", r.config.GroupID)
	})

	for {
		// Limit the number of attempts at waiting for the next
		// consumer generation.
		var err error
		var gen *Generation
		for attempt := 1; attempt <= r.config.MaxAttempts; attempt++ {
			gen, err = cg.Next(r.stctx)
			if err == nil {
				break
			}
			if errors.Is(err, r.stctx.Err()) {
				return
			}
			r.stats.errors.observe(1)
			r.withErrorLogger(func(l Logger) {
				l.Printf("%v", err)
			})
			// Continue with next attempt...
		}
		if err != nil {
			// All attempts have failed.
			select {
			case r.runError <- err:
				// If somebody's receiving on the runError, let
				// them know the error occurred.
			default:
				// Otherwise, don't block to allow healing.
			}
			continue
		}

		r.stats.rebalances.observe(1)

		r.subscribe(gen.Assignments)

		gen.Start(func(ctx context.Context) {
			r.commitLoop(ctx, gen)
		})
		gen.Start(func(ctx context.Context) {
			// wait for the generation to end and then unsubscribe.
			select {
			case <-ctx.Done():
				// continue to next generation
			case <-r.stctx.Done():
				// this will be the last loop because the reader is closed.
			}
			r.unsubscribe()
		})
	}
}