[fix][client] Fix client side memory leak when call MessageImpl.create and fix imprecise client-side metrics: pendingMessagesUpDownCounter, pendingBytesUpDownCounter, latencyHistogram

[poorbarcode]

Motivation

Issue 1: logs is below^[1]

• create a Message manually.
• call ProducerBase.sendAsync(Message)
• release the message that was created manually after the send is completed.
  • (Highlight) the variable Message.payload is null after calling Message.release.
  • Pulsar client will get an NPE at this line https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L421, and leads a client-side memory leak, eventually get an OOM error.

Issue 2:

• create a Message manually.
• call ProducerBase.sendAsync(msg1)
• call ProducerBase.sendAsync(msg2)
• release the message that was created manually after the send is completed.
  • (Highlight) msg2.payload was retained again when calling interceptor.onSendAcknowledgement, leading to this payload not being released. This retention is unnecessary. See https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L430

[1]:

2024-04-02T14:37:27,801 - WARN  - [pulsar-client-io-35-4:ProducerImpl] - [persistent://my-property/my-ns/tp-7c4d35f4-0c0b-4415-b565-e88722b4879a] [test-0-0] Got exception while completing the callback for msg 1:
java.lang.NullPointerException: Cannot invoke "io.netty.buffer.ByteBuf.release()" because the return value of "org.apache.pulsar.client.impl.MessageImpl.getDataBuffer()" is null
	at org.apache.pulsar.client.impl.ProducerImpl$1.sendComplete(ProducerImpl.java:421) ~[classes/:?]
	at org.apache.pulsar.client.impl.ProducerImpl$OpSendMsg.sendComplete(ProducerImpl.java:1594) ~[classes/:?]
	at org.apache.pulsar.client.impl.ProducerImpl.ackReceived(ProducerImpl.java:1277) ~[classes/:?]
	at org.apache.pulsar.client.impl.ClientCnx.handleSendReceipt(ClientCnx.java:485) ~[classes/:?]
	at org.apache.pulsar.common.protocol.PulsarDecoder.channelRead(PulsarDecoder.java:236) ~[classes/:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346) ~[netty-codec-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:318) ~[netty-codec-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.handler.flush.FlushConsolidationHandler.channelRead(FlushConsolidationHandler.java:152) ~[netty-handler-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562) ~[netty-transport-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) ~[netty-common-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.108.Final.jar:4.1.108.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.108.Final.jar:4.1.108.Final]
	at java.lang.Thread.run(Thread.java:833) ~[?:?]

The scope that the issue affects

Regarding issue-2 that was described in Motivation, it will happen if batch sending is enabled(without calling the API ProducerBase.sendAsync(Message)). But it will not cause an OOM because all the message payload created by the public API will be built this way:

• ByteBuffer jdkBuffer = ByteBuffer.wrap( byte[] )
• ByteBuf nettyByteBuf = Unpooled.wrappedBuffer(jdkBuffer)

So there is no memory leak because the UnpooledByteBuf will not cause issues even if it is eventually not being released.

Note: If someone builds messages by public static MessageImpl<T> create(ByteBuf payload) and enables batch-send, they will encounter issue 2, but we did not provide an API to send a message typed org.apache.pulsar.client.api.Message, so no worry about this. (Highlight)It only affects the users who use ProducerImpl or ProducerBase

Modifications

• Fix the memory leak
  • which is caused by issue 1.
  • which is caused by issue 1.
• Fix the inaccurate metrics:
  • pendingMessagesUpDownCounter
  • pendingBytesUpDownCounter
  • latencyHistogram

Documentation

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

Matching PR in forked repository

PR in forked repository: x

[merlimat]

Nice catch!

Is there any condition in which this was getting triggered from the public APIs?

[dao-jun]

lgtm

[dao-jun]

Just a minor comment:

Will

if (ReferenceCountUtil.refCnt(payload) > 0) {
   ReferenceCountUtil.release(payload);
} 

be better?

double release a ReferentCount Object will throw an exception, and ReferenceCountUtil.safeRelease(...) prints WARN log.

[lhotari]

Is there any condition in which this was getting triggered from the public APIs?

@poorbarcode Please answer to this question. I'm wondering about the valid usages of this API. Is it used internally in Pulsar since this is an internal API? How did you run into the problem?

[poorbarcode]

@merlimat

Is there any condition in which this was triggered by the public APIs?

Regarding issue 2 that was described in Motivation, it will happen if batch sending is enabled(without calling the API ProducerBase.sendAsync(Message)). But it will not cause an OOM because all the message payload created by the public API will be built this way:

• ByteBuffer jdkBuffer = ByteBuffer.wrap( byte[] )
• ByteBuf nettyByteBuf = Unpooled.wrappedBuffer(jdkBuffer)

So there is no memory leak because the UnpooledByteBuf will not cause issues even if it is eventually not being released.

Note: If someone builds messages by public static MessageImpl<T> create(ByteBuf payload) and enables batch-send, they will encounter issue 2, but we did not provide an API to send a message typed org.apache.pulsar.client.api.Message, so no worry about this. It only affects the users who use ProducerImpl or ProducerBase

[poorbarcode]

@lhotari

Is there any condition in which this was getting triggered from the public APIs?

@poorbarcode Please answer to this question. I'm wondering about the valid usages of this API. Is it used internally in Pulsar since this is an internal API? How did you run into the problem?

A custom plugin for our company uses this API.

[poorbarcode]

@dao-jun

double release a ReferentCount Object will throw an exception, and ReferenceCountUtil.safeRelease(...) prints WARN log.

I think it is ok. If it will be released twice, we need this error log to let us know something is not as expected.

[lhotari]

LGTM

[poorbarcode]

@dao-jun

@poorbarcode I just noticed there are some PRs are fixing WARN/ERROR logs recently, maybe it's better to consider it carefully.

ReferenceCountUtil.safeRelease(payload) will do nothing if the payload is null, I added a error log manually.

[lhotari]

LGTM. The current title of the PR isn't optimal since this does fix issues also in other cases and also fixes some glitches with the stats.

[poorbarcode]

LGTM. The current title of the PR isn't optimal since this does fix issues also in other cases and also fixes some glitches with the stats.

Renamed the title.

[nodece]

This PR broke the branch-2.10 CI. Fixed by f89fa72

Error:  Medium: Unread field: org.apache.pulsar.client.impl.ProducerImpl$DefaultSendMessageCallback.msgSize [org.apache.pulsar.client.impl.ProducerImpl$DefaultSendMessageCallback] At ProducerImpl.java:[line 342] URF_UNREAD_FIELD
Error:  Failed to execute goal com.github.spotbugs:spotbugs-maven-plugin:4.2.2:check (spotbugs) on project pulsar-client-original: failed with 1 bugs and 0 errors -> [Help 1]