(pulsar) 02/14: [improve][broker] Copy small entries into a flat buffer when prepending broker entry metadata (#26464)

[lh...@apache.org]

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

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

commit 77bab3bb4ae829fc7adc7c71b719cd3c419c5175
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Fri Sep 4 14:56:14 2026 -0700

    [improve][broker] Copy small entries into a flat buffer when prepending broker entry metadata (#26464)
    
    (cherry picked from commit 7010231da456da96d49dff50f90225d196451f2a)
---
 .../apache/pulsar/common/protocol/Commands.java    | 25 ++++++++++++++++++++++
 .../pulsar/common/protocol/CommandUtilsTest.java   | 24 +++++++++++++++++++++
 2 files changed, 49 insertions(+)

diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
index 275b6160061f..1b43686b33a4 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
@@ -132,6 +132,13 @@ public class Commands {
     public static final short magicCrc32c = 0x0e01;
     @SuppressWarnings("checkstyle:ConstantName")
     public static final short magicBrokerEntryMetadata = 0x0e02;
+
+    /**
+     * Payloads up to this size are copied into a single buffer when broker entry metadata is prepended; larger ones
+     * are wrapped in a composite instead. Matches the size below which the BookKeeper client copies entries anyway.
+     */
+    @VisibleForTesting
+    static final int BROKER_ENTRY_METADATA_COPY_THRESHOLD = 16 * 1024;
     private static final int checksumSize = 4;
 
     @VisibleForTesting
@@ -1761,6 +1768,24 @@ public class Commands {
         }
 
         int brokerMetaSize = brokerEntryMetadata.getSerializedSize();
+        int payloadSize = headerAndPayload.readableBytes();
+
+        if (payloadSize <= BROKER_ENTRY_METADATA_COPY_THRESHOLD) {
+            // For small payloads, produce a single contiguous buffer. A composite reaching the socket goes through
+            // nioBuffers(), which allocates two NIO views per component on every flush (BK write and each consumer
+            // dispatch), and makes every metadata parse walk the components. At this size the copy is cheaper than
+            // that, and it happens once at publish while the entry is written and parsed many more times.
+            int totalSize = 6 + brokerMetaSize + payloadSize;
+            ByteBuf entry = PulsarByteBufAllocator.DEFAULT.buffer(totalSize, totalSize);
+            entry.writeShort(Commands.magicBrokerEntryMetadata);
+            entry.writeInt(brokerMetaSize);
+            brokerEntryMetadata.writeTo(entry);
+            entry.writeBytes(headerAndPayload, headerAndPayload.readerIndex(), payloadSize);
+            headerAndPayload.release();
+            return entry;
+        }
+
+        // For large payloads the copy would cost more than the per-flush view allocations it saves.
         ByteBuf brokerMeta =
                 PulsarByteBufAllocator.DEFAULT.buffer(brokerMetaSize + 6, brokerMetaSize + 6);
         brokerMeta.writeShort(Commands.magicBrokerEntryMetadata);
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandUtilsTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandUtilsTest.java
index baea2b3f0279..8bf2d6f389ca 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandUtilsTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandUtilsTest.java
@@ -148,6 +148,30 @@ public class CommandUtilsTest {
         dataWithBrokerEntryMetadata.release();
     }
 
+    @Test
+    public void testAddBrokerEntryMetadataLargePayload() throws Exception {
+        int mockBatchSize = 10;
+        byte[] data = new byte[Commands.BROKER_ENTRY_METADATA_COPY_THRESHOLD + 1];
+        for (int i = 0; i < data.length; i++) {
+            data[i] = (byte) i;
+        }
+        ByteBuf byteBuf = PulsarByteBufAllocator.DEFAULT.buffer(data.length, data.length);
+        byteBuf.writeBytes(data);
+        ByteBuf dataWithBrokerEntryMetadata =
+                Commands.addBrokerEntryMetadata(byteBuf, getBrokerEntryMetadataInterceptors(), mockBatchSize);
+        assertTrue(dataWithBrokerEntryMetadata instanceof CompositeByteBuf);
+
+        BrokerEntryMetadata brokerMetadata =
+                Commands.parseBrokerEntryMetadataIfExist(dataWithBrokerEntryMetadata);
+        assertEquals(brokerMetadata.getIndex(), mockBatchSize - 1);
+        assertEquals(data.length, dataWithBrokerEntryMetadata.readableBytes());
+
+        byte[] content = new byte[dataWithBrokerEntryMetadata.readableBytes()];
+        dataWithBrokerEntryMetadata.readBytes(content);
+        assertEquals(content, data);
+        dataWithBrokerEntryMetadata.release();
+    }
+
     @Test
     public void testSkipBrokerEntryMetadata() throws Exception {
         String data = "test-message";