bugfix: AsyncProducer retryBatch causes goroutines to be leaked

k-wall · k-wall · commit de2038ad79d3 · 2022-04-11T11:53:45.000+01:00
retryBatch caused the brokerRefs count to be incremented (getBrokerProducer), but it is never decremented again, so the
goroutines related to the brokerProducer are leaked.
diff --git a/async_producer.go b/async_producer.go
@@ -1064,6 +1064,7 @@ func (p *asyncProducer) retryBatch(topic string, partition int32, pSet *partitio
 	}
 	bp := p.getBrokerProducer(leader)
 	bp.output <- produceSet
+	p.unrefBrokerProducer(leader, bp)
 }
 
 func (bp *brokerProducer) handleError(sent *produceSet, err error) {

Original file line number	Diff line number	Diff line change
`@@ -1064,6 +1064,7 @@ func (p asyncProducer) retryBatch(topic string, partition int32, pSet partitio`
`1064`	`1064`	`}`
`1065`	`1065`	`bp := p.getBrokerProducer(leader)`
`1066`	`1066`	`bp.output <- produceSet`
	`1067`	`+ p.unrefBrokerProducer(leader, bp)`
`1067`	`1068`	`}`
`1068`	`1069`
`1069`	`1070`	`func (bp brokerProducer) handleError(sent produceSet, err error) {`