func (c *cursor) setOffset(o cursorOffset) {
	c.cursorOffset = o
}
setOffset sets the cursors offset which will be used the next time a fetch
request is built. This function is called under the source mutex while the
source is stopped, and the caller is responsible for calling maybeConsume
after.
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()
}