(pulsar) branch branch-4.0 updated: [fix][client] Fix memory leak when message size exceeds max message size and batching is enabled (#23967)

[mmerli]

This is an automated email from the ASF dual-hosted git repository.

mmerli pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-4.0 by this push:
     new 866a82b2620 [fix][client] Fix memory leak when message size exceeds 
max message size and batching is enabled (#23967)
866a82b2620 is described below

commit 866a82b2620ab852d96f32c7623f024eba1fbd59
Author: Lari Hotari <lhot...@users.noreply.github.com>
AuthorDate: Thu Feb 13 01:33:55 2025 +0200

    [fix][client] Fix memory leak when message size exceeds max message size 
and batching is enabled (#23967)
---
 .../java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java   | 2 ++
 1 file changed, 2 insertions(+)

diff --git 
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
index 7262cfd11e0..403a804b605 100644
--- 
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
+++ 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
@@ -273,6 +273,7 @@ class BatchMessageContainerImpl extends 
AbstractBatchMessageContainer {
 
             // handle mgs size check as non-batched in 
`ProducerImpl.isMessageSizeExceeded`
             if (op.getMessageHeaderAndPayloadSize() > getMaxMessageSize()) {
+                cmd.release();
                 producer.semaphoreRelease(1);
                 producer.client.getMemoryLimitController().releaseMemory(
                         messages.get(0).getUncompressedSize() + 
batchAllocatedSizeBytes);
@@ -286,6 +287,7 @@ class BatchMessageContainerImpl extends 
AbstractBatchMessageContainer {
         ByteBuf encryptedPayload = producer.encryptMessage(messageMetadata, 
getCompressedBatchMetadataAndPayload());
         updateAndReserveBatchAllocatedSize(encryptedPayload.capacity());
         if (encryptedPayload.readableBytes() > getMaxMessageSize()) {
+            encryptedPayload.release();
             producer.semaphoreRelease(messages.size());
             messages.forEach(msg -> producer.client.getMemoryLimitController()
                     .releaseMemory(msg.getUncompressedSize()));