(pulsar) 06/14: [improve][ml] Parse message metadata at entry cache insert time instead of under the wrapper write lock (#26463)

[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 2d600cfc2c154f2960b2ac7693cc79b9cd285e8b
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Tue Sep 8 04:10:18 2026 -0700

    [improve][ml] Parse message metadata at entry cache insert time instead of under the wrapper write lock (#26463)
    
    (cherry picked from commit ce6079887474043f4055ee6c22d2c72e456ac40f)
---
 .../mledger/impl/cache/RangeCacheEntryWrapper.java |  3 +++
 .../mledger/impl/cache/RangeEntryCacheImpl.java    |  6 ++++++
 .../impl/cache/RangeEntryCacheImplTest.java        | 25 ++++++++++++++++++++++
 3 files changed, 34 insertions(+)

diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
index 02d148504baa..569a5135c344 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
@@ -65,6 +65,9 @@ class RangeCacheEntryWrapper {
             entryWrapper.key = key;
             entryWrapper.value = value;
             entryWrapper.size = size;
+            // When the value already carries parsed message metadata, there's no need to lazily initialize it
+            // under the write lock on the first read
+            entryWrapper.messageMetadataInitialized = value.getMessageMetadata() != null;
             // Set the timestamp to the current time in nanoseconds
             // This is used for time-based eviction of entries
             entryWrapper.timestampNanos = System.nanoTime();
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
index 352ac341e560..a68ba55eeb25 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
@@ -151,6 +151,12 @@ public class RangeEntryCacheImpl implements EntryCache {
             cachedData = entry.getDataBuffer().retain();
         }
 
+        // Parse the message metadata once at insert time so that cache reads don't have to do it lazily while
+        // holding the RangeCacheEntryWrapper write lock
+        if (entry instanceof EntryImpl entryImpl) {
+            entryImpl.initializeMessageMetadataIfNeeded(ml.getName());
+        }
+
         Position position = entry.getPosition();
         ReferenceCountedEntry cacheEntry =
                 EntryImpl.createWithRetainedDuplicate(position, cachedData, entry.getReadCountHandler(),
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
index 42ae262e8ddb..813ecc8b6d46 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
@@ -27,7 +27,9 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
@@ -45,6 +47,8 @@ import org.apache.bookkeeper.mledger.impl.EntryImpl;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryMBeanImpl;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerMBeanImpl;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.protocol.Commands;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
@@ -94,6 +98,27 @@ public class RangeEntryCacheImplTest {
         expectedReadCount = () -> 1;
     }
 
+    @Test
+    public void testInsertParsesMessageMetadata() {
+        MessageMetadata metadata = new MessageMetadata()
+                .setProducerName("producer")
+                .setSequenceId(7)
+                .setPublishTime(123456789L);
+        ByteBuf headersAndPayload = Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, metadata,
+                Unpooled.copiedBuffer("payload", StandardCharsets.UTF_8));
+        EntryImpl entry = EntryImpl.create(1, 50, headersAndPayload);
+        headersAndPayload.release();
+        assertThat(entry.getMessageMetadata()).isNull();
+
+        assertThat(rangeEntryCache.insert(entry)).isTrue();
+
+        // the metadata is parsed once at insert time instead of lazily on the first cache read
+        assertThat(entry.getMessageMetadata()).isNotNull();
+        assertThat(entry.getMessageMetadata().getProducerName()).isEqualTo("producer");
+        assertThat(entry.getMessageMetadata().getSequenceId()).isEqualTo(7);
+        entry.release();
+    }
+
     @Test
     public void testPartialCachingWithMiddleEntryInCache() {
         Entry entry = EntryImpl.create(1, 50, Unpooled.EMPTY_BUFFER);