Skip to content

Commit

Permalink
Forget to update memory usage when invalid message (#16835)
Browse files Browse the repository at this point in the history
### Modifications
release memory usage when invalid message.
Only need to release memory usage here, no need to release semaphore. Both add testcases.

coauthored by @pengxiangrui127.

### Verifying this change
- add unit tests for this change

### Documentation

Check the box below or label this PR directly.

Need to update docs?

- [x] `doc-not-needed`
bug fix, no need doc

(cherry picked from commit 57b008a)
  • Loading branch information
shoothzj authored and BewareMyPower committed Aug 5, 2022
1 parent 7107657 commit a501593
Show file tree
Hide file tree
Showing 3 changed files with 55 additions and 0 deletions.
Expand Up @@ -23,6 +23,8 @@
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SizeUnit;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
Expand All @@ -47,6 +49,31 @@ protected void cleanup() throws Exception {
super.internalCleanup();
}

@Test(timeOut = 10_000)
public void testProducerInvalidMessageMemoryRelease() throws Exception {
initClientWithMemoryLimit();
@Cleanup
ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
.topic("testProducerMemoryLimit")
.sendTimeout(5, TimeUnit.SECONDS)
.batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS)
.batchingMaxBytes(10240)
.enableBatching(true)
.create();
this.stopBroker();
try {
try (MockedStatic<ClientCnx> mockedStatic = Mockito.mockStatic(ClientCnx.class)) {
mockedStatic.when(ClientCnx::getMaxMessageSize).thenReturn(8);
producer.send("memory-test".getBytes(StandardCharsets.UTF_8));
}
throw new IllegalStateException("can not reach here");
} catch (PulsarClientException.InvalidMessageException ex) {
PulsarClientImpl clientImpl = (PulsarClientImpl) this.pulsarClient;
final MemoryLimitController memoryLimitController = clientImpl.getMemoryLimitController();
Assert.assertEquals(memoryLimitController.currentUsage(), 0);
}
}

@Test(timeOut = 10_000)
public void testProducerTimeoutMemoryRelease() throws Exception {
initClientWithMemoryLimit();
Expand Down
Expand Up @@ -25,12 +25,15 @@
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.util.FutureUtil;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
Expand All @@ -55,6 +58,29 @@ public void cleanup() throws Exception {
super.internalCleanup();
}

@Test(timeOut = 10_000)
public void testProducerSemaphoreInvalidMessage() throws Exception {
final int pendingQueueSize = 100;

@Cleanup
ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
.topic("testProducerSemaphoreAcquire")
.maxPendingMessages(pendingQueueSize)
.enableBatching(false)
.create();

this.stopBroker();
try {
try (MockedStatic<ClientCnx> mockedStatic = Mockito.mockStatic(ClientCnx.class)) {
mockedStatic.when(ClientCnx::getMaxMessageSize).thenReturn(2);
producer.send("semaphore-test".getBytes(StandardCharsets.UTF_8));
}
throw new IllegalStateException("can not reach here");
} catch (PulsarClientException.InvalidMessageException ex) {
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
}
}

@Test(timeOut = 30000)
public void testProducerSemaphoreAcquireAndRelease() throws PulsarClientException, ExecutionException, InterruptedException {

Expand Down
Expand Up @@ -200,6 +200,8 @@ public boolean isMultiBatches() {
public OpSendMsg createOpSendMsg() throws IOException {
ByteBuf encryptedPayload = producer.encryptMessage(messageMetadata, getCompressedBatchMetadataAndPayload());
if (encryptedPayload.readableBytes() > ClientCnx.getMaxMessageSize()) {
messages.forEach(msg -> producer.client.getMemoryLimitController()
.releaseMemory(msg.getUncompressedSize()));
discard(new PulsarClientException.InvalidMessageException(
"Message size is bigger than " + ClientCnx.getMaxMessageSize() + " bytes"));
return null;
Expand Down

0 comments on commit a501593

Please sign in to comment.