fix(producer): prevent deadlock in non-idempotent producer retryBatch

[lbcjbb]

PR #3644 introduced a regression that can cause a SyncProducer to block
indefinitely when using non-idempotent mode with connection retries.

The deadlock scenario:

1. A connection error triggers handleError(), which launches a goroutine
   calling retryBatch() for the failed batch.
2. retryBatch() creates a new brokerProducer via getBrokerProducer() (refcount=1).
3. retryBatch() sends the batch on bp.output, then immediately releases its
   reference via defer unrefBrokerProducer() (refcount=0).
4. With refcount=0, the brokerProducer shuts down while the batch is still
   in flight waiting for a response from the broker.
5. The response is lost, returnSuccesses() / returnErrors() is never called.
6. SendMessages() blocks forever waiting on ProducerMessage.expectation.

The fix stores the brokerProducer reference in the produceSet itself, so it
is only released after handleResponse() has fully processed the batch (success
or retry exhausted). This ensures the brokerProducer stays alive until the
batch lifecycle is complete.

Fixes #3689

[dnwe]

Thanks @lbcjbb for the original issue report. Having dug into this, I don't believe the premature shutdown is actually the failure mode: when the retryBatch-created brokerProducer is un-reffed to zero, its shutdown() still drains and handles every pending response after closing output, so the retried batch's response isn't lost and the partition does get unmuted. The deadlock is that this unmute happens on a goroutine the stranded brokerProducer never hears about, because its run() loop had no case to wake on an unmute. That's what the other PR fixes.

The handleSuccess changes here would also alter non-idempotent retry behaviour beyond the scope of the regression, so it's not obvious what issue they're be fixing or improving?

[lbcjbb]

Yes, I came to the same conclusion. I introduced a delay in the execution of retryBatch(), but the messages are not lost. The brokerProducer is closed/released while a batch is still being processed, but this has no impact now with #3692.

The handleSuccess changes here would also alter non-idempotent retry behaviour beyond the scope of the regression, so it's not obvious what issue they're fixing or improving?

I've included these changes in this PR, but they are more complementary to PR #3644 than directly related to the regression. I'm going to move them to a dedicated PR (#3693).

Thank you for your reply and for taking the time to look into this 🙏