[fix][broker]Fix memory leak when using a customized ManagedLedger implementation

[poorbarcode]

Motivation

Issue: if a user used a customized ManagedLedger implementation, which does not extend ManagedLedgerImpl, he will encounter a memory leak issue, you can reproduce it by the new test testNoMemoryLeakWhenExpireMessages

#24606 improved the function message expiration, but forgot to release EntryImpl here: https://github.com/apache/pulsar/pull/24606/files#diff-5edf14cc6f25857d0cfdd26b2d3b3141230ecfb0dfa95aebf7583fd76ede4c4bR3918-R3929

PersistentTopic.java

    public CompletableFuture<Boolean> isOldestMessageExpiredAsync(ManagedCursor cursor, int messageTTLInSeconds) {
        CompletableFuture<Boolean> res = new CompletableFuture<>();
        cursor.asyncGetNthEntry(1, IndividualDeletedEntries.Include, new AsyncCallbacks.ReadEntryCallback() {
            @Override
            public void readEntryComplete(Entry entry, Object ctx) {
                long entryTimestamp = 0;
                try {
                    res.complete(MessageImpl.isEntryExpired(
                            (int) (messageTTLInSeconds * MESSAGE_EXPIRY_THRESHOLD), entryTimestamp));
                } catch (IOException e) {
                    log.warn("[{}] [{}] Error while getting the oldest message", topic, cursor.toString(), e);
                    res.complete(false);
                }
                // (Highlight) the entry that was read out is not released here, which causes a memory leak issue
            }
        }, null);
        return res;
    }

2025-11-10T01:29:54,412+0000 [compact-task-publisher-OrderedScheduler-2-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImplWithTS - [public/default/persistent/test-partition-1] Published compact task for streamId: 68238366, ledger 938515 from entry 113 to 115
2025-11-10T01:29:54,428+0000 [pulsar-io-9-6] ERROR io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
#1:
	io.netty.buffer.AdvancedLeakAwareByteBuf.readByte(AdvancedLeakAwareByteBuf.java:401)
	org.apache.pulsar.common.api.proto.LightProtoCodec.readVarInt(LightProtoCodec.java:140)
	org.apache.pulsar.common.api.proto.MessageMetadata.parseFrom(MessageMetadata.java:1359)
	org.apache.pulsar.common.protocol.Commands.parseMessageMetadata(Commands.java:479)
	org.apache.pulsar.common.protocol.Commands.parseMessageMetadata(Commands.java:466)
	org.apache.pulsar.common.protocol.Commands.getEntryTimestamp(Commands.java:499)
	org.apache.pulsar.broker.service.persistent.PersistentTopic$13.readEntryComplete(PersistentTopic.java:3921)
	org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl$1.readEntriesComplete(RangeEntryCacheImpl.java:239)
	org.apache.bookkeeper.mledger.impl.cache.PendingReadsManager$PendingRead.readEntriesComplete(PendingReadsManager.java:278)
	org.apache.bookkeeper.mledger.impl.cache.PendingReadsManager$PendingRead.lambda$attach$0(PendingReadsManager.java:244)
	org.apache.bookkeeper.common.util.SingleThreadExecutor.safeRunTask(SingleThreadExecutor.java:128)
	org.apache.bookkeeper.common.util.SingleThreadExecutor.run(SingleThreadExecutor.java:105)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.base/java.lang.Thread.run(Unknown Source)

Modifications

• fix the bug

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: x

[lhotari]

LGTM

[lhotari]

Issue: if a user used a customized ManagedLedger implementation, which does not extend ManagedLedgerImpl

@poorbarcode Why does this only apply to a custom implementation?

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 74.37%. Comparing base (676ba07) to head (52304ed).
⚠️ Report is 50 commits behind head on master.

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #25016       +/-   ##
=============================================
+ Coverage     38.56%   74.37%   +35.81%     
- Complexity    13262    33733    +20471     
=============================================
  Files          1856     1920       +64     
  Lines        145287   150304     +5017     
  Branches      16877    17458      +581     
=============================================
+ Hits          56025   111790    +55765     
+ Misses        81696    29611    -52085     
- Partials       7566     8903     +1337     

Flag	Coverage Δ
inttests	26.33% <100.00%> (+0.15%)	⬆️
systests	22.86% <50.00%> (+0.09%)	⬆️
unittests	73.92% <100.00%> (+39.18%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...che/bookkeeper/mledger/impl/ManagedCursorImpl.java	78.62% <100.00%> (+28.93%)	⬆️
...ache/pulsar/broker/ManagedLedgerClientFactory.java	78.09% <ø> (+12.38%)	⬆️
...sar/broker/service/persistent/PersistentTopic.java	79.88% <100.00%> (+28.39%)	⬆️

... and 1422 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Technoboy-]

Better to paste the stack for the motivation.