diff --git a/pulsar-client/pom.xml b/pulsar-client/pom.xml index 0fb444909c54a..142671a8591a0 100644 --- a/pulsar-client/pom.xml +++ b/pulsar-client/pom.xml @@ -169,6 +169,13 @@ ${skyscreamer.version} test + + + org.awaitility + awaitility + test + + diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java index 2c669ea3285a6..90d037cc10984 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java @@ -47,6 +47,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; @@ -129,8 +130,6 @@ public class ConsumerImpl extends ConsumerBase implements ConnectionHandle private final int receiverQueueRefillThreshold; - private final ReadWriteLock lock = new ReentrantReadWriteLock(); - private final UnAckedMessageTracker unAckedMessageTracker; private final AcknowledgmentsGroupingTracker acknowledgmentsGroupingTracker; private final NegativeAcksTracker negativeAcksTracker; @@ -188,6 +187,7 @@ public class ConsumerImpl extends ConsumerBase implements ConnectionHandle private final boolean poolMessages; private final AtomicReference clientCnxUsedForConsumerRegistration = new AtomicReference<>(); + private final ExecutorService internalPinnedExecutor; static ConsumerImpl newConsumerImpl(PulsarClientImpl client, String topic, @@ -255,6 +255,7 @@ protected ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurat this.expireTimeOfIncompleteChunkedMessageMillis = conf.getExpireTimeOfIncompleteChunkedMessageMillis(); this.autoAckOldestChunkedMessageOnQueueFull = conf.isAutoAckOldestChunkedMessageOnQueueFull(); this.poolMessages = conf.isPoolMessages(); + this.internalPinnedExecutor = client.getInternalExecutorService(); if (client.getConfiguration().getStatsIntervalSeconds() > 0) { stats = new ConsumerStatsRecorderImpl(client, conf, this); @@ -412,25 +413,17 @@ protected Message internalReceive() throws PulsarClientException { protected CompletableFuture> internalReceiveAsync() { CompletableFutureCancellationHandler cancellationHandler = new CompletableFutureCancellationHandler(); CompletableFuture> result = cancellationHandler.createFuture(); - Message message = null; - lock.writeLock().lock(); - try { - message = incomingMessages.poll(0, TimeUnit.MILLISECONDS); + internalPinnedExecutor.execute(() -> { + Message message = incomingMessages.poll(); if (message == null) { pendingReceives.add(result); cancellationHandler.setCancelAction(() -> pendingReceives.remove(result)); } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - result.completeExceptionally(e); - } finally { - lock.writeLock().unlock(); - } - - if (message != null) { - messageProcessed(message); - result.complete(beforeConsume(message)); - } + if (message != null) { + messageProcessed(message); + result.complete(beforeConsume(message)); + } + }); return result; } @@ -475,8 +468,7 @@ protected Messages internalBatchReceive() throws PulsarClientException { protected CompletableFuture> internalBatchReceiveAsync() { CompletableFutureCancellationHandler cancellationHandler = new CompletableFutureCancellationHandler(); CompletableFuture> result = cancellationHandler.createFuture(); - lock.writeLock().lock(); - try { + internalPinnedExecutor.execute(() -> { if (pendingBatchReceives == null) { pendingBatchReceives = Queues.newConcurrentLinkedQueue(); } @@ -498,9 +490,7 @@ protected CompletableFuture> internalBatchReceiveAsync() { pendingBatchReceives.add(opBatchReceive); cancellationHandler.setCancelAction(() -> pendingBatchReceives.remove(opBatchReceive)); } - } finally { - lock.writeLock().unlock(); - } + }); return result; } @@ -959,15 +949,12 @@ private void closeConsumerTasks() { } private void failPendingReceive() { - lock.readLock().lock(); - try { + internalPinnedExecutor.execute(() -> { if (pinnedExecutor != null && !pinnedExecutor.isShutdown()) { failPendingReceives(this.pendingReceives); failPendingBatchReceives(this.pendingBatchReceives); } - } finally { - lock.readLock().unlock(); - } + }); } void activeConsumerChanged(boolean isActive) { @@ -1065,11 +1052,10 @@ uncompressedPayload, createEncryptionContext(msgMetadata), cnx, schema, redelive poolMessages); uncompressedPayload.release(); - lock.readLock().lock(); - try { - // Enqueue the message so that it can be retrieved when application calls receive() - // if the conf.getReceiverQueueSize() is 0 then discard message if no one is waiting for it. - // if asyncReceive is waiting then notify callback without adding to incomingMessages queue + // Enqueue the message so that it can be retrieved when application calls receive() + // if the conf.getReceiverQueueSize() is 0 then discard message if no one is waiting for it. + // if asyncReceive is waiting then notify callback without adding to incomingMessages queue + internalPinnedExecutor.execute(() -> { if (deadLetterPolicy != null && possibleSendToDeadLetterTopicMessages != null && redeliveryCount >= deadLetterPolicy.getMaxRedeliverCount()) { possibleSendToDeadLetterTopicMessages.put((MessageIdImpl) message.getMessageId(), @@ -1080,16 +1066,19 @@ uncompressedPayload, createEncryptionContext(msgMetadata), cnx, schema, redelive } else if (enqueueMessageAndCheckBatchReceive(message) && hasPendingBatchReceive()) { notifyPendingBatchReceivedCallBack(); } - } finally { - lock.readLock().unlock(); - } + }); } else { // handle batch message enqueuing; uncompressed payload has all messages in batch receiveIndividualMessagesFromBatch(msgMetadata, redeliveryCount, ackSet, uncompressedPayload, messageId, cnx); uncompressedPayload.release(); } + internalPinnedExecutor.execute(() + -> tryTriggerListener()); + } + + private void tryTriggerListener() { if (listener != null) { triggerListener(); } @@ -1292,17 +1281,14 @@ void receiveIndividualMessagesFromBatch(MessageMetadata msgMetadata, int redeliv if (possibleToDeadLetter != null) { possibleToDeadLetter.add(message); } - lock.readLock().lock(); - try { + internalPinnedExecutor.execute(() -> { if (peekPendingReceive() != null) { notifyPendingReceivedCallback(message, null); } else if (enqueueMessageAndCheckBatchReceive(message) && hasPendingBatchReceive()) { notifyPendingBatchReceivedCallBack(); } - } finally { - lock.readLock().unlock(); - } - singleMessagePayload.release(); + singleMessagePayload.release(); + }); } if (ackBitSet != null) { ackBitSet.recycle(); diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java index 8bb7bbc41a1c7..0adb165f066b1 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java @@ -21,7 +21,6 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.EventLoop; import io.netty.util.Timer; -import org.apache.bookkeeper.common.util.OrderedScheduler; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.apache.pulsar.client.util.ExecutorProvider; import org.mockito.Mockito; diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java index f3ccfd86e80b9..2702439cee345 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java @@ -25,12 +25,14 @@ import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import io.netty.util.concurrent.DefaultThreadFactory; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.Messages; @@ -38,6 +40,7 @@ import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; import org.apache.pulsar.client.util.ExecutorProvider; +import org.awaitility.Awaitility; import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; @@ -49,12 +52,15 @@ public class ConsumerImplTest { private ExecutorProvider executorProvider; private ConsumerImpl consumer; private ConsumerConfigurationData consumerConf; + private ExecutorService executorService; @BeforeMethod(alwaysRun = true) public void setUp() { executorProvider = new ExecutorProvider(1, "ConsumerImplTest"); consumerConf = new ConsumerConfigurationData<>(); PulsarClientImpl client = ClientTestFixtures.createPulsarClientMock(); + executorService = Executors.newSingleThreadExecutor(); + when(client.getInternalExecutorService()).thenReturn(executorService); ClientConfigurationData clientConf = client.getConfiguration(); clientConf.setOperationTimeoutMs(100); clientConf.setStatsIntervalSeconds(0); @@ -74,6 +80,10 @@ public void cleanup() { executorProvider.shutdownNow(); executorProvider = null; } + if (executorService != null) { + executorService.shutdownNow(); + executorService = null; + } } @Test(invocationTimeOut = 1000) @@ -162,7 +172,7 @@ public void testNotifyPendingReceivedCallback_WorkNormally() { public void testReceiveAsyncCanBeCancelled() { // given CompletableFuture> future = consumer.receiveAsync(); - Assert.assertEquals(consumer.peekPendingReceive(), future); + Awaitility.await().untilAsserted(() -> Assert.assertEquals(consumer.peekPendingReceive(), future)); // when future.cancel(true); // then @@ -173,7 +183,7 @@ public void testReceiveAsyncCanBeCancelled() { public void testBatchReceiveAsyncCanBeCancelled() { // given CompletableFuture> future = consumer.batchReceiveAsync(); - Assert.assertTrue(consumer.hasPendingBatchReceive()); + Awaitility.await().untilAsserted(() -> Assert.assertTrue(consumer.hasPendingBatchReceive())); // when future.cancel(true); // then diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java index 6e587f720e6b4..d0c4023e09e75 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ReaderImplTest.java @@ -22,32 +22,50 @@ import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.conf.ReaderConfigurationData; +import org.awaitility.Awaitility; +import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import static org.mockito.Mockito.when; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertNull; public class ReaderImplTest { ReaderImpl reader; + private ExecutorService executorService; @BeforeMethod void setupReader() { PulsarClientImpl mockedClient = ClientTestFixtures.createPulsarClientMockWithMockedClientCnx(); ReaderConfigurationData readerConfiguration = new ReaderConfigurationData<>(); readerConfiguration.setTopicName("topicName"); + executorService = Executors.newSingleThreadExecutor(); + when(mockedClient.getInternalExecutorService()).thenReturn(executorService); CompletableFuture> consumerFuture = new CompletableFuture<>(); reader = new ReaderImpl<>(mockedClient, readerConfiguration, ClientTestFixtures.createMockedExecutorProvider(), consumerFuture, Schema.BYTES); } + @AfterMethod + public void clean() { + if (executorService != null) { + executorService.shutdownNow(); + executorService = null; + } + } + @Test void shouldSupportCancellingReadNextAsync() { // given CompletableFuture> future = reader.readNextAsync(); - assertNotNull(reader.getConsumer().peekPendingReceive()); + Awaitility.await().untilAsserted(() -> { + assertNotNull(reader.getConsumer().peekPendingReceive()); + }); // when future.cancel(false);