func (s *source) fetch(consumerSession *consumerSession, doneFetch chan<- struct{}) (fetched bool) {
	req := s.createReq()
	var (
		alreadySentToDoneFetch bool
		setOffsets             bool
		buffered               bool
	)
	defer func() {
		if !buffered {
			if req.numOffsets > 0 {
				if setOffsets {
					req.usedOffsets.finishUsingAllWithSet()
				} else {
					req.usedOffsets.finishUsingAll()
				}
			}
			if !alreadySentToDoneFetch {
				doneFetch <- struct{}{}
			}
		}
	}()
	if req.numOffsets == 0 {  
		return
	}
	var (
		kresp       kmsg.Response
		requested   = make(chan struct{})
		ctx, cancel = context.WithCancel(consumerSession.ctx)
	)
	defer cancel()
	br, err := s.cl.brokerOrErr(ctx, s.nodeID, errUnknownBroker)
	if err != nil {
		close(requested)
	} else {
		br.do(ctx, req, func(k kmsg.Response, e error) {
			kresp, err = k, e
			close(requested)
		})
	}
	select {
	case <-requested:
		fetched = true
	case <-ctx.Done():
		return
	}
	if err != nil {
		doneFetch <- struct{}{}
		alreadySentToDoneFetch = true
		s.session.reset()
		s.cl.triggerUpdateMetadata(false, "opportunistic load during source backoff")  
		s.consecutiveFailures++
		after := time.NewTimer(s.cl.cfg.retryBackoff(s.consecutiveFailures))
		defer after.Stop()
		select {
		case <-after.C:
		case <-ctx.Done():
		}
		return
	}
	s.consecutiveFailures = 0
	resp := kresp.(*kmsg.FetchResponse)
	var (
		fetch         Fetch
		reloadOffsets listOrEpochLoads
		preferreds    cursorPreferreds
		updateMeta    bool
		handled       = make(chan struct{})
	)
	go func() {
		defer close(handled)
		fetch, reloadOffsets, preferreds, updateMeta = s.handleReqResp(br, req, resp)
	}()
	select {
	case <-handled:
	case <-ctx.Done():
		return
	}
	deleteReqUsedOffset := func(topic string, partition int32) {
		t := req.usedOffsets[topic]
		delete(t, partition)
		if len(t) == 0 {
			delete(req.usedOffsets, topic)
		}
	}
	preferreds.eachPreferred(func(c cursorOffsetPreferred) {
		c.move()
		deleteReqUsedOffset(c.from.topic, c.from.partition)
	})
	reloadOffsets.each(deleteReqUsedOffset)
	s.session = req.session
	switch err := kerr.ErrorForCode(resp.ErrorCode); err {
	case kerr.FetchSessionIDNotFound:
		if s.session.epoch == 0 {
			s.cl.cfg.logger.Log(LogLevelInfo, "session failed with SessionIDNotFound while trying to establish a session; broker likely maxed on sessions; continuing on without using sessions", "broker", logID(s.nodeID))
			s.session.kill()
		} else {
			s.cl.cfg.logger.Log(LogLevelInfo, "received SessionIDNotFound from our in use session, our session was likely evicted; resetting session", "broker", logID(s.nodeID))
			s.session.reset()
		}
		return
	case kerr.InvalidFetchSessionEpoch:
		s.cl.cfg.logger.Log(LogLevelInfo, "resetting fetch session", "broker", logID(s.nodeID), "err", err)
		s.session.reset()
		return
	case kerr.FetchSessionTopicIDError, kerr.UnknownTopicID, kerr.InconsistentTopicID:
		s.cl.cfg.logger.Log(LogLevelInfo, "topic id issues, resetting session and updating metadata", "broker", logID(s.nodeID), "err", err)
		s.session.reset()
		s.cl.triggerUpdateMetadataNow("topic id issues")
		return
	}
	setOffsets = true
	if resp.Version < 7 {
		s.session.kill()
	} else if resp.SessionID > 0 {
		s.session.bumpEpoch(resp.SessionID)
	}
	if len(preferreds) > 0 {
		s.session.reset()
	}
	if updateMeta && !reloadOffsets.loadWithSessionNow(consumerSession, "out of range offset / fenced leader epoch caused reload offsets on fetch") {
		s.cl.triggerUpdateMetadataNow("fetch partitition had an error causing immediate metadata update")
	}
	if fetch.hasErrorsOrRecords() {
		buffered = true
		s.buffered = bufferedFetch{
			fetch:       fetch,
			doneFetch:   doneFetch,
			usedOffsets: req.usedOffsets,
		}
		s.sem = make(chan struct{})
		s.hook(&fetch, true, false)  
		s.cl.consumer.addSourceReadyForDraining(s)
	}
	return
}
fetch is the main logic center of fetching messages.

This is a long function, made much longer by winded documentation, that
contains a lot of the side effects of fetching and updating. The function
consists of two main bulks of logic:

* First, issue a request that can be killed if the source needs to be
stopped. Processing the response modifies no state on the source.

* Second, we keep the fetch response and update everything relevant
(session, trigger some list or epoch updates, buffer the fetch).

One small part between the first and second step is to update preferred
replicas. We always keep the preferred replicas from the fetch response
*even if* the source needs to be stopped. The knowledge of which preferred
replica to use would not be out of date even if the consumer session is
changing.
package kgo
import (
	"context"
	"encoding/binary"
	"fmt"
	"hash/crc32"
	"sync"
	"sync/atomic"
	"time"
	"github.com/twmb/franz-go/pkg/kbin"
	"github.com/twmb/franz-go/pkg/kerr"
	"github.com/twmb/franz-go/pkg/kmsg"
)
type readerFrom interface {
	ReadFrom([]byte) error
}
type source struct {
	cl     *Client  
	nodeID int32    
	consecutiveFailures int
	fetchState workLoop
	sem        chan struct{}  
	buffered   bufferedFetch  
	session fetchSession  
	cursorsMu    sync.Mutex
	cursors      []*cursor  
	cursorsStart int        
}
func (cl *Client) newSource(nodeID int32) *source {
	s := &source{
		cl:     cl,
		nodeID: nodeID,
		sem:    make(chan struct{}),
	}
	close(s.sem)
	return s
}
func (s *source) addCursor(add *cursor) {
	s.cursorsMu.Lock()
	add.cursorsIdx = len(s.cursors)
	s.cursors = append(s.cursors, add)
	s.cursorsMu.Unlock()
	s.maybeConsume()
}
func (s *source) removeCursor(rm *cursor) {
	s.cursorsMu.Lock()
	defer s.cursorsMu.Unlock()
	if rm.cursorsIdx != len(s.cursors)-1 {
		s.cursors[rm.cursorsIdx], s.cursors[len(s.cursors)-1] = s.cursors[len(s.cursors)-1], nil
		s.cursors[rm.cursorsIdx].cursorsIdx = rm.cursorsIdx
	} else {
		s.cursors[rm.cursorsIdx] = nil  
	}
	s.cursors = s.cursors[:len(s.cursors)-1]
	if s.cursorsStart == len(s.cursors) {
		s.cursorsStart = 0
	}
}
type cursor struct {
	topic     string
	topicID   [16]byte
	partition int32
	keepControl bool  
	cursorsIdx int  
	source *source
	useState uint32
	topicPartitionData  
	cursorOffset
}
type cursorOffset struct {
	offset int64
	lastConsumedEpoch int32
}
func (c *cursor) use() *cursorOffsetNext {
	c.useState = 0
	return &cursorOffsetNext{
		cursorOffset:       c.cursorOffset,
		from:               c,
		currentLeaderEpoch: c.leaderEpoch,
	}
}
func (c *cursor) unset() {
	c.useState = 0
	c.setOffset(cursorOffset{
		offset:            -1,
		lastConsumedEpoch: -1,
	})
}
func (c *cursor) usable() bool {
	return atomic.LoadUint32(&c.useState) == 1
}
func (c *cursor) allowUsable() {
	atomic.SwapUint32(&c.useState, 1)
	c.source.maybeConsume()
}
func (c *cursor) setOffset(o cursorOffset) {
	c.cursorOffset = o
}
type cursorOffsetNext struct {
	cursorOffset
	from *cursor
	currentLeaderEpoch int32
}
type cursorOffsetPreferred struct {
	cursorOffsetNext
	preferredReplica int32
}
func (p *cursorOffsetPreferred) move() {
	c := p.from
	defer c.allowUsable()
	c.source.cl.sinksAndSourcesMu.Lock()
	sns, exists := c.source.cl.sinksAndSources[p.preferredReplica]
	c.source.cl.sinksAndSourcesMu.Unlock()
	if !exists {
		c.source.cl.triggerUpdateMetadataNow("cursor moving to a different broker that is not yet known")
		return
	}
	c.source.removeCursor(c)
	c.source = sns.source
	c.source.addCursor(c)
}
type cursorPreferreds []cursorOffsetPreferred
func (cs cursorPreferreds) eachPreferred(fn func(cursorOffsetPreferred)) {
	for _, c := range cs {
		fn(c)
	}
}
type usedOffsets map[string]map[int32]*cursorOffsetNext
func (os usedOffsets) eachOffset(fn func(*cursorOffsetNext)) {
	for _, ps := range os {
		for _, o := range ps {
			fn(o)
		}
	}
}
func (os usedOffsets) finishUsingAllWithSet() {
	os.eachOffset(func(o *cursorOffsetNext) { o.from.setOffset(o.cursorOffset); o.from.allowUsable() })
}
func (os usedOffsets) finishUsingAll() {
	os.eachOffset(func(o *cursorOffsetNext) { o.from.allowUsable() })
}
type bufferedFetch struct {
	fetch Fetch
	doneFetch   chan<- struct{}  
	usedOffsets usedOffsets      
}
func (s *source) hook(f *Fetch, buffered, polled bool) {
	s.cl.cfg.hooks.each(func(h Hook) {
		switch h := h.(type) {
		case HookFetchRecordBuffered:
			if !buffered {
				return
			}
			for i := range f.Topics {
				t := &f.Topics[i]
				for j := range t.Partitions {
					p := &t.Partitions[j]
					for _, r := range p.Records {
						h.OnFetchRecordBuffered(r)
					}
				}
			}
		case HookFetchRecordUnbuffered:
			if buffered {
				return
			}
			for i := range f.Topics {
				t := &f.Topics[i]
				for j := range t.Partitions {
					p := &t.Partitions[j]
					for _, r := range p.Records {
						h.OnFetchRecordUnbuffered(r, polled)
					}
				}
			}
		}
	})
	var nrecs int
	for i := range f.Topics {
		t := &f.Topics[i]
		for j := range t.Partitions {
			nrecs += len(t.Partitions[j].Records)
		}
	}
	if buffered {
		atomic.AddInt64(&s.cl.consumer.bufferedRecords, int64(nrecs))
	} else {
		atomic.AddInt64(&s.cl.consumer.bufferedRecords, -int64(nrecs))
	}
}
func (s *source) takeBuffered() Fetch {
	return s.takeBufferedFn(true, usedOffsets.finishUsingAllWithSet)
}
func (s *source) discardBuffered() {
	s.takeBufferedFn(false, usedOffsets.finishUsingAll)
}
func (s *source) takeNBuffered(n int) (Fetch, int, bool) {
	var r Fetch
	var taken int
	b := &s.buffered
	bf := &b.fetch
	for len(bf.Topics) > 0 && n > 0 {
		t := &bf.Topics[0]
		r.Topics = append(r.Topics, *t)
		rt := &r.Topics[len(r.Topics)-1]
		rt.Partitions = nil
		tCursors := b.usedOffsets[t.Topic]
		for len(t.Partitions) > 0 && n > 0 {
			p := &t.Partitions[0]
			rt.Partitions = append(rt.Partitions, *p)
			rp := &rt.Partitions[len(rt.Partitions)-1]
			take := n
			if take > len(p.Records) {
				take = len(p.Records)
			}
			rp.Records = p.Records[:take:take]
			p.Records = p.Records[take:]
			n -= take
			taken += take
			pCursor := tCursors[p.Partition]
			if len(p.Records) == 0 {
				t.Partitions = t.Partitions[1:]
				pCursor.from.setOffset(pCursor.cursorOffset)
				pCursor.from.allowUsable()
				delete(tCursors, p.Partition)
				if len(tCursors) == 0 {
					delete(b.usedOffsets, t.Topic)
				}
				break
			}
			lastReturnedRecord := rp.Records[len(rp.Records)-1]
			pCursor.from.setOffset(cursorOffset{
				offset:            lastReturnedRecord.Offset + 1,
				lastConsumedEpoch: lastReturnedRecord.LeaderEpoch,
			})
		}
		if len(t.Partitions) == 0 {
			bf.Topics = bf.Topics[1:]
		}
	}
	s.hook(&r, false, true)  
	drained := len(bf.Topics) == 0
	if drained {
		s.takeBuffered()
	}
	return r, taken, drained
}
func (s *source) takeBufferedFn(polled bool, offsetFn func(usedOffsets)) Fetch {
	r := s.buffered
	s.buffered = bufferedFetch{}
	offsetFn(r.usedOffsets)
	r.doneFetch <- struct{}{}
	close(s.sem)
	s.hook(&r.fetch, false, polled)  
	return r.fetch
}
func (s *source) createReq() *fetchRequest {
	req := &fetchRequest{
		maxWait:        s.cl.cfg.maxWait,
		minBytes:       s.cl.cfg.minBytes,
		maxBytes:       s.cl.cfg.maxBytes,
		maxPartBytes:   s.cl.cfg.maxPartBytes,
		rack:           s.cl.cfg.rack,
		isolationLevel: s.cl.cfg.isolationLevel,
		session: s.session,
	}
	paused := s.cl.consumer.loadPaused()
	s.cursorsMu.Lock()
	defer s.cursorsMu.Unlock()
	cursorIdx := s.cursorsStart
	for i := 0; i < len(s.cursors); i++ {
		c := s.cursors[cursorIdx]
		cursorIdx = (cursorIdx + 1) % len(s.cursors)
		if !c.usable() || paused.has(c.topic, c.partition) {
			continue
		}
		req.addCursor(c)
	}
	if len(s.cursors) > 0 {
		s.cursorsStart = (s.cursorsStart + 1) % len(s.cursors)
	}
	return req
}
func (s *source) maybeConsume() {
	if s.fetchState.maybeBegin() {
		go s.loopFetch()
	}
}
func (s *source) loopFetch() {
	consumer := &s.cl.consumer
	session := consumer.loadSession()
	if session == noConsumerSession {
		s.fetchState.hardFinish()
		return
	}
	session.incWorker()
	defer session.decWorker()
	select {
	case <-session.ctx.Done():
		s.fetchState.hardFinish()
		return
	default:
	}
	canFetch := make(chan chan struct{}, 1)
	again := true
	for again {
		select {
		case <-session.ctx.Done():
			s.fetchState.hardFinish()
			return
		case <-s.sem:
		}
		select {
		case <-session.ctx.Done():
			s.fetchState.hardFinish()
			return
		case session.desireFetch() <- canFetch:
		}
		select {
		case <-session.ctx.Done():
			session.cancelFetchCh <- canFetch
			s.fetchState.hardFinish()
			return
		case doneFetch := <-canFetch:
			again = s.fetchState.maybeFinish(s.fetch(session, doneFetch))
		}
	}
}