func (cl *Client) addUnknownTopicRecord(pr promisedRec) {
	unknown := cl.producer.unknownTopics[pr.Topic]
	if unknown == nil {
		unknown = &unknownTopicProduces{
			buffered: make([]promisedRec, 0, 100),
			wait:     make(chan error, 5),
		}
		cl.producer.unknownTopics[pr.Topic] = unknown
	}
	unknown.buffered = append(unknown.buffered, pr)
	if len(unknown.buffered) == 1 {
		go cl.waitUnknownTopic(pr.Topic, unknown)
	}
}
addUnknownTopicRecord adds a record to a topic whose partitions are
currently unknown. This is always called with the unknownTopicsMu held.
package kgo
import (
	"context"
	"errors"
	"fmt"
	"math"
	"sync"
	"sync/atomic"
	"time"
	"github.com/twmb/franz-go/pkg/kerr"
	"github.com/twmb/franz-go/pkg/kmsg"
)
type producer struct {
	topicsMu sync.Mutex  
	topics   *topicsPartitions
	hooks *struct {
		buffered   []HookProduceRecordBuffered
		unbuffered []HookProduceRecordUnbuffered
	}
	unknownTopicsMu sync.Mutex
	unknownTopics   map[string]*unknownTopicProduces
	bufferedRecords int64
	id           atomic.Value
	producingTxn uint32  
	flushing int32  
	aborting int32  
	idMu       sync.Mutex
	idVersion  int16
	waitBuffer chan struct{}
	notifyMu   sync.Mutex
	notifyCond *sync.Cond
	txnMu sync.Mutex
	inTxn bool
}
func (cl *Client) BufferedProduceRecords() int64 {
	return atomic.LoadInt64(&cl.producer.bufferedRecords)
}
type unknownTopicProduces struct {
	buffered []promisedRec
	wait     chan error
}
func (p *producer) init(cl *Client) {
	p.topics = newTopicsPartitions()
	p.unknownTopics = make(map[string]*unknownTopicProduces)
	p.waitBuffer = make(chan struct{}, 32)
	p.idVersion = -1
	p.id.Store(&producerID{
		id:    -1,
		epoch: -1,
		err:   errReloadProducerID,
	})
	p.notifyCond = sync.NewCond(&p.notifyMu)
	inithooks := func() {
		if p.hooks == nil {
			p.hooks = &struct {
				buffered   []HookProduceRecordBuffered
				unbuffered []HookProduceRecordUnbuffered
			}{}
		}
	}
	cl.cfg.hooks.each(func(h Hook) {
		if h, ok := h.(HookProduceRecordBuffered); ok {
			inithooks()
			p.hooks.buffered = append(p.hooks.buffered, h)
		}
		if h, ok := h.(HookProduceRecordUnbuffered); ok {
			inithooks()
			p.hooks.unbuffered = append(p.hooks.unbuffered, h)
		}
	})
}
func (p *producer) isAborting() bool { return atomic.LoadInt32(&p.aborting) > 0 }
func noPromise(*Record, error) {}
type ProduceResult struct {
	Record *Record
	Err error
}
type ProduceResults []ProduceResult
func (rs ProduceResults) FirstErr() error {
	for _, r := range rs {
		if r.Err != nil {
			return r.Err
		}
	}
	return nil
}
func (rs ProduceResults) First() (*Record, error) {
	return rs[0].Record, rs[0].Err
}
func (cl *Client) ProduceSync(ctx context.Context, rs ...*Record) ProduceResults {
	var (
		wg      sync.WaitGroup
		mu      sync.Mutex
		results = make(ProduceResults, 0, len(rs))
		promise = func(r *Record, err error) {
			mu.Lock()
			results = append(results, ProduceResult{r, err})
			mu.Unlock()
			wg.Done()
		}
	)
	wg.Add(len(rs))
	for _, r := range rs {
		cl.Produce(ctx, r, promise)
	}
	wg.Wait()
	return results
}
type FirstErrPromise struct {
	wg   sync.WaitGroup
	once uint32
	err  error
	cl   *Client
}
func AbortingFirstErrPromise(cl *Client) *FirstErrPromise {
	return &FirstErrPromise{
		cl: cl,
	}
}
func (f *FirstErrPromise) promise(_ *Record, err error) {
	defer f.wg.Done()
	if err != nil && atomic.SwapUint32(&f.once, 1) == 0 {
		f.err = err
		if f.cl != nil {
			f.wg.Add(1)
			go func() {
				defer f.wg.Done()
				f.cl.AbortBufferedRecords(context.Background())
			}()
		}
	}
}
func (f *FirstErrPromise) Promise() func(*Record, error) {
	f.wg.Add(1)
	return f.promise
}
func (f *FirstErrPromise) Err() error {
	f.wg.Wait()
	return f.err
}
func (cl *Client) Produce(
	ctx context.Context,
	r *Record,
	promise func(*Record, error),
) {
	if promise == nil {
		promise = noPromise
	}
	if r.Topic == "" {
		if def := cl.cfg.defaultProduceTopic; def != "" {
			r.Topic = def
		} else {
			go promise(r, errors.New("cannot produce to a record that does not have a topic set"))
			return
		}
	}
	p := &cl.producer
	if cl.cfg.txnID != nil && atomic.LoadUint32(&p.producingTxn) != 1 {
		go promise(r, errNotInTransaction)  
		return
	}
	if p.hooks != nil {
		for _, h := range p.hooks.buffered {
			h.OnProduceRecordBuffered(r)
		}
	}
	if atomic.AddInt64(&p.bufferedRecords, 1) > cl.cfg.maxBufferedRecords {
		drainBuffered := func(err error) {
			go func() { <-p.waitBuffer }()
			go cl.finishRecordPromise(promisedRec{ctx, promise, r}, err)
		}
		if cl.cfg.manualFlushing {
			drainBuffered(ErrMaxBuffered)
			return
		}
		select {
		case <-p.waitBuffer:
		case <-cl.ctx.Done():
			drainBuffered(ErrClientClosed)
			return
		case <-ctx.Done():
			drainBuffered(ctx.Err())
			return
		}
	}
	cl.partitionRecord(promisedRec{ctx, promise, r})
}
func (cl *Client) finishRecordPromise(pr promisedRec, err error) {
	p := &cl.producer
	if p.hooks != nil {
		for _, h := range p.hooks.unbuffered {
			h.OnProduceRecordUnbuffered(pr.Record, err)
		}
	}
	pr.promise(pr.Record, err)
	buffered := atomic.AddInt64(&p.bufferedRecords, -1)
	if buffered >= cl.cfg.maxBufferedRecords {
		go func() { p.waitBuffer <- struct{}{} }()
	} else if buffered == 0 && atomic.LoadInt32(&p.flushing) > 0 {
		p.notifyMu.Lock()
		p.notifyMu.Unlock()
		p.notifyCond.Broadcast()
	}
}
func (cl *Client) partitionRecord(pr promisedRec) {
	parts, partsData := cl.partitionsForTopicProduce(pr)
	if parts == nil {  
		return
	}
	cl.doPartitionRecord(parts, partsData, pr)
}
func (cl *Client) doPartitionRecord(parts *topicPartitions, partsData *topicPartitionsData, pr promisedRec) {
	if partsData.loadErr != nil && !kerr.IsRetriable(partsData.loadErr) {
		cl.finishRecordPromise(pr, partsData.loadErr)
		return
	}
	parts.partsMu.Lock()
	defer parts.partsMu.Unlock()
	if parts.partitioner == nil {
		parts.partitioner = cl.cfg.partitioner.ForTopic(pr.Topic)
	}
	mapping := partsData.writablePartitions
	if parts.partitioner.RequiresConsistency(pr.Record) {
		mapping = partsData.partitions
	}
	if len(mapping) == 0 {
		cl.finishRecordPromise(pr, errors.New("unable to partition record due to no usable partitions"))
		return
	}
	var pick int
	tlp, _ := parts.partitioner.(TopicBackupPartitioner)
	if tlp != nil {
		if parts.lb == nil {
			parts.lb = new(leastBackupInput)
		}
		parts.lb.mapping = mapping
		pick = tlp.PartitionByBackup(pr.Record, len(mapping), parts.lb)
	} else {
		pick = parts.partitioner.Partition(pr.Record, len(mapping))
	}
	if pick < 0 || pick >= len(mapping) {
		cl.finishRecordPromise(pr, fmt.Errorf("invalid record partitioning choice of %d from %d available", pick, len(mapping)))
		return
	}
	partition := mapping[pick]
	onNewBatch, _ := parts.partitioner.(TopicPartitionerOnNewBatch)
	abortOnNewBatch := onNewBatch != nil
	processed := partition.records.bufferRecord(pr, abortOnNewBatch)  
	if !processed {
		onNewBatch.OnNewBatch()
		if tlp != nil {
			parts.lb.mapping = mapping
			pick = tlp.PartitionByBackup(pr.Record, len(mapping), parts.lb)
		} else {
			pick = parts.partitioner.Partition(pr.Record, len(mapping))
		}
		if pick < 0 || pick >= len(mapping) {
			cl.finishRecordPromise(pr, fmt.Errorf("invalid record partitioning choice of %d from %d available", pick, len(mapping)))
			return
		}
		partition = mapping[pick]
		partition.records.bufferRecord(pr, false)  
	}
}
type producerID struct {
	id    int64
	epoch int16
	err   error
}
var errReloadProducerID = errors.New("producer id needs reloading")
func (cl *Client) producerID() (int64, int16, error) {
	p := &cl.producer
	id := p.id.Load().(*producerID)
	if id.err == errReloadProducerID {
		p.idMu.Lock()
		defer p.idMu.Unlock()
		if id = p.id.Load().(*producerID); id.err == errReloadProducerID {
			if cl.cfg.disableIdempotency {
				cl.cfg.logger.Log(LogLevelInfo, "skipping producer id initialization because the client was configured to disable idempotent writes")
				id = &producerID{
					id:    -1,
					epoch: -1,
					err:   nil,
				}
				p.id.Store(id)
			} else if cl.cfg.txnID == nil && id.id >= 0 && id.epoch < math.MaxInt16-1 {
				cl.resetAllProducerSequences()
				id = &producerID{
					id:    id.id,
					epoch: id.epoch + 1,
					err:   nil,
				}
				p.id.Store(id)
			} else {
				newID, keep := cl.doInitProducerID(id.id, id.epoch)
				if keep {
					id = newID
					cl.resetAllProducerSequences()
					p.id.Store(id)
				} else {
					id = &producerID{
						id:    id.id,
						epoch: id.epoch,
						err:   errProducerIDLoadFail,
					}
				}
			}
		}
	}
	return id.id, id.epoch, id.err
}
func (cl *Client) resetAllProducerSequences() {
	for _, tp := range cl.producer.topics.load() {
		for _, p := range tp.load().partitions {
			p.records.mu.Lock()
			p.records.needSeqReset = true
			p.records.mu.Unlock()
		}
	}
}
func (cl *Client) failProducerID(id int64, epoch int16, err error) {
	p := &cl.producer
	p.idMu.Lock()
	defer p.idMu.Unlock()
	current := p.id.Load().(*producerID)
	if current.id != id || current.epoch != epoch {
		cl.cfg.logger.Log(LogLevelInfo, "ignoring a fail producer id request due to current id being different",
			"current_id", current.id,
			"current_epoch", current.epoch,
			"current_err", current.err,
			"fail_id", id,
			"fail_epoch", epoch,
			"fail_err", err,
		)
		return  
	}
	p.id.Store(&producerID{
		id:    id,
		epoch: epoch,
		err:   err,
	})
}
func (cl *Client) doInitProducerID(lastID int64, lastEpoch int16) (*producerID, bool) {
	cl.cfg.logger.Log(LogLevelInfo, "initializing producer id")
	req := kmsg.NewPtrInitProducerIDRequest()
	req.TransactionalID = cl.cfg.txnID
	req.ProducerID = lastID
	req.ProducerEpoch = lastEpoch
	if cl.cfg.txnID != nil {
		req.TransactionTimeoutMillis = int32(cl.cfg.txnTimeout.Milliseconds())
	}
	resp, err := req.RequestWith(cl.ctx, cl)
	if err != nil {
		if err == errUnknownRequestKey || err == errBrokerTooOld {
			cl.cfg.logger.Log(LogLevelInfo, "unable to initialize a producer id because the broker is too old or the client is pinned to an old version, continuing without a producer id")
			return &producerID{-1, -1, nil}, true
		}
		if err == errChosenBrokerDead {
			select {
			case <-cl.ctx.Done():
				cl.cfg.logger.Log(LogLevelInfo, "producer id initialization failure due to dying client", "err", err)
				return &producerID{lastID, lastEpoch, ErrClientClosed}, true
			default:
			}
		}
		cl.cfg.logger.Log(LogLevelInfo, "producer id initialization failure, discarding initialization attempt", "err", err)
		return &producerID{lastID, lastEpoch, err}, false
	}
	if err = kerr.ErrorForCode(resp.ErrorCode); err != nil {
		if kerr.IsRetriable(err) {  
			cl.cfg.logger.Log(LogLevelInfo, "producer id initialization resulted in retriable error, discarding initialization attempt", "err", err)
			return &producerID{lastID, lastEpoch, err}, false
		}
		cl.cfg.logger.Log(LogLevelInfo, "producer id initialization errored", "err", err)
		return &producerID{lastID, lastEpoch, err}, true
	}
	cl.cfg.logger.Log(LogLevelInfo, "producer id initialization success", "id", resp.ProducerID, "epoch", resp.ProducerEpoch)
	if cl.producer.idVersion == -1 {
		cl.producer.idVersion = req.Version
	}
	return &producerID{resp.ProducerID, resp.ProducerEpoch, nil}, true
}
func (cl *Client) partitionsForTopicProduce(pr promisedRec) (*topicPartitions, *topicPartitionsData) {
	p := &cl.producer
	topic := pr.Topic
	topics := p.topics.load()
	parts, exists := topics[topic]
	if exists {
		if v := parts.load(); len(v.partitions) > 0 {
			return parts, v
		}
	}
	if !exists {  
		p.topicsMu.Lock()
		defer p.topicsMu.Unlock()
		if parts, exists = p.topics.load()[topic]; !exists {  
			p.unknownTopicsMu.Lock()
			defer p.unknownTopicsMu.Unlock()
			p.topics.storeTopics([]string{topic})
			cl.addUnknownTopicRecord(pr)
			cl.triggerUpdateMetadataNow("forced load due to unknown produce topic")
			return nil, nil
		}
	}
	p.unknownTopicsMu.Lock()
	defer p.unknownTopicsMu.Unlock()
	if v := parts.load(); len(v.partitions) > 0 {
		return parts, v
	}
	cl.addUnknownTopicRecord(pr)
	cl.triggerUpdateMetadata(false, "reload trigger due to produce topic still not known")
	return nil, nil  
}