From de2038ad79d3260877a3f873a6a0c67261e4a5ae Mon Sep 17 00:00:00 2001
From: kwall <kwall@apache.org>
Date: Mon, 11 Apr 2022 11:47:10 +0100
Subject: [PATCH] bugfix: AsyncProducer retryBatch causes goroutines to be
 leaked

retryBatch caused the brokerRefs count to be incremented (getBrokerProducer), but it is never decremented again, so the
goroutines related to the brokerProducer are leaked.
---
 async_producer.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/async_producer.go b/async_producer.go
index 343d3de9bd..ad0c789f90 100644
--- 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) {