[fix][client]Fix Producer exceeds PulsarClient memory limit when sending fail timeout using CompressionType.NONE

[AnonHxy]

Fixes #17662

Motivation

• Fixes [Bug] Producer exceeds PulsarClient memory limit when sending fail timeout using CompressionType.NONE #17662

• The root cause is that:

  1. After we ever sent a big message, e.g, 2M bytes, the AbstractBatchMessageContainer#maxBatchSize will never be less than 2M bytes(see line171) in the following batches, even if the message size we added is 1 bytes only.
     

     pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java

     Lines 167 to 173 in a98f025

     	}
     	// Update the current max batch size using the uncompressed size, which is what we need in any case to
     	// accumulate the batch content
     	maxBatchSize = Math.max(maxBatchSize, uncompressedSize);
     	return compressedPayload;
     	}

  2. So when we add the first message in the following batches, we will always allocate a byte buffer greater or equals to 2M, see line102
     

     pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java

     Lines 98 to 103 in a98f025

     	messageMetadata.setSequenceId(msg.getSequenceId());
     	lowestSequenceId = Commands.initBatchMessageMetadata(messageMetadata, msg.getMessageBuilder());
     	this.firstCallback = callback;
     	batchedMessageMetadataAndPayload = allocator.buffer(
     	Math.min(maxBatchSize, ClientCnx.getMaxMessageSize()));
     	if (msg.getMessageBuilder().hasTxnidMostBits() && currentTxnidMostBits == -1) {

  3. When we use CompressionCodecNone to encode the 2M bytes buffer it just retain the buffer, which means the 2M bytes buffer will be added to ProducerImpl#pendingMessages and will be release until timeout(30s by default). Note that we sent 1 bytes message(or batched, but it should much less that 2M), but retain 2M bytes buffer, so it will exceed PulsarClient memory limit.

Modifications

Copy readable bytes to a new Bytebuf if necessary, like CompressionCodecLZ4 or other CompressionCodec implement

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

• doc-required
  (Your PR needs to update docs and you will update later)

• doc-not-needed
  (Please explain why)

• doc
  (Your PR contains doc changes)

• doc-complete
  (Docs have been already added)

[HQebupt]

Do you mean: The MemoryLimitController manages memory according to the real size of message, but the message is allocated a bigger memory by BatchMessageContainerImpl?

[AnonHxy]

Right.

[HQebupt]

LGTM

[lhotari]

Good work.

I just wonder if the title of the PR should mention that this applies to CompressionType.NONE .

[AnonHxy]

Good work.

I just wonder if the title of the PR should mention that this applies to CompressionType.NONE .

Sure

[merlimat]

I don't think this is the right solution because it would cause a memory copy for all the batches.

[Jason918]

+1.

1. We should take the whole buffer size into account in MemoryLimitController. Because it's allocated by our producer.
2. There is a memory waste problem in producer batch container, as the max buffer size only grows.

[Jason918]

+1.

1. We should take the whole buffer size into account in MemoryLimitController. Because it's allocated by our producer.
2. There is a memory waste problem in producer batch container, as the max buffer size only grows.

[Jason918]

@AnonHxy Please take a look at this PR #15033. It tries to solve the memory waste problem.

[AnonHxy]

closed this pr becase it' not the best solution