func (cl *Client) PauseFetchPartitions(topicPartitions map[string][]int32) map[string][]int32 {
	c := &cl.consumer
	if len(topicPartitions) == 0 {
		return c.loadPaused().pausedPartitions()
	}
	c.pausedMu.Lock()
	defer c.pausedMu.Unlock()
	paused := c.clonePaused()
	paused.addPartitions(topicPartitions)
	c.storePaused(paused)
	return paused.pausedPartitions()
}
PauseFetchPartitions sets the client to no longer fetch the given partitions
and returns all currently paused partitions. Paused partitions persist until
resumed. You can call this function with no partitions to simply receive the
list of currently paused partitions.

In contrast to the canonical Java client, this function does not clear
anything currently buffered. Buffered fetches containing paused partitions
are still returned from polling.

Pausing individual partitions is independent from pausing topics with the
PauseFetchTopics method. If you pause partitions for a topic with
PauseFetchPartitions, and then pause that same topic with PauseFetchTopics,
the individually paused partitions will not be unpaused if you only call
ResumeFetchTopics.
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()
}
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()
}
func (cl *Client) PollFetches(ctx context.Context) Fetches {
	return cl.PollRecords(ctx, 0)
}
func (cl *Client) PollRecords(ctx context.Context, maxPollRecords int) Fetches {
	if maxPollRecords == 0 {
		maxPollRecords = -1
	}
	c := &cl.consumer
	c.g.undirtyUncommitted()
	var fetches Fetches
	fill := func() {
		c.mu.Lock()
		defer c.mu.Unlock()
		c.sourcesReadyMu.Lock()
		if maxPollRecords < 0 {
			for _, ready := range c.sourcesReadyForDraining {
				fetches = append(fetches, ready.takeBuffered())
			}
			c.sourcesReadyForDraining = nil
		} else {
			for len(c.sourcesReadyForDraining) > 0 && maxPollRecords > 0 {
				source := c.sourcesReadyForDraining[0]
				fetch, taken, drained := source.takeNBuffered(maxPollRecords)
				if drained {
					c.sourcesReadyForDraining = c.sourcesReadyForDraining[1:]
				}
				maxPollRecords -= taken
				fetches = append(fetches, fetch)
			}
		}
		realFetches := fetches
		fetches = append(fetches, c.fakeReadyForDraining...)
		c.fakeReadyForDraining = nil
		c.sourcesReadyMu.Unlock()
		if len(realFetches) == 0 {
			return
		}
		if c.g != nil {
			c.g.updateUncommitted(realFetches)
		}
	}
	fill()
	if len(fetches) > 0 || ctx == nil {
		return fetches
	}
	select {
	case <-ctx.Done():
		return fetches
	default:
	}
	done := make(chan struct{})
	quit := false
	go func() {
		c.sourcesReadyMu.Lock()
		defer c.sourcesReadyMu.Unlock()
		defer close(done)
		for !quit && len(c.sourcesReadyForDraining) == 0 {
			c.sourcesReadyCond.Wait()
		}
	}()
	exit := func() {
		c.sourcesReadyMu.Lock()
		quit = true
		c.sourcesReadyMu.Unlock()
		c.sourcesReadyCond.Broadcast()
	}
	select {
	case <-cl.ctx.Done():
		c.addFakeReadyForDraining("", 0, ErrClientClosed)
		exit()
	case <-ctx.Done():
		exit()
	case <-done:
	}
	fill()
	return fetches
}
func (cl *Client) PauseFetchTopics(topics ...string) []string {
	c := &cl.consumer
	if len(topics) == 0 {
		return c.loadPaused().pausedTopics()
	}
	c.pausedMu.Lock()
	defer c.pausedMu.Unlock()
	paused := c.clonePaused()
	paused.addTopics(topics...)
	c.storePaused(paused)
	return paused.pausedTopics()
}