func (cl *Client) controller(ctx context.Context) (*broker, error) {
	get := func() int32 {
		cl.controllerIDMu.Lock()
		defer cl.controllerIDMu.Unlock()
		return cl.controllerID
	}
	var id int32
	if id = get(); id < 0 {
		if err := cl.fetchBrokerMetadata(ctx); err != nil {
			return nil, err
		}
		if id = get(); id < 0 {
			return nil, &errUnknownController{id}
		}
	}
	return cl.brokerOrErr(nil, id, &errUnknownController{id})
}
controller returns the controller broker, forcing a broker load if
necessary.
package kgo
import (
	"context"
	"errors"
	"fmt"
	"hash/crc32"
	"math/rand"
	"net"
	"reflect"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/twmb/franz-go/pkg/kerr"
	"github.com/twmb/franz-go/pkg/kmsg"
)
var crc32c = crc32.MakeTable(crc32.Castagnoli)  
type Client struct {
	cfg cfg
	ctx       context.Context
	ctxCancel func()
	rng *rand.Rand
	brokersMu    sync.RWMutex
	brokers      map[int32]*broker  
	anyBrokerIdx int32
	anySeedIdx   int32
	stopBrokers  bool  
	sinksAndSourcesMu sync.Mutex
	sinksAndSources   map[int32]sinkAndSource
	reqFormatter  *kmsg.RequestFormatter
	connTimeouter connTimeouter
	bufPool bufPool  
	pnrPool pnrPool  
	controllerIDMu sync.Mutex
	controllerID   int32
	fetchingBrokersMu sync.Mutex
	fetchingBrokers   *struct {
		done chan struct{}
		err  error
	}
	producer producer
	consumer consumer
	compressor   *compressor
	decompressor *decompressor
	coordinatorsMu sync.Mutex
	coordinators   map[coordinatorKey]*coordinatorLoad
	updateMetadataCh    chan string
	updateMetadataNowCh chan string  
	metawait            metawait
	metadone            chan struct{}
}
func (cl *Client) idempotent() bool { return !cl.cfg.disableIdempotency }
type sinkAndSource struct {
	sink   *sink
	source *source
}
type hostport struct {
	host string
	port int32
}
const defaultKafkaPort = 9092
func NewClient(opts ...Opt) (*Client, error) {
	cfg := defaultCfg()
	for _, opt := range opts {
		opt.apply(&cfg)
	}
	if cfg.retryTimeout == nil {
		cfg.retryTimeout = func(key int16) time.Duration {
			switch key {
			case ((*kmsg.JoinGroupRequest)(nil)).Key(),
				((*kmsg.SyncGroupRequest)(nil)).Key(),
				((*kmsg.HeartbeatRequest)(nil)).Key():
				return cfg.sessionTimeout
			}
			return 30 * time.Second
		}
	}
	if err := cfg.validate(); err != nil {
		return nil, err
	}
	seeds := make([]hostport, 0, len(cfg.seedBrokers))
	for _, seedBroker := range cfg.seedBrokers {
		hp, err := parseBrokerAddr(seedBroker)
		if err != nil {
			return nil, err
		}
		seeds = append(seeds, hp)
	}
	ctx, cancel := context.WithCancel(context.Background())
	cl := &Client{
		cfg:       cfg,
		ctx:       ctx,
		ctxCancel: cancel,
		rng:       rand.New(rand.NewSource(time.Now().UnixNano())),
		controllerID: unknownControllerID,
		brokers:      make(map[int32]*broker),
		sinksAndSources: make(map[int32]sinkAndSource),
		reqFormatter:  kmsg.NewRequestFormatter(),
		connTimeouter: connTimeouter{def: cfg.requestTimeoutOverhead},
		bufPool: newBufPool(),
		pnrPool: newPnrPool(),
		decompressor: newDecompressor(),
		coordinators: make(map[coordinatorKey]*coordinatorLoad),
		updateMetadataCh:    make(chan string, 1),
		updateMetadataNowCh: make(chan string, 1),
		metadone:            make(chan struct{}),
	}
	compressor, err := newCompressor(cl.cfg.compression...)
	if err != nil {
		return nil, err
	}
	cl.compressor = compressor
	cl.cfg.hooks.each(func(h Hook) {
		if h, ok := h.(HookNewClient); ok {
			h.OnNewClient(cl)
		}
	})
	cl.producer.init(cl)
	cl.consumer.init(cl)
	cl.metawait.init()
	if cfg.id != nil {
		cl.reqFormatter = kmsg.NewRequestFormatter(kmsg.FormatterClientID(*cfg.id))
	}
	for i, seed := range seeds {
		b := cl.newBroker(unknownSeedID(i), seed.host, seed.port, nil)
		cl.brokers[b.meta.NodeID] = b
	}
	go cl.updateMetadataLoop()
	go cl.reapConnectionsLoop()
	return cl, nil
}
func parseBrokerAddr(addr string) (hostport, error) {
	if strings.IndexByte(addr, '[') == 0 {
		parts := strings.Split(addr[1:], "]")
		if len(parts) != 2 {
			return hostport{}, fmt.Errorf("invalid addr: %s", addr)
		}
		if len(parts[1]) == 0 {
			return hostport{parts[0], defaultKafkaPort}, nil
		}
		port, err := strconv.ParseInt(parts[1][1:], 10, 32)
		if err != nil {
			return hostport{}, fmt.Errorf("unable to parse port from addr: %w", err)
		}
		return hostport{parts[0], int32(port)}, nil
	}
	if strings.IndexByte(addr, ':') == -1 {
		return hostport{addr, defaultKafkaPort}, nil
	}
	h, p, err := net.SplitHostPort(addr)
	if err != nil {
		return hostport{addr, defaultKafkaPort}, nil
	}
	port, err := strconv.ParseInt(p, 10, 32)
	if err != nil {
		return hostport{}, fmt.Errorf("unable to parse port from addr: %w", err)
	}
	return hostport{h, int32(port)}, nil
}
type connTimeouter struct {
	def                  time.Duration
	joinMu               sync.Mutex
	lastRebalanceTimeout time.Duration
}
func (c *connTimeouter) timeouts(req kmsg.Request) (r, w time.Duration) {
	def := c.def
	millis := func(m int32) time.Duration { return time.Duration(m) * time.Millisecond }
	switch t := req.(type) {
	default:
		if timeoutRequest, ok := req.(kmsg.TimeoutRequest); ok {
			timeoutMillis := timeoutRequest.Timeout()
			return def + millis(timeoutMillis), def
		}
		return def, def
	case *produceRequest:
		return def + millis(t.timeout), def
	case *fetchRequest:
		return def + millis(t.maxWait), def
	case *kmsg.FetchRequest:
		return def + millis(t.MaxWaitMillis), def
	case *kmsg.JoinGroupRequest:
		c.joinMu.Lock()
		c.lastRebalanceTimeout = millis(t.RebalanceTimeoutMillis)
		c.joinMu.Unlock()
		return def + millis(t.RebalanceTimeoutMillis), def
	case *kmsg.SyncGroupRequest:
		read := def
		c.joinMu.Lock()
		if c.lastRebalanceTimeout != 0 {
			read = c.lastRebalanceTimeout
		}
		c.joinMu.Unlock()
		return read, def
	}
}
func (cl *Client) broker() *broker {
	cl.brokersMu.Lock()  
	defer cl.brokersMu.Unlock()
	b, exists := cl.brokers[cl.anyBrokerIdx]
	if !exists && cl.anyBrokerIdx != 0 {
		cl.anyBrokerIdx = 0
		b, exists = cl.brokers[cl.anyBrokerIdx]
	}
	cl.anyBrokerIdx++
	if !exists {
		b, exists = cl.brokers[unknownSeedID(int(cl.anySeedIdx))]
		if !exists {
			cl.anySeedIdx = 0  
			b = cl.brokers[unknownSeedID(int(cl.anySeedIdx))]
		}
		cl.anySeedIdx++
	}
	return b
}
func (cl *Client) waitTries(ctx context.Context, backoff time.Duration) bool {
	after := time.NewTimer(backoff)
	defer after.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-cl.ctx.Done():
		return false
	case <-after.C:
		return true
	}
}
func (cl *Client) supportsOffsetForLeaderEpoch() bool {
	cl.brokersMu.RLock()
	defer cl.brokersMu.RUnlock()
	for _, b := range cl.brokers {
		if v := b.loadVersions(); v != nil && v.versions[23] >= 2 {
			return true
		}
	}
	return false
}
func (cl *Client) fetchBrokerMetadata(ctx context.Context) error {
	cl.fetchingBrokersMu.Lock()
	wait := cl.fetchingBrokers
	if wait != nil {
		cl.fetchingBrokersMu.Unlock()
		<-wait.done
		return wait.err
	}
	wait = &struct {
		done chan struct{}
		err  error
	}{done: make(chan struct{})}
	cl.fetchingBrokers = wait
	cl.fetchingBrokersMu.Unlock()
	defer func() {
		cl.fetchingBrokersMu.Lock()
		defer cl.fetchingBrokersMu.Unlock()
		cl.fetchingBrokers = nil
		close(wait.done)
	}()
	_, _, wait.err = cl.fetchMetadata(ctx, kmsg.NewPtrMetadataRequest(), true)
	return wait.err
}
func (cl *Client) fetchMetadataForTopics(ctx context.Context, all bool, topics []string) (*broker, *kmsg.MetadataResponse, error) {
	req := kmsg.NewPtrMetadataRequest()
	req.AllowAutoTopicCreation = cl.cfg.allowAutoTopicCreation
	if all {
		req.Topics = nil
	} else if len(topics) == 0 {
		req.Topics = []kmsg.MetadataRequestTopic{}
	} else {
		for _, topic := range topics {
			reqTopic := kmsg.NewMetadataRequestTopic()
			reqTopic.Topic = kmsg.StringPtr(topic)
			req.Topics = append(req.Topics, reqTopic)
		}
	}
	return cl.fetchMetadata(ctx, req, true)
}
func (cl *Client) fetchMetadata(ctx context.Context, req *kmsg.MetadataRequest, limitRetries bool) (*broker, *kmsg.MetadataResponse, error) {
	r := cl.retriable()
	if limitRetries {
		r.limitRetries = 3
	}
	meta, err := req.RequestWith(ctx, r)
	if err == nil {
		if meta.ControllerID >= 0 {
			cl.controllerIDMu.Lock()
			cl.controllerID = meta.ControllerID
			cl.controllerIDMu.Unlock()
		}
		cl.updateBrokers(meta.Brokers)
	}
	return r.last, meta, err
}
func (cl *Client) updateBrokers(brokers []kmsg.MetadataResponseBroker) {
	newBrokers := make(map[int32]*broker, len(brokers))
	cl.brokersMu.Lock()
	defer cl.brokersMu.Unlock()
	if cl.stopBrokers {
		return
	}
	for _, broker := range brokers {
		b, exists := cl.brokers[broker.NodeID]
		if exists {
			delete(cl.brokers, broker.NodeID)
			if !b.meta.equals(broker) {
				b.stopForever()
				b = cl.newBroker(broker.NodeID, broker.Host, broker.Port, broker.Rack)
			}
		} else {
			b = cl.newBroker(broker.NodeID, broker.Host, broker.Port, broker.Rack)
		}
		newBrokers[broker.NodeID] = b
	}
	for goneID, goneBroker := range cl.brokers {
		if goneID < -1 {  
			newBrokers[goneID] = goneBroker
		} else {
			goneBroker.stopForever()
		}
	}
	cl.brokers = newBrokers
}
func (cl *Client) Close() {
	cl.LeaveGroup()
	cl.ctxCancel()
	cl.brokersMu.Lock()
	cl.stopBrokers = true
	for _, broker := range cl.brokers {
		broker.stopForever()
	}
	cl.brokersMu.Unlock()
	<-cl.metadone
	for _, sns := range cl.sinksAndSources {
		sns.sink.maybeDrain()      
		sns.source.maybeConsume()  
	}
	cl.failBufferedRecords(ErrClientClosed)
}
func (cl *Client) Request(ctx context.Context, req kmsg.Request) (kmsg.Response, error) {
	resps, merge := cl.shardedRequest(ctx, req)
	if merge == nil {
		return resps[0].Resp, resps[0].Err
	}
	return merge(resps)
}
func (cl *Client) retriable() *retriable {
	return cl.retriableBrokerFn(func() (*broker, error) { return cl.broker(), nil })
}
func (cl *Client) retriableBrokerFn(fn func() (*broker, error)) *retriable {
	return &retriable{cl: cl, br: fn}
}
func (cl *Client) shouldRetry(tries int, err error) bool {
	return (kerr.IsRetriable(err) || isRetriableBrokerErr(err)) && int64(tries) < cl.cfg.retries
}
func (cl *Client) shouldRetryNext(tries int, err error) bool {
	return isSkippableBrokerErr(err) && int64(tries) < cl.cfg.retries
}
type retriable struct {
	cl   *Client
	br   func() (*broker, error)
	last *broker
	limitRetries int
	parseRetryErr func(kmsg.Response) error
}
func (r *retriable) Request(ctx context.Context, req kmsg.Request) (kmsg.Response, error) {
	tries := 0
	tryStart := time.Now()
	retryTimeout := r.cl.cfg.retryTimeout(req.Key())
	next, nextErr := r.br()
start:
	tries++
	br, err := next, nextErr
	r.last = br
	var resp kmsg.Response
	var retryErr error
	if err == nil {
		resp, err = r.last.waitResp(ctx, req)
		if err == nil && r.parseRetryErr != nil {
			retryErr = r.parseRetryErr(resp)
		}
	}
	if err != nil || retryErr != nil {
		if r.limitRetries == 0 || tries < r.limitRetries {
			backoff := r.cl.cfg.retryBackoff(tries)
			if retryTimeout == 0 || time.Now().Add(backoff).Sub(tryStart) <= retryTimeout {
				if r.cl.shouldRetry(tries, err) || r.cl.shouldRetry(tries, retryErr) {
					if r.cl.waitTries(ctx, backoff) {
						next, nextErr = r.br()
						goto start
					}
				} else if r.cl.shouldRetryNext(tries, err) {
					next, nextErr = r.br()
					if next != br && r.cl.waitTries(ctx, backoff) {
						goto start
					}
				}
			}
		}
	}
	return resp, err
}
type ResponseShard struct {
	Meta BrokerMetadata
	Req kmsg.Request
	Resp kmsg.Response
	Err error
}
func (cl *Client) RequestSharded(ctx context.Context, req kmsg.Request) []ResponseShard {
	resps, _ := cl.shardedRequest(ctx, req)
	sort.Slice(resps, func(i, j int) bool {
		l := &resps[i].Meta
		r := &resps[j].Meta
		if l.NodeID < r.NodeID {
			return true
		}
		if r.NodeID < l.NodeID {
			return false
		}
		if l.Host < r.Host {
			return true
		}
		if r.Host < l.Host {
			return false
		}
		if l.Port < r.Port {
			return true
		}
		if r.Port < l.Port {
			return false
		}
		if l.Rack == nil {
			return true
		}
		if r.Rack == nil {
			return false
		}
		return *l.Rack < *r.Rack
	})
	return resps
}
type shardMerge func([]ResponseShard) (kmsg.Response, error)
func (cl *Client) shardedRequest(ctx context.Context, req kmsg.Request) ([]ResponseShard, shardMerge) {
	ctx, cancel := context.WithCancel(ctx)
	done := make(chan struct{})
	defer close(done)
	go func() {
		defer cancel()
		select {
		case <-done:
		case <-ctx.Done():
		case <-cl.ctx.Done():
		}
	}()
	switch t := req.(type) {
	case *kmsg.ListOffsetsRequest,  
		*kmsg.OffsetFetchRequest,              
		*kmsg.DescribeGroupsRequest,           
		*kmsg.ListGroupsRequest,               
		*kmsg.DeleteRecordsRequest,            
		*kmsg.OffsetForLeaderEpochRequest,     
		*kmsg.DescribeConfigsRequest,          
		*kmsg.AlterConfigsRequest,             
		*kmsg.AlterReplicaLogDirsRequest,      
		*kmsg.DescribeLogDirsRequest,          
		*kmsg.DeleteGroupsRequest,             
		*kmsg.IncrementalAlterConfigsRequest,  
		*kmsg.DescribeProducersRequest,        
		*kmsg.DescribeTransactionsRequest,     
		*kmsg.ListTransactionsRequest:         
		return cl.handleShardedReq(ctx, req)
	case *kmsg.FindCoordinatorRequest:
		last, resp, err := cl.findCoordinator(ctx, t)
		return shards(shard(last, req, resp, err)), nil
	}
	if metaReq, isMetaReq := req.(*kmsg.MetadataRequest); isMetaReq {
		br, resp, err := cl.fetchMetadata(ctx, metaReq, false)
		return shards(shard(br, req, resp, err)), nil
	} else if adminReq, admin := req.(kmsg.AdminRequest); admin {
		return shards(cl.handleAdminReq(ctx, adminReq)), nil
	} else if groupReq, isGroupReq := req.(kmsg.GroupCoordinatorRequest); isGroupReq {
		return shards(cl.handleCoordinatorReq(ctx, groupReq, coordinatorTypeGroup)), nil
	} else if txnReq, isTxnReq := req.(kmsg.TxnCoordinatorRequest); isTxnReq {
		return shards(cl.handleCoordinatorReq(ctx, txnReq, coordinatorTypeTxn)), nil
	} else if apiVersReq, isApiVersReq := req.(*kmsg.ApiVersionsRequest); isApiVersReq {
		if apiVersReq.ClientSoftwareName == "" && apiVersReq.ClientSoftwareVersion == "" {
			dup := *apiVersReq
			dup.ClientSoftwareName = cl.cfg.softwareName
			dup.ClientSoftwareVersion = cl.cfg.softwareVersion
			req = &dup
		}
	}
	r := cl.retriable()
	resp, err := r.Request(ctx, req)
	return shards(shard(r.last, req, resp, err)), nil
}
func shard(br *broker, req kmsg.Request, resp kmsg.Response, err error) ResponseShard {
	if br == nil {  
		return ResponseShard{unknownBrokerMetadata, req, resp, err}
	}
	return ResponseShard{br.meta, req, resp, err}
}
func shards(shard ...ResponseShard) []ResponseShard {
	return shard
}
func (cl *Client) brokerOrErr(ctx context.Context, id int32, err error) (*broker, error) {
	if id < 0 {
		return nil, err
	}
	tryLoad := ctx != nil
	tries := 0
start:
	cl.brokersMu.RLock()
	broker := cl.brokers[id]
	cl.brokersMu.RUnlock()
	if broker == nil {
		if tryLoad {
			if loadErr := cl.fetchBrokerMetadata(ctx); loadErr != nil {
				return nil, loadErr
			}
			tries++
			if tries < 2 {
				goto start
			}
		}
		return nil, err
	}
	return broker, nil
}