[pulsar] branch master updated: Use PulsarByteBufAllocator to allocate buffer for chunks at consumer side (#13536)

[xyz]

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

xyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new a5d3473  Use PulsarByteBufAllocator to allocate buffer for chunks at 
consumer side (#13536)
a5d3473 is described below

commit a5d347370286cb67ffa4ee179ec6fb5117546a4e
Author: Yunze Xu <xyzinfern...@163.com>
AuthorDate: Tue Dec 28 20:35:21 2021 +0800

    Use PulsarByteBufAllocator to allocate buffer for chunks at consumer side 
(#13536)
    
    ### Motivation
    
    Currently Pulsar consumer allocates memory from direct memory via 
`Unpooled.directBuffer` directly, which doesn't make use of the widely used 
allocator in Pulsar.
    
    ### Modifications
    
    Use `PulsarByteBufAllocator` as the memory allocator for chunks buffer.
---
 .../src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java      | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git 
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 30ad4ec..946995f 100644
--- 
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -26,7 +26,6 @@ import com.google.common.collect.Iterables;
 import com.scurrilous.circe.checksum.Crc32cIntChecksum;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.util.Recycler;
 import io.netty.util.Recycler.Handle;
@@ -1270,7 +1269,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> 
implements ConnectionHandle
         }
 
         if (msgMetadata.getChunkId() == 0) {
-            ByteBuf chunkedMsgBuffer = 
Unpooled.directBuffer(msgMetadata.getTotalChunkMsgSize(),
+            ByteBuf chunkedMsgBuffer = 
PulsarByteBufAllocator.DEFAULT.buffer(msgMetadata.getTotalChunkMsgSize(),
                     msgMetadata.getTotalChunkMsgSize());
             int totalChunks = msgMetadata.getNumChunksFromMsg();
             chunkedMessagesMap.computeIfAbsent(msgMetadata.getUuid(),