Skip to content

[improve][ml] Defer cached message metadata parsing until first read #26628

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lhotari merged 1 commit into from
Sep 17, 2026
Merged

[improve][ml] Defer cached message metadata parsing until first read #26628

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,10 @@ protected EntryImpl newObject(Handle<EntryImpl> handle) {
ByteBuf data;
private EntryReadCountHandler readCountHandler;
private boolean decreaseReadCountOnRelease = true;
// Cache readers publish metadata lazily; entry copies must see a fully initialized instance.
@Getter @Setter
private MessageMetadata messageMetadata;
private volatile MessageMetadata messageMetadata;
private boolean messageMetadataInitializationFailed;

private Runnable onDeallocate;

Expand Down Expand Up @@ -290,6 +292,7 @@ protected void deallocate() {
readCountHandler = null;
decreaseReadCountOnRelease = true;
messageMetadata = null;
messageMetadataInitializationFailed = false;
recyclerHandle.recycle(this);
}

Expand All @@ -308,12 +311,14 @@ public void setDecreaseReadCountOnRelease(boolean enabled) {
}

public synchronized void initializeMessageMetadataIfNeeded(String managedLedgerName) {
if (messageMetadata == null) {
if (messageMetadata == null && !messageMetadataInitializationFailed) {
try {
MessageMetadata msgMetadata = new MessageMetadata();
Commands.parseMessageMetadata(data.duplicate(), msgMetadata);
this.messageMetadata = msgMetadata;
} catch (Throwable t) {
// The entry bytes are immutable; another cache reader cannot make a failed parse succeed.
messageMetadataInitializationFailed = true;
log.warn().attr("managedLedgerName", managedLedgerName)
.attr("ledgerId", ledgerId)
.attr("entryId", entryId)
Expand Down
16 changes: 10 additions & 6 deletions ...ed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,10 +169,6 @@ public boolean insert(Entry entry, boolean copy) {
EntryImpl cacheEntry =
EntryImpl.createWithRetainedDuplicate(position, cachedData, entry.getReadCountHandler(),
copy ? null : entry.getMessageMetadata());
if (ml.getConfig().isPulsarMessageEntries()) {
// Parse the message metadata once at insert time so that cache reads don't have to do it lazily
cacheEntry.initializeMessageMetadataIfNeeded(ml.getName());
}
cachedData.release();
if (entries.put(position, cacheEntry, entryLength)) {
totalAddedEntriesSize.add(entryLength);
Expand Down Expand Up @@ -404,7 +400,8 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx2) {
void doAsyncReadEntriesByPosition(ReadHandle lh, Position firstPosition, Position lastPosition, int numberOfEntries,
long maxSizeBytes, IntSupplier expectedReadCount,
final ReadEntriesCallback callback, Object ctx) {
CachedEntries cachedEntries = new CachedEntries(firstPosition.getEntryId(), numberOfEntries);
CachedEntries cachedEntries = new CachedEntries(firstPosition.getEntryId(), numberOfEntries,
ml.getConfig().isPulsarMessageEntries() ? ml.getName() : null);
if (firstPosition.compareTo(lastPosition) == 0) {
ReferenceCountedEntry cachedEntry = entries.get(firstPosition);
if (cachedEntry != null) {
Expand Down Expand Up @@ -508,13 +505,15 @@ void doAsyncReadEntriesByPosition(ReadHandle lh, Position firstPosition, Positio
static final class CachedEntries implements Consumer<ReferenceCountedEntry> {
private final long firstEntryId;
private final int numberOfEntries;
private final String managedLedgerName;
List<Entry> entries;
private int count;
private long totalSize;

CachedEntries(long firstEntryId, int numberOfEntries) {
CachedEntries(long firstEntryId, int numberOfEntries, String managedLedgerName) {
this.firstEntryId = firstEntryId;
this.numberOfEntries = numberOfEntries;
this.managedLedgerName = managedLedgerName;
}

@Override
Expand All @@ -525,6 +524,11 @@ public void accept(ReferenceCountedEntry entry) {
entries.add(null);
}
}
// The visitor retains the cached entry while parsing. Initialize on the shared cached entry
// before copying, so fanout readers reuse one instance backed by the cache-owned buffer.
if (managedLedgerName != null && entry.getMessageMetadata() == null) {
((EntryImpl) entry).initializeMessageMetadataIfNeeded(managedLedgerName);
}
int index = (int) (entry.getPosition().getEntryId() - firstEntryId);
entries.set(index, EntryImpl.create(entry));
count++;
Expand Down
22 changes: 22 additions & 0 deletions managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,11 @@
*/
package org.apache.bookkeeper.mledger.impl;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
Expand All @@ -34,6 +38,24 @@

public class EntryImplTest {

@Test
public void testFailedMetadataInitializationIsNotRetried() {
ByteBuf bytes = Unpooled.buffer(4).writeInt(-1);
EntryImpl entry = EntryImpl.create(1, 0, bytes);
bytes.release();
entry.data = spy(entry.data);
try {
entry.initializeMessageMetadataIfNeeded("ledger");
entry.initializeMessageMetadataIfNeeded("ledger");
assertThat(entry.getMessageMetadata()).isNull();
assertThat(entry.getDataBuffer().readerIndex()).isZero();
assertThat(entry.getDataBuffer().getInt(0)).isEqualTo(-1);
verify(entry.data, times(1)).duplicate();
} finally {
entry.release();
}
}

@Test
public void testCreateWithLedgerIdEntryIdAndByteBuf() {
// Given
Expand Down
112 changes: 95 additions & 17 deletions ...edger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,10 @@
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntSupplier;
import org.apache.bookkeeper.client.LedgerHandle;
Expand Down Expand Up @@ -173,7 +177,7 @@ private static ByteBuf serializeMessage(String producerName) {
}

@Test
public void testInsertParsesMessageMetadata() {
public void testInsertDefersMetadataUntilFirstReadAndSharesIt() {
managedLedgerConfig.setPulsarMessageEntries(true);
ByteBuf headersAndPayload = serializeMessage("producer");
EntryImpl entry = EntryImpl.create(1, 50, headersAndPayload);
Expand All @@ -183,14 +187,78 @@ public void testInsertParsesMessageMetadata() {
assertThat(rangeEntryCache.insert(entry)).isTrue();
entry.release();

// the metadata is parsed once at insert time. Reading the entry back out of the cache doesn't parse
// anything any more, so this asserts what insert actually stored
ReferenceCountedEntry cached = rangeEntryCache.getEntries().get(PositionFactory.create(1, 50));
assertThat(cached).isNotNull();
assertThat(cached.getMessageMetadata()).isNotNull();
assertThat(cached.getMessageMetadata().getProducerName()).isEqualTo("producer");
assertThat(cached.getMessageMetadata().getSequenceId()).isEqualTo(7);
cached.release();
assertThat(cached.getMessageMetadata()).isNull();
Entry first = readSingleEntryFromCache(1, 50);
Entry second = readSingleEntryFromCache(1, 50);
try {
try {
assertThat(first.getMessageMetadata()).isNotNull().isSameAs(cached.getMessageMetadata());
assertThat(second.getMessageMetadata()).isSameAs(first.getMessageMetadata());
rangeEntryCache.clear();
} finally {
cached.release();
first.release();
}
// The second read still retains the cache-owned buffer after eviction and the first read's release.
assertThat(second.getMessageMetadata().getProducerName()).isEqualTo("producer");
assertThat(second.getMessageMetadata().getSequenceId()).isEqualTo(7);
} finally {
second.release();
}
}

@Test(timeOut = 30_000)
public void testConcurrentCacheReadsShareMetadata() throws Exception {
managedLedgerConfig.setPulsarMessageEntries(true);
ByteBuf bytes = serializeMessage("producer");
EntryImpl source = EntryImpl.create(1, 50, bytes);
bytes.release();
assertThat(rangeEntryCache.insert(source)).isTrue();
source.release();
int readers = 8;
CountDownLatch ready = new CountDownLatch(readers);
CountDownLatch start = new CountDownLatch(1);
List<CompletableFuture<Entry>> reads = new ArrayList<>();
ExecutorService executor = Executors.newFixedThreadPool(readers);
try {
try {
for (int i = 0; i < readers; i++) {
reads.add(CompletableFuture.supplyAsync(() -> {
ready.countDown();
try {
start.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(e);
}
return readSingleEntryFromCache(1, 50);
}, executor));
}
assertThat(ready.await(5, TimeUnit.SECONDS)).isTrue();
} finally {
start.countDown();
}
CompletableFuture.allOf(reads.toArray(CompletableFuture[]::new)).get(5, TimeUnit.SECONDS);
MessageMetadata metadata = reads.get(0).join().getMessageMetadata();
assertThat(metadata).isNotNull();
for (CompletableFuture<Entry> read : reads) {
assertThat(read.join().getMessageMetadata()).isSameAs(metadata);
assertThat(read.join().getMessageMetadata().getSequenceId()).isEqualTo(7);
}
rangeEntryCache.clear();
assertThat(metadata.getProducerName()).isEqualTo("producer");
} finally {
executor.shutdownNow();
executor.awaitTermination(5, TimeUnit.SECONDS);
for (CompletableFuture<Entry> read : reads) {
if (read.isDone() && !read.isCompletedExceptionally()) {
read.join().release();
}
}
rangeEntryCache.clear();
}
}

@Test
Expand Down Expand Up @@ -268,8 +336,10 @@ public void testCachedEntryMetadataStaysReadableWhenEntriesAreCopied() {

ReferenceCountedEntry cached = copyingCache.getEntries().get(PositionFactory.create(1, 50));
assertThat(cached).isNotNull();
// MessageMetadata decodes its string and bytes fields lazily from the buffer it was parsed from, so the
// cached entry must not share metadata that was parsed from the now released source buffer
assertThat(cached.getMessageMetadata()).isNull();
Entry readBack = readSingleEntryFromCache(copyingCache, 1, 50);
readBack.release();
// Metadata is initialized from the retained cache copy, after the source buffer has been released.
assertThat(cached.getMessageMetadata()).isNotNull();
assertThat(cached.getMessageMetadata().getProducerName()).isEqualTo("producer");
assertThat(cached.getMessageMetadata().getSequenceId()).isEqualTo(7);
Expand Down Expand Up @@ -310,7 +380,7 @@ public void testReadFromStorageDoesNotShareSourceMetadataWithTheCopiedCacheEntry
assertThat(cached).isNotNull();
// the cached entry is backed by a copy of the payload, so it must not share the metadata that was parsed
// from the source buffer
assertThat(cached.getMessageMetadata()).isNotNull().isNotSameAs(sourceMetadata);
assertThat(cached.getMessageMetadata()).isNull();

// overwrite the source payload while it is still referenced, the way the pooled buffer behind it gets
// overwritten once it has been recycled. This turns a leftover dependency on the source buffer into a
Expand All @@ -328,7 +398,10 @@ public void testReadFromStorageDoesNotShareSourceMetadataWithTheCopiedCacheEntry
assertThat(headersAndPayload.refCnt()).isZero();

// MessageMetadata decodes its string and bytes fields lazily from the buffer it was parsed from, so the
// cached entry stays readable only because its metadata was parsed from the buffer the cache owns
// cached entry stays readable only because its metadata is parsed from the buffer the cache owns
Entry readBack = readSingleEntryFromCache(copyingCache, 1, 0);
assertThat(readBack.getMessageMetadata()).isNotNull().isNotSameAs(sourceMetadata);
readBack.release();
assertThat(cached.getMessageMetadata().getProducerName()).isEqualTo("producer");
assertThat(cached.getMessageMetadata().getSequenceId()).isEqualTo(7);
cached.release();
Expand All @@ -352,9 +425,7 @@ public void testInsertDoesNotParseMessageMetadataWhenTheEntriesArentPulsarMessag
assertThat(cached.getMessageMetadata()).isNull();
cached.release();

// reading the entry back through the cache must not parse it either. This is what pins the removal of
// the lazy initialization that RangeCacheEntryWrapper used to do under its write lock, which would have
// defeated skipping the parse at insert time
// Reading back through the cache must also skip metadata initialization for raw ledger entries.
Entry readBack = readSingleEntryFromCache(1, 50);
assertThat(readBack.getMessageMetadata()).isNull();
readBack.release();
Expand All @@ -373,8 +444,11 @@ public void testInsertDoesNotParseMessageMetadataWhenTheEntriesArentPulsarMessag

ReferenceCountedEntry cachedControl = rangeEntryCache.getEntries().get(PositionFactory.create(1, 51));
assertThat(cachedControl).isNotNull();
assertThat(cachedControl.getMessageMetadata()).isNotNull();
assertThat(cachedControl.getMessageMetadata().getProducerName()).isEqualTo("producer");
assertThat(cachedControl.getMessageMetadata()).isNull();
Entry controlRead = readSingleEntryFromCache(1, 51);
assertThat(controlRead.getMessageMetadata()).isNotNull().isSameAs(cachedControl.getMessageMetadata());
assertThat(controlRead.getMessageMetadata().getProducerName()).isEqualTo("producer");
controlRead.release();
cachedControl.release();
}

Expand Down Expand Up @@ -404,8 +478,12 @@ public void testReadFromStorageDoesNotParseMessageMetadataWhenTheEntriesArentPul
* @apiNote the returned entry must be released by the caller
*/
private Entry readSingleEntryFromCache(long ledgerId, long entryId) {
return readSingleEntryFromCache(rangeEntryCache, ledgerId, entryId);
}

private Entry readSingleEntryFromCache(RangeEntryCacheImpl cache, long ledgerId, long entryId) {
CompletableFuture<Entry> future = new CompletableFuture<>();
rangeEntryCache.asyncReadEntry(lh, PositionFactory.create(ledgerId, entryId),
cache.asyncReadEntry(lh, PositionFactory.create(ledgerId, entryId),
new AsyncCallbacks.ReadEntryCallback() {
@Override
public void readEntryComplete(Entry entry, Object ctx) {
Expand Down
Loading
Loading