This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 7712aa396bf [fix][client] Fix failover/exclusive consumer with batch
cumulate ack issue. (#18454)
7712aa396bf is described below
commit 7712aa396bfca55a79a45761e0a405e90185e0f8
Author: Jiwei Guo <[email protected]>
AuthorDate: Tue Nov 15 11:37:16 2022 +0800
[fix][client] Fix failover/exclusive consumer with batch cumulate ack
issue. (#18454)
---
.../impl/ConsumerDedupPermitsUpdateTest.java | 21 ++++--
.../pulsar/client/impl/NegativeAcksTest.java | 77 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 9 ++-
.../client/impl/MultiTopicsConsumerImpl.java | 6 +-
4 files changed, 105 insertions(+), 8 deletions(-)
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerDedupPermitsUpdateTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerDedupPermitsUpdateTest.java
index 6fc8268ae21..5ef3054d165 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerDedupPermitsUpdateTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerDedupPermitsUpdateTest.java
@@ -116,10 +116,23 @@ public class ConsumerDedupPermitsUpdateTest extends
ProducerConsumerBase {
}
producer.flush();
- for (int i = 0; i < 30; i++) {
- Message<String> msg = consumer.receive();
- assertEquals(msg.getValue(), "new-message-" + i);
- consumer.acknowledge(msg);
+ if (batchingEnabled) {
+ for (int i = 0; i < 30; i++) {
+ Message<String> msg = consumer.receive();
+ assertEquals(msg.getValue(), "hello-" + i);
+ consumer.acknowledge(msg);
+ }
+ for (int i = 0; i < 30; i++) {
+ Message<String> msg = consumer.receive();
+ assertEquals(msg.getValue(), "new-message-" + i);
+ consumer.acknowledge(msg);
+ }
+ } else {
+ for (int i = 0; i < 30; i++) {
+ Message<String> msg = consumer.receive();
+ assertEquals(msg.getValue(), "new-message-" + i);
+ consumer.acknowledge(msg);
+ }
}
}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
index 9250d085bad..71de1ff544e 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
@@ -24,6 +24,7 @@ import static org.testng.Assert.assertTrue;
import java.util.HashSet;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
@@ -35,6 +36,7 @@ import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
+import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
@@ -362,4 +364,79 @@ public class NegativeAcksTest extends ProducerConsumerBase
{
// There should be no more messages
assertNull(consumer.receive(100, TimeUnit.MILLISECONDS));
}
+
+ @Test
+ public void testFailoverConsumerBatchCumulateAck() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("my-topic");
+ admin.topics().createPartitionedTopic(topic, 2);
+
+ @Cleanup
+ Consumer<Integer> consumer = pulsarClient.newConsumer(Schema.INT32)
+ .topic(topic)
+ .subscriptionName("sub")
+ .subscriptionType(SubscriptionType.Failover)
+ .enableBatchIndexAcknowledgment(true)
+ .acknowledgmentGroupTime(100, TimeUnit.MILLISECONDS)
+ .receiverQueueSize(10)
+ .subscribe();
+
+ @Cleanup
+ Producer<Integer> producer = pulsarClient.newProducer(Schema.INT32)
+ .topic(topic)
+ .batchingMaxMessages(10)
+ .batchingMaxPublishDelay(3, TimeUnit.SECONDS)
+ .blockIfQueueFull(true)
+ .create();
+
+ int count = 0;
+ Set<Integer> datas = new HashSet<>();
+ CountDownLatch producerLatch = new CountDownLatch(10);
+ while (count < 10) {
+ datas.add(count);
+ producer.sendAsync(count).whenComplete((m, e) -> {
+ producerLatch.countDown();
+ });
+ count++;
+ }
+ producerLatch.await();
+ CountDownLatch consumerLatch = new CountDownLatch(1);
+ new Thread(new Runnable() {
+ @Override
+ public void run() {
+ consumer.receiveAsync()
+ .thenCompose(m -> {
+ log.info("received one msg : {}",
m.getMessageId());
+ datas.remove(m.getValue());
+ return consumer.acknowledgeCumulativeAsync(m);
+ })
+ .thenAccept(ignore -> {
+ try {
+ Thread.sleep(500);
+ consumer.redeliverUnacknowledgedMessages();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .whenComplete((r, e) -> {
+ consumerLatch.countDown();
+ });
+ }
+ }).start();
+ consumerLatch.await();
+ Thread.sleep(500);
+ count = 0;
+ while(true) {
+ Message<Integer> msg = consumer.receive(5, TimeUnit.SECONDS);
+ if (msg == null) {
+ break;
+ }
+ consumer.acknowledgeCumulative(msg);
+ Thread.sleep(200);
+ datas.remove(msg.getValue());
+ log.info("received msg : {}", msg.getMessageId());
+ count++;
+ }
+ Assert.assertEquals(count, 9);
+ Assert.assertEquals(0, datas.size());
+ }
}
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 99441631bb9..3063426ea69 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
@@ -1275,9 +1275,9 @@ public class ConsumerImpl<T> extends ConsumerBase<T>
implements ConnectionHandle
final int numMessages = msgMetadata.getNumMessagesInBatch();
final int numChunks = msgMetadata.hasNumChunksFromMsg() ?
msgMetadata.getNumChunksFromMsg() : 0;
final boolean isChunkedMessage = numChunks > 1;
-
MessageIdImpl msgId = new MessageIdImpl(messageId.getLedgerId(),
messageId.getEntryId(), getPartitionIndex());
- if (acknowledgmentsGroupingTracker.isDuplicate(msgId)) {
+ if (numMessages == 1 && !msgMetadata.hasNumMessagesInBatch()
+ && acknowledgmentsGroupingTracker.isDuplicate(msgId)) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Ignoring message as it was already being
acked earlier by same consumer {}/{}",
topic, subscription, consumerName, msgId);
@@ -1541,7 +1541,10 @@ public class ConsumerImpl<T> extends ConsumerBase<T>
implements ConnectionHandle
skippedMessages++;
continue;
}
-
+ }
+ if
(acknowledgmentsGroupingTracker.isDuplicate(message.getMessageId())) {
+ skippedMessages++;
+ continue;
}
executeNotifyCallback(message);
}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index 22078bf1532..3c50d85db4c 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -261,7 +261,11 @@ public class MultiTopicsConsumerImpl<T> extends
ConsumerBase<T> {
return;
}
// Process the message, add to the queue and trigger listener or
async callback
- messages.forEach(msg -> messageReceived(consumer, msg));
+ messages.forEach(msg -> {
+ if (isValidConsumerEpoch((MessageImpl<T>) msg)) {
+ messageReceived(consumer, msg);
+ }
+ });
int size = incomingMessages.size();
int maxReceiverQueueSize = getCurrentReceiverQueueSize();
