func (s *sink) createReq(id int64, epoch int16) (*produceRequest, *kmsg.AddPartitionsToTxnRequest, bool) {
	req := &produceRequest{
		txnID:   s.cl.cfg.txnID,
		acks:    s.cl.cfg.acks.val,
		timeout: int32(s.cl.cfg.produceTimeout.Milliseconds()),
		batches: make(seqRecBatches, 5),
		producerID:    id,
		producerEpoch: epoch,
		compressor: s.cl.compressor,
		wireLength:      s.cl.baseProduceRequestLength(),  
		wireLengthLimit: s.cl.cfg.maxBrokerWriteBytes,
	}
	txnBuilder := txnReqBuilder{
		txnID: req.txnID,
		id:    id,
		epoch: epoch,
	}
	var moreToDrain bool
	s.recBufsMu.Lock()
	defer s.recBufsMu.Unlock()
	recBufsIdx := s.recBufsStart
	for i := 0; i < len(s.recBufs); i++ {
		recBuf := s.recBufs[recBufsIdx]
		recBufsIdx = (recBufsIdx + 1) % len(s.recBufs)
		recBuf.mu.Lock()
		if recBuf.failing || len(recBuf.batches) == recBuf.batchDrainIdx || recBuf.inflightOnSink != nil && recBuf.inflightOnSink != s {
			recBuf.mu.Unlock()
			continue
		}
		batch := recBuf.batches[recBuf.batchDrainIdx]
		if added := req.tryAddBatch(atomic.LoadInt32(&s.produceVersion), recBuf, batch); !added {
			recBuf.mu.Unlock()
			moreToDrain = true
			continue
		}
		recBuf.inflightOnSink = s
		recBuf.inflight++
		recBuf.batchDrainIdx++
		recBuf.seq += int32(len(batch.records))
		moreToDrain = moreToDrain || recBuf.tryStopLingerForDraining()
		recBuf.mu.Unlock()
		txnBuilder.add(recBuf)
	}
	if len(s.recBufs) > 0 {
		s.recBufsStart = (s.recBufsStart + 1) % len(s.recBufs)
	}
	return req, txnBuilder.req, moreToDrain
}
createReq returns a produceRequest from currently buffered records
and whether there are more records to create more requests immediately.
package kgo
import (
	"bytes"
	"context"
	"fmt"
	"hash/crc32"
	"strings"
	"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 sink struct {
	cl     *Client  
	nodeID int32    
	inflightSem    atomic.Value
	produceVersion int32  
	drainState workLoop
	seqResps ringSeqResp
	backoffMu   sync.Mutex  
	needBackoff bool
	backoffSeq  uint32  
	consecutiveFailures uint32
	recBufsMu    sync.Mutex  
	recBufs      []*recBuf   
	recBufsStart int         
}
type seqResp struct {
	resp    kmsg.Response
	err     error
	done    chan struct{}
	br      *broker
	promise func(*broker, kmsg.Response, error)
}
func (cl *Client) newSink(nodeID int32) *sink {
	s := &sink{
		cl:             cl,
		nodeID:         nodeID,
		produceVersion: -1,
	}
	s.inflightSem.Store(make(chan struct{}, 1))
	return s
}