Method start

reader.go:1181–1222

(offsetsByPartition map[topicPartition]int64)

}

func (r *Reader) start(offsetsByPartition map[topicPartition]int64) {
	if r.closed {
		// don't start child reader if parent Reader is closed
		return
	}

	ctx, cancel := context.WithCancel(context.Background())

	r.cancel() // always cancel the previous reader
	r.cancel = cancel
	r.version++

	r.join.Add(len(offsetsByPartition))
	for key, offset := range offsetsByPartition {
		go func(ctx context.Context, key topicPartition, offset int64, join *sync.WaitGroup) {
			defer join.Done()

			(&reader{
				dialer:           r.config.Dialer,
				logger:           r.config.Logger,
				errorLogger:      r.config.ErrorLogger,
				brokers:          r.config.Brokers,
				topic:            key.topic,
				partition:        int(key.partition),
				minBytes:         r.config.MinBytes,
				maxBytes:         r.config.MaxBytes,
				maxWait:          r.config.MaxWait,
				readBatchTimeout: r.config.ReadBatchTimeout,
				backoffDelayMin:  r.config.ReadBackoffMin,
				backoffDelayMax:  r.config.ReadBackoffMax,
				version:          r.version,
				msgs:             r.msgs,
				stats:            r.stats,
				isolationLevel:   r.config.IsolationLevel,
				maxAttempts:      r.config.MaxAttempts,

				// backwards-compatibility flags
				offsetOutOfRangeError: r.config.OffsetOutOfRangeError,
			}).run(ctx, offset)
		}(ctx, key, offset, &r.join)
	}
}

// A reader reads messages from kafka and produces them on its channels, it's
// used as a way to asynchronously fetch messages while the main program reads