[fix][client]Fix client memory limit currentUsage leak and semaphore release duplicated in ProducerImpl

[Nicklee007]

Motivation

Fix client memory limit currentUsage leak in ProducerImpl. When our pulsar cluster occur some error, producer send message fail and we find the currentUsage always keep high value like the leaked, and cause the producer send rate is slow.
And find producer semaphore release duplicated when createOpSendMsg occur some excrption.

Follow 1 point only release the message count semaphore, but not release the memory limit.
memory limit currentUsage leak point

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

Lines 2031 to 2033 in c217b8f

	final int numMessagesInBatch = batchMessageContainer.getNumMessagesInBatch();
	batchMessageContainer.discard(ex);
	semaphoreRelease(numMessagesInBatch);

producer semaphore release duplicated

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

Lines 2116 to 2120 in 4d64e2e

	} catch (PulsarClientException e) {
	semaphoreRelease(batchMessageContainer.getNumMessagesInBatch());
	} catch (Throwable t) {
	semaphoreRelease(batchMessageContainer.getNumMessagesInBatch());
	log.warn("[{}] [{}] error while create opSendMsg by batch message container", topic, producerName, t);

After the exception the  memory limit leak occured.
org.apache.pulsar.client.api.PulsarClientException$TimeoutException: The producer XXXX-366-15151 can not send message to the topic persistent://XXXX/XXXX/XXXX within given timeout : createdAt 30.005 seconds ago, firstSentAt 30.005 seconds ago, lastSentAt 30.005 seconds ago, retryCount 1
        at org.apache.pulsar.client.impl.ProducerImpl$OpSendMsg.sendComplete(ProducerImpl.java:1287)
        at org.apache.pulsar.client.impl.ProducerImpl.lambda$failPendingMessages$18(ProducerImpl.java:1826)
        at java.base/java.util.ArrayDeque.forEach(ArrayDeque.java:889)
        at org.apache.pulsar.client.impl.ProducerImpl$OpSendMsgQueue.forEach(ProducerImpl.java:1369)
        at org.apache.pulsar.client.impl.ProducerImpl.failPendingMessages(ProducerImpl.java:1816)
        at org.apache.pulsar.client.impl.ProducerImpl.lambda$failPendingMessages$19(ProducerImpl.java:1848)
        at org.apache.pulsar.shade.io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
        at org.apache.pulsar.shade.io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469)
        at org.apache.pulsar.shade.io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:384)
        at org.apache.pulsar.shade.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
        at org.apache.pulsar.shade.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at org.apache.pulsar.shade.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.base/java.lang.Thread.run(Thread.java:834)

Modifications

1. add the MemoryLimitController release.

Documentation

• doc-not-needed
  (Please explain why)

[codelipenghui]

A related one #16835

[codelipenghui]

@Nicklee007 Could you please help add some unit tests?

[hezhangjian]

@codelipenghui +1 we need some tests to cover this change

[Nicklee007]

@Nicklee007 Could you please help add some unit tests?

@codelipenghui @Shoothzj Add some unit test, PTAL Thx~

[Nicklee007]

@codelipenghui @Jason918 @Shoothzj Also find producer semaphore release duplicated in batchMessageAndSend, if some error cause in batchMessageAndSend, the semaphoreRelease will be invoked in failPendingBatchMessages .
Fixed it and add some unit test to cover it.

[Jason918]

LGTM

[mattisonchao]

@Nicklee007 Could you help to open another PR to branch-2.9?

[BewareMyPower]

Remove the release/2.7.5 label because memory limiter was introduced since 2.8, see #8965.

[Nicklee007]

@Nicklee007 Could you help to open another PR to branch-2.9?

@mattisonchao ok, I'll cherry-pick it to branch 2.9.

[BewareMyPower]

Move this PR to the next release of 2.8.5. For the release manager of next 2.8.x release, this PR requires an extra import of Mockito in #16835. However, even after fixing the compilation error, the test still failed. So let's just delay this PR to the next release.

@Nicklee007 If you want to include this PR in 2.8.4, please open a new PR to migrate this PR to branch-2.8.

[Nicklee007]

@Nicklee007 Could you help to open another PR to branch-2.9?

@mattisonchao cherry-pick to PR #16971 in branch 2.9.

[Nicklee007]

Remove the release/2.7.5 label because memory limiter was introduced since 2.8, see #8965.

@BewareMyPower @Jason918 Created a new PR #16972 to branch-2.7 which only fix the semaphore release duplicated problem in ProducerImpl, PTAL.

[Nicklee007]

Move this PR to the next release of 2.8.5. For the release manager of next 2.8.x release, this PR requires an extra import of Mockito in #16835. However, even after fixing the compilation error, the test still failed. So let's just delay this PR to the next release.

@Nicklee007 If you want to include this PR in 2.8.4, please open a new PR to migrate this PR to branch-2.8.

@BewareMyPower Created a new PR #16985 branch-2.8, PTAL.