[fix][client] Fix producer thread block forever on memory limit controller

[wenbingshen]

Motivation

When the modification of the current PR is rolled back, the following unit test will fail, and the main producer thread will be stuck forever in org.apache.pulsar.client.impl.MemoryLimitController#reserveMemory condition.await();

 @Test(timeOut = 10_000)
    public void testProducerBlockReserveMemory() throws Exception {
        replacePulsarClient(PulsarClient.builder().
                serviceUrl(lookupUrl.toString())
                .memoryLimit(1, SizeUnit.KILO_BYTES));
        @Cleanup
        ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
                .topic("testProducerMemoryLimit")
                .sendTimeout(5, TimeUnit.SECONDS)
                .compressionType(CompressionType.SNAPPY)
                .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
                .maxPendingMessages(0)
                .blockIfQueueFull(true)
                .enableBatching(true)
                .batchingMaxMessages(100)
                .batchingMaxBytes(65536)
                .batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS)
                .create();
        int msgCount = 5;
        CountDownLatch cdl = new CountDownLatch(msgCount);
        for (int i = 0; i < msgCount; i++) {
            producer.sendAsync("memory-test".getBytes(StandardCharsets.UTF_8)).whenComplete(((messageId, throwable) -> {
                cdl.countDown();
            }));
        }

        cdl.await();

        producer.close();
        PulsarClientImpl clientImpl = (PulsarClientImpl) this.pulsarClient;
        final MemoryLimitController memoryLimitController = clientImpl.getMemoryLimitController();
        Assert.assertEquals(memoryLimitController.currentUsage(), 0);
    }

According to my investigation this is related to this PR #17936

1. When the producer turns on the following parameters and uses asynchronous sending:
   .compressionType(CompressionType.SNAPPY)
   .blockIfQueueFull(true)
   .enableBatching(true)
   Set .memoryLimit(1, SizeUnit.KILO_BYTES)); small enough to make it easier to reproduce the problem.

2. BatchMessageContainer will apply for a batchAllocatedSize from MemoryLimitController when building BatchMsgMetadata for the first time. At this time, MemoryLimitController.currentUsage=(msg payload size + batchedMessageMetadataAndPayload size),

3. The main production thread continues to send data asynchronously, but at this time the MemoryLimitController has reached the memoryLimit limit, and the main thread will be stuck in condition.await(); waiting to wake up;

4. When the batch message compression is completed, the BatchMessageContainer will call updateAndReserveBatchAllocatedSize again. At this time, the memory will be released to the MemoryLimitController to the actual batch message size, but it will not wake up those threads that are blocked and waiting due to insufficient memory.

5. After the batch message is sent, call org.apache.pulsar.client.impl.MemoryLimitController#releaseMemory to release the batch size memory. However, since the size released by MemoryLimitController.currentUsage + is smaller than memoryLimit, no threads are awakened.

6. The main production thread is always stuck in condition.await();

Documentation

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

[wenbingshen]

These are some logs I printed during debugging. I can clearly see that forceReserveMemory applied for 1024 bytes once and applied for -969 bytes the second time.

[lhotari]

Great catch @wenbingshen ! this has been a long outstanding issue.

[AnonHxy]

Nice catch! Thanks

[wenbingshen]

/pulsarbot rerun-failure-checks

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (8beac8b) 73.42% compared to head (1359217) 73.58%.
Report is 5 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #21790      +/-   ##
============================================
+ Coverage     73.42%   73.58%   +0.15%     
+ Complexity    32795    32247     -548     
============================================
  Files          1897     1858      -39     
  Lines        140656   138021    -2635     
  Branches      15491    15111     -380     
============================================
- Hits         103282   101561    -1721     
+ Misses        29297    28607     -690     
+ Partials       8077     7853     -224     

Flag	Coverage Δ
inttests	24.18% <25.00%> (-0.07%)	⬇️
systests	23.74% <50.00%> (-1.11%)	⬇️
unittests	72.86% <100.00%> (+0.15%)	⬆️

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

Files	Coverage Δ
.../pulsar/client/impl/BatchMessageContainerImpl.java	80.68% <100.00%> (+0.22%)	⬆️

... and 127 files with indirect coverage changes

[poorbarcode]

Maybe modify the method tryReserveMemory and forceReserveMemory like below is better?

public void forceReserveMemory(long size) {
    if (size < 0) {
        releaseMemory(-size);
        return;
    }
    // do original implementation
    ....
}

public void tryReserveMemory(long size) {
    if (size < 0) {
        releaseMemory(-size);
        return;
    }
    // do original implementation
    ....
}

[lhotari]

Maybe modify the method tryReserveMemory and forceReserveMemory like below is better?

I think it's not a good idea. It's better to add validation and throw IllegalArgumentException if the value is negative

If there would have been validation, the API would have never been misused and the issue wouldn't have happened in the past.

[poorbarcode]

I think it's not a good idea. It's better to add validation and throw IllegalArgumentException if the value is negative

Sure

[Technoboy-]

Maybe modify the method tryReserveMemory and forceReserveMemory like below is better?

I think it's not a good idea. It's better to add validation and throw IllegalArgumentException if the value is negative

If there would have been validation, the API would have never been misused and the issue wouldn't have happened in the past.

yes, agree

[wenbingshen]

Maybe modify the method tryReserveMemory and forceReserveMemory like below is better?

I think it's not a good idea. It's better to add validation and throw IllegalArgumentException if the value is negative

If there would have been validation, the API would have never been misused and the issue wouldn't have happened in the past.

@lhotari Good idea! I think we can add validation that the value is negative and throw an IllegalArgumentException by submitting another PR.

[poorbarcode]

@lhotari Good idea! I think we can add validation that the value is negative and throw an IllegalArgumentException by submitting another PR.

At least, we should add a WARN log when calling tryReserveMemory/forceReserveMemory with a negative value

[wenbingshen]

@lhotari Good idea! I think we can add validation that the value is negative and throw an IllegalArgumentException by submitting another PR.

At least, we should add a WARN log when calling tryReserveMemory/forceReserveMemory with a negative value

@poorbarcode I will add an assert after later.

assert in separate follow up PR

[poorbarcode]

@lhotari @wenbingshen @Technoboy-

assert in separate follow up PR

I never see the following PR. To prevent new PRs from incorrectly using tryReserveMemory, reserveMemory, forceReserveMemory leads to the same issue, I create a new PR to add this validation. Please take a look

• [improve] [client] Prevent reserve memory with a negative memory size to avoid send task stuck #21804