(pulsar) branch master updated: [fix][broker] Fix BucketDelayedDeliveryTracker recovery after LightProto migration (#26160)

[zixuan]

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

nodece 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 a74268dc2fa7 [fix][broker] Fix BucketDelayedDeliveryTracker recovery 
after LightProto migration (#26160)
a74268dc2fa7 is described below

commit a74268dc2fa787cf657713aa3fea36f66a59b782
Author: Zixuan Liu <node...@gmail.com>
AuthorDate: Fri Jul 10 09:41:47 2026 +0800

    [fix][broker] Fix BucketDelayedDeliveryTracker recovery after LightProto 
migration (#26160)
---
 .../bucket/BookkeeperBucketSnapshotStorage.java    |  9 +++++-
 .../BookkeeperBucketSnapshotStorageTest.java       | 37 ++++++++++++++++++++++
 2 files changed, 45 insertions(+), 1 deletion(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
index 09bbb23d8c03..5c8b2d13dc00 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.broker.delayed.bucket;
 
+import com.google.common.annotations.VisibleForTesting;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import java.util.ArrayList;
@@ -37,6 +38,7 @@ import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.delayed.proto.SnapshotMetadata;
 import org.apache.pulsar.broker.delayed.proto.SnapshotSegment;
+import org.apache.pulsar.broker.delayed.proto.SnapshotSegmentMetadata;
 import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.jspecify.annotations.NonNull;
@@ -134,12 +136,17 @@ public class BookkeeperBucketSnapshotStorage implements 
BucketSnapshotStorage {
         return FutureUtil.waitForAll(addFutures);
     }
 
-    private SnapshotMetadata parseSnapshotMetadataEntry(LedgerEntry 
ledgerEntry) {
+    @VisibleForTesting
+    public SnapshotMetadata parseSnapshotMetadataEntry(LedgerEntry 
ledgerEntry) {
         ByteBuf entryBuffer = null;
         try {
             entryBuffer = ledgerEntry.getEntryBuffer();
             SnapshotMetadata metadata = new SnapshotMetadata();
             metadata.parseFrom(entryBuffer, entryBuffer.readableBytes());
+            for (int i = 0; i < metadata.getMetadataListCount(); i++) {
+                SnapshotSegmentMetadata segment = metadata.getMetadataAt(i);
+                
segment.forEachDelayedIndexBitMap(segment::putDelayedIndexBitMap);
+            }
             return metadata;
         } finally {
             if (entryBuffer != null) {
diff --git 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/BookkeeperBucketSnapshotStorageTest.java
 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/BookkeeperBucketSnapshotStorageTest.java
index 964b0e40f88f..82396e66fa3d 100644
--- 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/BookkeeperBucketSnapshotStorageTest.java
+++ 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/BookkeeperBucketSnapshotStorageTest.java
@@ -18,12 +18,21 @@
  */
 package org.apache.pulsar.broker.delayed;
 
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+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.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.LedgerEntry;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.delayed.bucket.BookkeeperBucketSnapshotStorage;
 import org.apache.pulsar.broker.delayed.proto.DelayedIndex;
@@ -230,4 +239,32 @@ public class BookkeeperBucketSnapshotStorageTest extends 
MockedPulsarServiceBase
         FutureUtil.waitForAll(futures).join();
     }
 
+    @Test
+    public void testParseSnapshotMetadataEntryMaterializesLazyBytes() {
+        long ts = System.currentTimeMillis();
+        SnapshotSegmentMetadata segmentMetadata = new 
SnapshotSegmentMetadata();
+        segmentMetadata.setMinScheduleTimestamp(ts);
+        segmentMetadata.setMaxScheduleTimestamp(ts);
+        segmentMetadata.putDelayedIndexBitMap(100L, 
"payload-1".getBytes(StandardCharsets.UTF_8));
+        segmentMetadata.putDelayedIndexBitMap(200L, 
"payload-2".getBytes(StandardCharsets.UTF_8));
+        SnapshotMetadata original = new SnapshotMetadata();
+        original.addMetadata().copyFrom(segmentMetadata);
+
+        ByteBuf buffer = Unpooled.wrappedBuffer(original.toByteArray());
+        assertEquals(buffer.refCnt(), 1);
+
+        LedgerEntry entry = mock(LedgerEntry.class);
+        when(entry.getEntryBuffer()).thenReturn(buffer);
+
+        SnapshotMetadata parsed = 
bucketSnapshotStorage.parseSnapshotMetadataEntry(entry);
+        assertEquals(buffer.refCnt(), 0);
+
+        AtomicInteger entryCount = new AtomicInteger();
+        parsed.getMetadataAt(0).forEachDelayedIndexBitMap((k, v) -> {
+            assertNotNull(v);
+            assertTrue(v.length > 0);
+            entryCount.incrementAndGet();
+        });
+        assertEquals(entryCount.get(), 2);
+    }
 }