From a97446e95cc8faea480c332818ff169a3f2695f3 Mon Sep 17 00:00:00 2001 From: ruihongzhou Date: Thu, 14 Sep 2023 11:07:01 +0800 Subject: [PATCH 1/2] [fix][txn] fix the consumer stuck due to deduplicated messages in pending ack state --- .../service/AbstractBaseDispatcher.java | 11 ++-- .../client/impl/TransactionEndToEndTest.java | 56 +++++++++++++++++++ 2 files changed, 62 insertions(+), 5 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java index 50f05f80e3f8b..eb8b015139586 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java @@ -213,12 +213,7 @@ public int filterEntriesForConsumer(@Nullable MessageMetadata[] metadataArray, i this.filterAcceptedMsgs.add(entryMsgCnt); } - totalEntries++; int batchSize = msgMetadata.getNumMessagesInBatch(); - totalMessages += batchSize; - totalBytes += metadataAndPayload.readableBytes(); - totalChunkedMessages += msgMetadata.hasChunkId() ? 1 : 0; - batchSizes.setBatchSize(i, batchSize); long[] ackSet = null; if (indexesAcks != null && cursor != null) { PositionImpl position = PositionImpl.get(entry.getLedgerId(), entry.getEntryId()); @@ -262,6 +257,12 @@ public int filterEntriesForConsumer(@Nullable MessageMetadata[] metadataArray, i } } + totalEntries++; + totalMessages += batchSize; + totalBytes += metadataAndPayload.readableBytes(); + totalChunkedMessages += msgMetadata.hasChunkId() ? 1 : 0; + batchSizes.setBatchSize(i, batchSize); + BrokerInterceptor interceptor = subscription.interceptor(); if (null != interceptor) { // keep for compatibility if users has implemented the old interface diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java index 34cc3bc1ca526..d0aac21b2382c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java @@ -254,6 +254,62 @@ private void testFilterMsgsInPendingAckStateWhenConsumerDisconnect(boolean enabl Assert.assertEquals(receiveCounter, count / 2); } + @Test + private void testMsgsInPendingAckStateWouldNotGetTheConsumerStuck() throws Exception { + final String topicName = NAMESPACE1 + "/testMsgsInPendingAckStateWouldNotGetTheConsumerStuck"; + final String subscription = "test"; + + Producer producer = pulsarClient.newProducer(Schema.INT32) + .topic(topicName) + .create(); + Consumer consumer = pulsarClient.newConsumer(Schema.INT32) + .topic(topicName) + .subscriptionName(subscription) + .subscriptionType(SubscriptionType.Shared) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + + int numStep1Receive = 2, numStep2Receive = 2, numStep3Receive = 2; + int numTotalMessage = numStep1Receive + numStep2Receive + numStep3Receive; + + for (int i = 0; i < numTotalMessage; i++) { + producer.send(i); + } + + Transaction step1Txn = getTxn(); + Transaction step2Txn = getTxn(); + + // Step 1, try to consume some messages but do not commit the transaction + for (int i = 0; i < numStep1Receive; i++) { + consumer.acknowledgeAsync(consumer.receive().getMessageId(), step1Txn).get(); + } + + // Step 2, try to consume some messages and commit the transaction + for (int i = 0; i < numStep2Receive; i++) { + consumer.acknowledgeAsync(consumer.receive().getMessageId(), step2Txn).get(); + } + + // commit step2Txn + step2Txn.commit().get(); + + // close and re-create consumer + consumer.close(); + Consumer consumer2 = pulsarClient.newConsumer(Schema.INT32) + .topic(topicName) + .receiverQueueSize(numStep3Receive) + .subscriptionName(subscription) + .subscriptionType(SubscriptionType.Shared) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + + // Step 3, try to consume the rest messages and should receive all of them + for (int i = 0; i < numStep3Receive; i++) { + // should get the message instead of timeout + Message msg = consumer2.receive(3, TimeUnit.SECONDS); + Assert.assertEquals(msg.getValue(), numStep1Receive + numStep2Receive + i); + } + } + @Test(dataProvider="enableBatch") private void produceCommitTest(boolean enableBatch) throws Exception { @Cleanup From 1174b57c3a81f8e1d9bece2d65d206207cc9c35b Mon Sep 17 00:00:00 2001 From: ruihongzhou Date: Tue, 26 Sep 2023 14:25:49 +0800 Subject: [PATCH 2/2] [fix][txn] fix test --- .../org/apache/pulsar/client/impl/TransactionEndToEndTest.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java index d0aac21b2382c..348fb04b7dd23 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java @@ -259,9 +259,11 @@ private void testMsgsInPendingAckStateWouldNotGetTheConsumerStuck() throws Excep final String topicName = NAMESPACE1 + "/testMsgsInPendingAckStateWouldNotGetTheConsumerStuck"; final String subscription = "test"; + @Cleanup Producer producer = pulsarClient.newProducer(Schema.INT32) .topic(topicName) .create(); + @Cleanup Consumer consumer = pulsarClient.newConsumer(Schema.INT32) .topic(topicName) .subscriptionName(subscription) @@ -294,6 +296,7 @@ private void testMsgsInPendingAckStateWouldNotGetTheConsumerStuck() throws Excep // close and re-create consumer consumer.close(); + @Cleanup Consumer consumer2 = pulsarClient.newConsumer(Schema.INT32) .topic(topicName) .receiverQueueSize(numStep3Receive)