[fix][client] Cannot access message data inside ProducerInterceptor#onSendAcknowledgement

[Shawyeok]

Motivation

In #22393, the payload of message get released before calling ProducerInterceptor#onSendAcknowledgement, hence access message data inside ProducerInterceptor#onSendAcknowledgement will get exception below:

io.netty.util.IllegalReferenceCountException: refCnt: 0
	at io.netty.buffer.AbstractByteBuf.ensureAccessible(AbstractByteBuf.java:1454) ~[netty-buffer-4.1.115.Final.jar:4.1.115.Final]
	at io.netty.buffer.UnpooledHeapByteBuf.array(UnpooledHeapByteBuf.java:147) ~[netty-buffer-4.1.115.Final.jar:4.1.115.Final]
	at org.apache.pulsar.client.impl.MessageImpl.getData(MessageImpl.java:371) ~[classes/:?]
	at org.apache.pulsar.client.api.InterceptorsTest$6.onSendAcknowledgement(InterceptorsTest.java:239) ~[test-classes/:?]
	at org.apache.pulsar.client.api.interceptor.ProducerInterceptorWrapper.onSendAcknowledgement(ProducerInterceptorWrapper.java:58) ~[classes/:?]
	at org.apache.pulsar.client.impl.ProducerInterceptors.onSendAcknowledgement(ProducerInterceptors.java:100) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerBase.onSendAcknowledgement(ProducerBase.java:162) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerImpl$DefaultSendMessageCallback.onSendComplete(ProducerImpl.java:451) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerImpl$DefaultSendMessageCallback.sendComplete(ProducerImpl.java:423) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerImpl$OpSendMsg.sendComplete(ProducerImpl.java:1643) [classes/:?]
	at org.apache.pulsar.client.impl.ProducerImpl.ackReceived(ProducerImpl.java:1316) [classes/:?]
	at org.apache.pulsar.client.impl.ClientCnx.handleSendReceipt(ClientCnx.java:490) [classes/:?]
	at org.apache.pulsar.common.protocol.PulsarDecoder.channelRead(PulsarDecoder.java:236) [classes/:?]

Modifications

• Release the message payload after ProducerInterceptor#onSendAcknowledgement is invoked to ensure the payload is accessible during the callback
• Add test case InterceptorsTest#testProducerInterceptorAccessMessageData

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• InterceptorsTest#testProducerInterceptorAccessMessageData

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

Matching PR in forked repository

PR in forked repository: Shawyeok#22

[poorbarcode]

LGTM

[thetumbled]

LGTM.

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 74.09%. Comparing base (bbc6224) to head (ea703c8).
Report is 819 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23791      +/-   ##
============================================
+ Coverage     73.57%   74.09%   +0.51%     
+ Complexity    32624    31759     -865     
============================================
  Files          1877     1853      -24     
  Lines        139502   143374    +3872     
  Branches      15299    16276     +977     
============================================
+ Hits         102638   106229    +3591     
+ Misses        28908    28774     -134     
- Partials       7956     8371     +415     

Flag	Coverage Δ
inttests	26.74% <54.54%> (+2.16%)	⬆️
systests	23.12% <100.00%> (-1.20%)	⬇️
unittests	73.61% <100.00%> (+0.77%)	⬆️

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

Files with missing lines	Coverage Δ
...va/org/apache/pulsar/client/impl/ProducerImpl.java	84.23% <100.00%> (+0.64%)	⬆️

... and 1019 files with indirect coverage changes