func (c *consumer) addFakeReadyForDraining(topic string, partition int32, err error) {
	c.sourcesReadyMu.Lock()
	c.fakeReadyForDraining = append(c.fakeReadyForDraining, Fetch{Topics: []FetchTopic{{
		Topic: topic,
		Partitions: []FetchPartition{{
			Partition: partition,
			Err:       err,
		}},
	}}})
	c.sourcesReadyMu.Unlock()
	c.sourcesReadyCond.Broadcast()
}
addFakeReadyForDraining saves a fake fetch that has important partition
errors--data loss or auth failures.
package kgo
import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/twmb/franz-go/pkg/kerr"
	"github.com/twmb/franz-go/pkg/kmsg"
)
type Offset struct {
	at           int64
	relative     int64
	epoch        int32
	currentEpoch int32  
}
func (o Offset) MarshalJSON() ([]byte, error) {
	if o.relative == 0 {
		return []byte(fmt.Sprintf(`{"At":%d,"Epoch":%d,"CurrentEpoch":%d}`, o.at, o.epoch, o.currentEpoch)), nil
	}
	return []byte(fmt.Sprintf(`{"At":%d,"Relative":%d,"Epoch":%d,"CurrentEpoch":%d}`, o.at, o.relative, o.epoch, o.currentEpoch)), nil
}
func NewOffset() Offset {
	return Offset{
		at:    -1,
		epoch: -1,
	}
}
func (o Offset) AtStart() Offset {
	o.at = -2
	return o
}
func (o Offset) AtEnd() Offset {
	o.at = -1
	return o
}
func (o Offset) Relative(n int64) Offset {
	o.relative = n
	return o
}
func (o Offset) WithEpoch(e int32) Offset {
	if e < 0 {
		e = -1
	}
	o.epoch = e
	return o
}
func (o Offset) At(at int64) Offset {
	if at < -2 {
		at = -2
	}
	o.at = at
	return o
}
type consumer struct {
	cl *Client
	bufferedRecords int64
	pausedMu sync.Mutex    
	paused   atomic.Value  
	mu sync.Mutex
	d  *directConsumer  
	g  *groupConsumer   
	outstandingMetadataUpdates workLoop
	sessionChangeMu sync.Mutex
	session atomic.Value  
	usingCursors usedCursors
	sourcesReadyMu          sync.Mutex
	sourcesReadyCond        *sync.Cond
	sourcesReadyForDraining []*source
	fakeReadyForDraining    []Fetch
}
func (c *consumer) loadPaused() pausedTopics   { return c.paused.Load().(pausedTopics) }
func (c *consumer) clonePaused() pausedTopics  { return c.paused.Load().(pausedTopics).clone() }
func (c *consumer) storePaused(p pausedTopics) { c.paused.Store(p) }
func (cl *Client) BufferedFetchRecords() int64 {
	return atomic.LoadInt64(&cl.consumer.bufferedRecords)
}
type usedCursors map[*cursor]struct{}
func (u *usedCursors) use(c *cursor) {
	if *u == nil {
		*u = make(map[*cursor]struct{})
	}
	(*u)[c] = struct{}{}
}
func (c *consumer) init(cl *Client) {
	c.cl = cl
	c.paused.Store(make(pausedTopics))
	c.sourcesReadyCond = sync.NewCond(&c.sourcesReadyMu)
	if len(cl.cfg.topics) == 0 && len(cl.cfg.partitions) == 0 {
		return  
	}
	defer cl.triggerUpdateMetadata(true, "client initialization")  
	if len(cl.cfg.group) == 0 {
		c.initDirect()
	} else {
		c.initGroup()
	}
}
func (c *consumer) consuming() bool {
	return c.g != nil || c.d != nil
}
func (c *consumer) unset() {
	if !c.consuming() {
		return
	}
	wait := func() {}  
	defer func() { wait() }()
	c.mu.Lock()  
	defer c.mu.Unlock()
	c.assignPartitions(nil, assignInvalidateAll, noTopicsPartitions)
	if c.g != nil {
		wait = c.g.leave()
	}
}
func (c *consumer) addSourceReadyForDraining(source *source) {
	c.sourcesReadyMu.Lock()
	c.sourcesReadyForDraining = append(c.sourcesReadyForDraining, source)
	c.sourcesReadyMu.Unlock()
	c.sourcesReadyCond.Broadcast()
}