Revert "[fix][client]Fix client memory limit currentUsage leak and se…

…maphore release duplicated in ProducerImpl (#16837)" Even after importing the mockito dependency, the testProducerBatchSendTimeoutMemoryRelease will still fail. This reverts commit 9610640.
apache · Aug 5, 2022 · 38b93b2 · 38b93b2
1 parent 9610640
commit 38b93b2
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 66 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLimitTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLimitTest.java
@@ -69,35 +69,6 @@ public void testProducerTimeoutMemoryRelease() throws Exception {
 
     }
 
-    @Test(timeOut = 10_000)
-    public void testProducerBatchSendTimeoutMemoryRelease() throws Exception {
-        initClientWithMemoryLimit();
-        @Cleanup
-        ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
-                .topic("testProducerMemoryLimit")
-                .sendTimeout(5, TimeUnit.SECONDS)
-                .maxPendingMessages(0)
-                .enableBatching(true)
-                .batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS)
-                .batchingMaxBytes(12)
-                .create();
-        this.stopBroker();
-        try {
-            producer.newMessage().value("memory-test".getBytes(StandardCharsets.UTF_8)).sendAsync();
-            try {
-                producer.newMessage().value("memory-test".getBytes(StandardCharsets.UTF_8)).sendAsync().get();
-            } catch (Exception e) {
-                throw PulsarClientException.unwrap(e);
-            }
-
-            throw new IllegalStateException("can not reach here");
-        } catch (PulsarClientException.TimeoutException ex) {
-            PulsarClientImpl clientImpl = (PulsarClientImpl) this.pulsarClient;
-            final MemoryLimitController memoryLimitController = clientImpl.getMemoryLimitController();
-            Assert.assertEquals(memoryLimitController.currentUsage(), 0);
-        }
-    }
-
     @Test(timeOut = 10_000)
     public void testProducerCloseMemoryRelease() throws Exception {
         initClientWithMemoryLimit();

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
@@ -18,10 +18,6 @@
  */
 package org.apache.pulsar.client.impl;
 
-import static org.mockito.ArgumentMatchers.any;
-import java.lang.reflect.Field;
-import java.nio.charset.StandardCharsets;
-import java.util.concurrent.TimeUnit;
 import lombok.Cleanup;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.ProducerConsumerBase;
@@ -210,35 +206,4 @@ public void testEnsureNotBlockOnThePendingQueue() throws Exception {
         Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
         Assert.assertFalse(producer.isErrorStat());
     }
-
-    @Test(timeOut = 10_000)
-    public void testBatchMessageSendTimeoutProducerSemaphoreRelease() throws Exception {
-        final int pendingQueueSize = 10;
-        @Cleanup
-        ProducerImpl<byte[]> producer =
-                (ProducerImpl<byte[]>) pulsarClient.newProducer()
-                        .topic("testProducerSemaphoreRelease")
-                        .sendTimeout(5, TimeUnit.SECONDS)
-                        .maxPendingMessages(pendingQueueSize)
-                        .enableBatching(true)
-                        .batchingMaxPublishDelay(500, TimeUnit.MILLISECONDS)
-                        .batchingMaxBytes(12)
-                        .create();
-        this.stopBroker();
-        try {
-            ProducerImpl<byte[]> spyProducer = Mockito.spy(producer);
-            Mockito.doThrow(new PulsarClientException.CryptoException("crypto error")).when(spyProducer)
-                    .encryptMessage(any(),any());
-
-            Field batchMessageContainerField = ProducerImpl.class.getDeclaredField("batchMessageContainer");
-            batchMessageContainerField.setAccessible(true);
-            BatchMessageContainerImpl batchMessageContainer = (BatchMessageContainerImpl) batchMessageContainerField.get(spyProducer);
-            batchMessageContainer.setProducer(spyProducer);
-            spyProducer.send("semaphore-test".getBytes(StandardCharsets.UTF_8));
-
-            throw new IllegalStateException("can not reach here");
-        } catch (PulsarClientException.TimeoutException ex) {
-            Assert.assertEquals(producer.getSemaphore().get().availablePermits(), 10);
-        }
-    }
 }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -1787,10 +1787,8 @@ private void failPendingBatchMessages(PulsarClientException ex) {
             return;
         }
         final int numMessagesInBatch = batchMessageContainer.getNumMessagesInBatch();
-        final long currentBatchSize = batchMessageContainer.getCurrentBatchSize();
         batchMessageContainer.discard(ex);
         semaphoreRelease(numMessagesInBatch);
-        client.getMemoryLimitController().releaseMemory(currentBatchSize);
     }
 
     @Override
@@ -1832,7 +1830,10 @@ private void batchMessageAndSend() {
                 for (OpSendMsg opSendMsg : opSendMsgs) {
                     processOpSendMsg(opSendMsg);
                 }
+            } catch (PulsarClientException e) {
+                semaphoreRelease(batchMessageContainer.getNumMessagesInBatch());
             } catch (Throwable t) {
+                semaphoreRelease(batchMessageContainer.getNumMessagesInBatch());
                 log.warn("[{}] [{}] error while create opSendMsg by batch message container", topic, producerName, t);
             }
         }