You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2026/09/08 12:30:21 UTC
[PR] [fix][ml] Parse the cached entry's message metadata from the buffer the cache keeps [pulsar]
lhotari opened a new pull request, #26495:
URL: https://github.com/apache/pulsar/pull/26495
Fixes #26493
### Motivation
`RangeEntryCacheImpl.insert` stores on the **cached** entry a `MessageMetadata` instance that was parsed from the **incoming** entry's buffer. That is only sound when the two share a buffer, and with `managedLedgerCacheCopyEntries=true` they don't: the cached entry is backed by a fresh copy from `copyEntry(entry)` while the incoming entry is released right after the insert returns.
It matters because a `MessageMetadata` keeps a reference to the buffer it was parsed from and decodes part of itself out of it lazily. lightproto's generated `parseFrom` ends with
```java
_parsedBuffer = _buffer; // no retain, no copy
```
and `producerName`, `replicatedFrom`, `partitionKey`, `replicateTo`, `encryptionAlgo`, `encryptionParam`, `schemaVersion`, `orderingKey`, `uuid`, `schemaId` and every `properties` key/value are decoded from `_parsedBuffer` on first access. Scalars such as `sequenceId` and `publishTime` are decoded eagerly by `parseFrom`, which is what makes the failure selective rather than obvious.
So for as long as the entry stays cached, those fields are read out of a released, pool-recycled buffer. They are read on hot paths — `Commands.resolveStickyKey` via `AbstractBaseDispatcher.peekStickyKey`, the metadata handed to entry filters in `filterEntriesForConsumer`, and `PersistentTopicsBase`'s `peek`/`examine` — so the symptoms are wrong Key_Shared routing or wrong filter decisions rather than anything that points at a buffer lifetime problem.
Two paths reach it:
* **Read path** — `readFromStorage` parses the metadata on each entry it creates and then inserts that entry into the cache, while the entries it returns are released once dispatch completes. Present since #24682 + #24836, so `branch-4.2` is affected too.
* **Write path** — since #26463 `insert` also parses the metadata on the incoming entry, and `OpAddEntry.run` releases that entry, and then its buffer in `completeAdd`, right after inserting. `master` only.
`managedLedgerCacheCopyEntries` defaults to `false`, so a default broker is not affected.
### Modifications
- Parse the metadata on the **cache** entry rather than on the incoming one, and only inherit the incoming entry's instance when the cache retains that same buffer (`copyEntries == false`).
- The parse stays ahead of `entries.put`, because `RangeCacheEntryWrapper.withNewInstance` reads `value.getMessageMetadata() != null` to decide whether anything still needs lazy initialization.
- Nothing depended on `insert` parsing the *source* entry: `readFromStorage` already parses the entries it returns, and the write path releases the incoming entry immediately. Dropping that parse also removes a redundant one on the read path.
- Added a package-private `@VisibleForTesting RangeEntryCacheImpl#getEntries()` so tests can assert on what the cache actually holds rather than on the entry that was handed to it.
The invariant this establishes, and which is worth keeping in mind when touching `EntryImpl.messageMetadata`, is that an entry's message metadata must always have been parsed from that entry's own `data` buffer.
### Verifying this change
Added `RangeEntryCacheImplTest.testCachedEntryMetadataStaysReadableWhenEntriesAreCopied`, which inserts into a `copyEntries=true` cache, releases the source entry exactly as `OpAddEntry` does on the write path, and then reads a lazily decoded field off the cached entry. Without the fix it fails with:
```
io.netty.util.IllegalReferenceCountException: refCnt: 0
at org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImplTest
.testCachedEntryMetadataStaysReadableWhenEntriesAreCopied(RangeEntryCacheImplTest.java:155)
```
An exception is the lucky outcome — the cached metadata holds an unretained duplicate that shares its parent's reference count, so once the pooled parent has been released and reused the duplicate sees a healthy `refCnt` again and simply decodes whatever now occupies that memory.
Also added `RangeEntryCacheImplTest.testInsertReusesTheMessageMetadataTheEntryAlreadyCarries`, which supplies a sentinel instance whose `producerName` deliberately disagrees with the serialized buffer, so that a silent re-parse cannot pass it. `testInsertParsesMessageMetadata` (added in #26463) now asserts on the cached entry instead of the incoming one, which is what it was really about.
### Does this pull request potentially affect one of the following parts:
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### Documentation
- [ ] `doc`
- [ ] `doc-required`
- [x] `doc-not-needed`
- [ ] `doc-complete`
### Matching PR in forked repository
PR in forked repository: N/A — the change is covered by the unit tests above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Parse the cached entry's message metadata from the buffer the cache keeps [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26495:
URL: https://github.com/apache/pulsar/pull/26495#discussion_r3961639909
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java:
##########
@@ -90,33 +95,93 @@ public void setup() {
callback.readEntriesComplete(entries, ctx);
return null;
}).when(pendingReadsManager).readEntries(any(), anyLong(), anyLong(), any(), any(), any());
- rangeEntryCache =
- new RangeEntryCacheImpl(mockEntryCacheManager, mockManagedLedger, false, mockRangeCacheRemovalQueue,
- EntryLengthFunction.DEFAULT, pendingReadsManager);
+ rangeEntryCache = createRangeEntryCache(false);
lh = mock(ReadHandle.class);
when(lh.getId()).thenReturn(1L);
expectedReadCount = () -> 1;
}
- @Test
- public void testInsertParsesMessageMetadata() {
+ private RangeEntryCacheImpl createRangeEntryCache(boolean copyEntries) {
+ return new RangeEntryCacheImpl(mockEntryCacheManager, mockManagedLedger, copyEntries,
+ mockRangeCacheRemovalQueue, EntryLengthFunction.DEFAULT, pendingReadsManager);
+ }
+
+ private static ByteBuf serializeMessage(String producerName) {
MessageMetadata metadata = new MessageMetadata()
- .setProducerName("producer")
+ .setProducerName(producerName)
.setSequenceId(7)
.setPublishTime(123456789L);
- ByteBuf headersAndPayload = Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, metadata,
+ return Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, metadata,
Unpooled.copiedBuffer("payload", StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void testInsertParsesMessageMetadata() {
+ ByteBuf headersAndPayload = serializeMessage("producer");
EntryImpl entry = EntryImpl.create(1, 50, headersAndPayload);
headersAndPayload.release();
assertThat(entry.getMessageMetadata()).isNull();
assertThat(rangeEntryCache.insert(entry)).isTrue();
+ entry.release();
- // 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);
+ // the metadata is parsed once at insert time instead of lazily on the first cache read. This has to be
+ // asserted before reading the entry back, because reading it would itself trigger the lazy path
+ assertThat(rangeEntryCache.getEntries().isMessageMetadataInitialized(PositionFactory.create(1, 50)))
+ .isTrue();
+
+ 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();
+ }
+
+ @Test
+ public void testInsertReusesTheMessageMetadataTheEntryAlreadyCarries() {
+ ByteBuf headersAndPayload = serializeMessage("in-buffer");
+ EntryImpl entry = EntryImpl.create(1, 50, headersAndPayload);
+ headersAndPayload.release();
+ // deliberately disagrees with the buffer, so that a silent re-parse cannot pass this test
+ MessageMetadata suppliedByTheCaller = new MessageMetadata()
+ .setProducerName("from-publish-path")
+ .setSequenceId(7)
+ .setPublishTime(123456789L);
+ entry.setMessageMetadata(suppliedByTheCaller);
+
+ assertThat(rangeEntryCache.insert(entry)).isTrue();
entry.release();
+
+ // the cached entry reuses the instance instead of parsing the same bytes a second time
+ ReferenceCountedEntry cached = rangeEntryCache.getEntries().get(PositionFactory.create(1, 50));
+ assertThat(cached).isNotNull();
+ assertThat(cached.getMessageMetadata()).isSameAs(suppliedByTheCaller);
+ assertThat(cached.getMessageMetadata().getProducerName()).isEqualTo("from-publish-path");
+ cached.release();
+ }
+
+ @Test
+ public void testCachedEntryMetadataStaysReadableWhenEntriesAreCopied() {
Review Comment:
You're right, and the gap turned out to be bigger than a missing scenario. I reverted the guard (`copyEntries ? null : entry.getMessageMetadata()` back to plain `entry.getMessageMetadata()`) and the whole class still passed, `testCachedEntryMetadataStaysReadableWhenEntriesAreCopied` included: it hands `insert()` an entry whose metadata is still `null`, so both arms of the ternary evaluate to `null` and the guard is never exercised. That test does fail against the full pre-fix `insert()`, but on its own it doesn't pin the guard.
Added `testReadFromStorageDoesNotShareSourceMetadataWithTheCopiedCacheEntry`, which drives the real `readFromStorage()` path with `copyEntries=true`, so `insert()` receives an entry whose metadata was already parsed from the source buffer. It asserts the cached entry's `MessageMetadata` is a different instance from the source entry's, then zeroes the source payload while it is still referenced and reads `producerName` back off the cached entry after the source has been released — so a leftover dependency on the source buffer shows up as a wrong value rather than depending on when the pooled buffer happens to be reused. An `isNotEqualTo` check on the source metadata keeps that from going vacuous if `MessageMetadata` ever stops decoding those fields lazily.
Mutation results, each run with `--rerun-tasks --no-build-cache -PtestFailFast=false -PtestRetryCount=0`:
| mutation | outcome |
|---|---|
| guard reverted to `entry.getMessageMetadata()` | 1 failure, only the new test (`isNotSameAs`), 5/5 runs |
| exact pre-fix `insert()` shape | 2 failures: the new test, and `testCachedEntryMetadataStaysReadableWhenEntriesAreCopied` with `IllegalReferenceCountException: refCnt: 0` |
| insert-time parse moved back onto the source entry | 1 failure, only the new test (`isMessageMetadataInitialized`) |
| `entry.initializeMessageMetadataIfNeeded(...)` dropped from `readFromStorage` | 1 failure, only the new test |
Also dropped the `instanceof` around the insert-time parse in a follow-up commit: it was needed when it tested the incoming `Entry`, but after the parse moved onto `cacheEntry` it is statically always true, since `EntryImpl.createWithRetainedDuplicate` is declared to return `EntryImpl`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Parse the cached entry's message metadata from the buffer the cache keeps [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26495:
URL: https://github.com/apache/pulsar/pull/26495#issuecomment-5593139345
There's one more issue: when the entry doesn't contain a message, parsing the metadata will throw an exception (which is caught, however that will cause an overhead). This happens for MLTransactionLogImpl and MLPendingAckStore which store BatchedTransactionMetadataEntry and PendingAckMetadataEntry bytes directly to the managed ledger. There's no "magic bytes" as a header to differentiate these entries from ordinary message entries in a managed ledger.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Parse the cached entry's message metadata from the buffer the cache keeps [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26495:
URL: https://github.com/apache/pulsar/pull/26495#issuecomment-5590897422
@merlimat Thanks, I looked into it. `materialize()` does remove the buffer dependency, but it can't be applied to `entry.getMessageMetadata()` at this call site: on the read path that instance belongs to the entry `readFromStorage` is about to hand to the dispatcher — it is parsed at `entry.initializeMessageMetadataIfNeeded(ml.getName())` and the same entry is passed to `insert()` four lines later — and `materialize()` mutates it: `_parsedBuffer` goes `null` and every `*BufferIdx` is reset to `-1`, which also flips `_writeTo` from the byte-exact `_parsedBuffer.getBytes(...)` passthrough to re-encoding from the decoded `String`. So the cache needs its own copy either way.
`copyFrom` would be a correct copy — it does come out fully buffer-independent, since the bytes fields go through `getSchemaVersion()`'s `byte[]` copy and every setter clears its `*BufferIdx`. The cost is that it eagerly decodes everything: a `String` per string field, a `String` per property key *and* value, a `byte[]` + `Unpooled` copy per bytes field, and it memoizes into the source instance on the way through the getters. Re-parsing from the buffer the cache already owns yields the same independent instance while leaving the string and bytes fields lazy, so nothing is decoded that no consumer reads. The allocation profile also differs in a way that matters for a cache: the second parse's temporaries die with the read callback, whereas everything `materialize()` or `copyFrom` decodes stays live for the entry's whole cache residency.
Where `materialize()` would be the right tool is if the cached metadata had to outlive the cache's own buffer — here it doesn't, since the cache entry retains that buffer for as long as it is cached and reader copies `retainedDuplicate()` it. So I've kept the re-parse, but happy to switch if you read the tradeoff differently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Parse the cached entry's message metadata from the buffer the cache keeps [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece merged PR #26495:
URL: https://github.com/apache/pulsar/pull/26495
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Parse the cached entry's message metadata from the buffer the cache keeps [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26495:
URL: https://github.com/apache/pulsar/pull/26495#discussion_r3958288202
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java:
##########
@@ -90,33 +95,93 @@ public void setup() {
callback.readEntriesComplete(entries, ctx);
return null;
}).when(pendingReadsManager).readEntries(any(), anyLong(), anyLong(), any(), any(), any());
- rangeEntryCache =
- new RangeEntryCacheImpl(mockEntryCacheManager, mockManagedLedger, false, mockRangeCacheRemovalQueue,
- EntryLengthFunction.DEFAULT, pendingReadsManager);
+ rangeEntryCache = createRangeEntryCache(false);
lh = mock(ReadHandle.class);
when(lh.getId()).thenReturn(1L);
expectedReadCount = () -> 1;
}
- @Test
- public void testInsertParsesMessageMetadata() {
+ private RangeEntryCacheImpl createRangeEntryCache(boolean copyEntries) {
+ return new RangeEntryCacheImpl(mockEntryCacheManager, mockManagedLedger, copyEntries,
+ mockRangeCacheRemovalQueue, EntryLengthFunction.DEFAULT, pendingReadsManager);
+ }
+
+ private static ByteBuf serializeMessage(String producerName) {
MessageMetadata metadata = new MessageMetadata()
- .setProducerName("producer")
+ .setProducerName(producerName)
.setSequenceId(7)
.setPublishTime(123456789L);
- ByteBuf headersAndPayload = Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, metadata,
+ return Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, metadata,
Unpooled.copiedBuffer("payload", StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void testInsertParsesMessageMetadata() {
+ ByteBuf headersAndPayload = serializeMessage("producer");
EntryImpl entry = EntryImpl.create(1, 50, headersAndPayload);
headersAndPayload.release();
assertThat(entry.getMessageMetadata()).isNull();
assertThat(rangeEntryCache.insert(entry)).isTrue();
+ entry.release();
- // 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);
+ // the metadata is parsed once at insert time instead of lazily on the first cache read. This has to be
+ // asserted before reading the entry back, because reading it would itself trigger the lazy path
+ assertThat(rangeEntryCache.getEntries().isMessageMetadataInitialized(PositionFactory.create(1, 50)))
+ .isTrue();
+
+ 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();
+ }
+
+ @Test
+ public void testInsertReusesTheMessageMetadataTheEntryAlreadyCarries() {
+ ByteBuf headersAndPayload = serializeMessage("in-buffer");
+ EntryImpl entry = EntryImpl.create(1, 50, headersAndPayload);
+ headersAndPayload.release();
+ // deliberately disagrees with the buffer, so that a silent re-parse cannot pass this test
+ MessageMetadata suppliedByTheCaller = new MessageMetadata()
+ .setProducerName("from-publish-path")
+ .setSequenceId(7)
+ .setPublishTime(123456789L);
+ entry.setMessageMetadata(suppliedByTheCaller);
+
+ assertThat(rangeEntryCache.insert(entry)).isTrue();
entry.release();
+
+ // the cached entry reuses the instance instead of parsing the same bytes a second time
+ ReferenceCountedEntry cached = rangeEntryCache.getEntries().get(PositionFactory.create(1, 50));
+ assertThat(cached).isNotNull();
+ assertThat(cached.getMessageMetadata()).isSameAs(suppliedByTheCaller);
+ assertThat(cached.getMessageMetadata().getProducerName()).isEqualTo("from-publish-path");
+ cached.release();
+ }
+
+ @Test
+ public void testCachedEntryMetadataStaysReadableWhenEntriesAreCopied() {
Review Comment:
Thanks for fixing this buffer-lifetime issue and adding the regression test. The current test covers the case where the source entry has no initialized metadata before `insert()`, so the cached entry parses metadata from its own buffer.
`readFromStorage()` has another case: it initializes the source entry’s metadata before calling `insert()`. With `copyEntries=true`, the cached entry owns a different buffer and must not reuse that already-initialized source metadata; otherwise, releasing the source entry leaves the cached metadata referencing the released source buffer.
Initializing the source metadata before `insert()`, then verifying that the cached entry uses a different `MessageMetadata` instance and remains readable after the source entry is released, would cover this case explicitly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][ml] Parse the cached entry's message metadata from the buffer the cache keeps [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on PR #26495:
URL: https://github.com/apache/pulsar/pull/26495#issuecomment-5587862889
@lhotari Not sure if it helps here, though there's also a `materialize()` method in the lightproto objects, that will force the parsing of all fields, so that it doesn't depend anymore on the original buffer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org