[sakshichitnis27]

Fixes #27076

Motivation

With poolMessages=true, the messages of a batch share the same BrokerEntryMetadata instance (ConsumerImpl.newSingleMessage passes the batch's reference to each message). MessageImpl.recycle() called clear() on that instance, so releasing one message also removed the broker publish time and index of the other messages from the same batch that were still in use.

I was able to reproduce this with a regression test: after releasing one of two pooled messages that share broker entry metadata, the other message reported hasBrokerPublishTime() == false.

Modifications

• MessageImpl.recycle() now sets brokerEntryMetadata to null instead of clearing it, so the shared instance is no longer modified. A BrokerEntryMetadata is parsed per entry and isn't pooled, so nothing relies on it being cleared.
• Added MessageImplTest#testReleaseKeepsBrokerMetadataOfPooledBatchSiblings, which releases the two messages of a batch in both orders and checks the broker publish time and index of the message that is still live.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• The new test fails on master without the change and passes with it:
  ./gradlew :pulsar-client-original:test --tests "MessageImplTest"
• I also ran ./gradlew :pulsar-client-original:test --tests "org.apache.pulsar.client.impl.*" and ./gradlew quickCheck locally.

Does this pull request potentially affect one of the following parts:

• 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
• doc-not-needed
• doc-complete

[void-ptr974]

Thanks for the fix!

One small, non-blocking issue in the new test: pooled MessageImpl.create() retains the payload buffer, so message.release() leaves the test’s original buffer reference outstanding. Also, live.release() is skipped if an assertion fails. Keeping each buffer in a local variable, releasing the original reference after construction, and using finally for message cleanup would address both cases.

[codelipenghui]

Thanks for addressing the test cleanup. The previous finding is resolved, and all 25 focused test cases passed at 373277b. LGTM.