Method newBrokerProducer

async_producer.go:980–1089

one per broker; also constructs an associated flusher

(broker *Broker)

// one per broker; also constructs an associated flusher
func (p *asyncProducer) newBrokerProducer(broker *Broker) *brokerProducer {
	var (
		input     = make(chan *ProducerMessage)
		bridge    = make(chan *produceSet)
		pending   = make(chan *brokerProducerResponse)
		responses = make(chan *brokerProducerResponse)
	)

	bp := &brokerProducer{
		parent:            p,
		broker:            broker,
		input:             input,
		output:            bridge,
		responses:         responses,
		accumulatingBatch: newProduceSet(p),
		currentRetries:    make(map[string]map[int32]error),
	}
	go withRecover(bp.run)

	// minimal bridge to make the network response `select`able
	go withRecover(func() {
		// Use a wait group to know if we still have in flight requests
		var wg sync.WaitGroup

		for set := range bridge {
			request := set.buildRequest()

			// Count the in flight requests to know when we can close the pending channel safely
			wg.Add(1)
			// capture the muted set. unmuting is deferred to handleResponse to ensure that
			// retries block subsequent batches for the same partition.
			mutedSet := set
			sendResponse := func(response *ProduceResponse, err error) {
				pending <- &brokerProducerResponse{
					set: mutedSet,
					err: err,
					res: response,
				}
				wg.Done()
			}

			if p.IsTransactional() {
				// Add partition to tx before sending current batch
				err := p.txnmgr.publishTxnPartitions()
				if err != nil {
					// Request failed to be sent
					sendResponse(nil, err)
					continue
				}
			}

			// Use AsyncProduce vs Produce to not block waiting for the response
			// so that we can pipeline multiple produce requests and achieve higher throughput, see:
			// https://kafka.apache.org/protocol#protocol_network
			err := broker.AsyncProduce(request, sendResponse)
			if err != nil {
				// Request failed to be sent
				sendResponse(nil, err)