[fix][client] Prevent retry topic and dead letter topic producer leaks when sending of message fails

[lhotari]

Fixes #20635

Motivation

The client currently leaks producers when sending of the message fails and a new producer is created after an exception occurs. The previous producer wasn't closed in these cases.
In case of such failures, there should be a retry backoff so that such cases don't cause excessive load on the broker side.

Modifications

• properly close retry topic and dead letter topic producers when they are discarded
• add backoff for the creation of new producers when there are subsequent failures. Reset the backoff when a message is sent successfully.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[Technoboy-]

LGTM

[shibd]

Left a small comment.

And I have question about retry create deadletterProducer or retryProducer.

1. For schema exceptions mentioned in the ticket, retries won't solve the issue, right? So what's the point?
2. If there are network issues, the Producer has automatic retries internally.

I'm wondering why we need to implement a retry mechanism here, or why the original logic sets the deadletterProducer=null. If creation truly fails, shouldn't the user handle it?

[lhotari]

Left a small comment.

And I have question about retry create deadletterProducer or retryProducer.

1. For schema exceptions mentioned in the ticket, retries won't solve the issue, right? So what's the point?

Thanks for the review feedback, @shibd.

Correct. This PR doesn't try to resolve that. The point is that it's a way to reproduce the issue reported in #20635 (comment) . The test case has that comment too, but I guess I should have been more verbose about it.

2. If there are network issues, the Producer has automatic retries internally.

I'm wondering why we need to implement a retry mechanism here, or why the original logic sets the deadletterProducer=null. If creation truly fails, shouldn't the user handle it?

That's a valid point. The original intention of this PR was to address the producer leak without changing the existing logic. However, the original logic is not great in the first place when a new producer is created whenever an exception happens in message sending.

[shibd]

hi, @lhotari

Thanks for the discussion. So, can we directly remove the logic that sets deadLetterProducer = null when creation exception?

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 729 in 109f042

deadLetterProducer = null;

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 2240 in 109f042

deadLetterProducer = null;

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 752 in 109f042

retryLetterProducer = null;

This would simplify things, as re-create doesn't seem very meaningful to me.

[lhotari]

hi, @lhotari

Thanks for the discussion. So, can we directly remove the logic that sets deadLetterProducer = null when creation exception?

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 729 in 109f042

deadLetterProducer = null;

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 2240 in 109f042

deadLetterProducer = null;

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 752 in 109f042

retryLetterProducer = null;

This would simplify things, as re-create doesn't seem very meaningful to me.

@shibd Sure, that's a question that comes into mind. There seems to be a reason why it has been reset to null in the first place. I'll check the commit history to find out.

[shibd]

/LGTM