You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by bo...@apache.org on 2022/11/30 08:43:03 UTC
[pulsar] branch branch-2.9 updated (8deccd75fe5 -> 91dac3e244e)
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a change to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 8deccd75fe5 [fix][flaky-test] PersistentFailoverE2ETest.testSimpleConsumerEventsWithPartition (#16493)
new 532d4b8fe65 [fix][test] Fix jvm oom on Unit Test broker group 1 (#16542)
new 91dac3e244e [fix][txn] fix ack with txn compute ackedCount error (#17016)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/pulsar/broker/service/Consumer.java | 25 ++++++------
.../pulsar/client/api/MultiTopicsConsumerTest.java | 32 +++++++--------
.../client/impl/TransactionEndToEndTest.java | 46 ++++++++++++++++++++++
3 files changed, 75 insertions(+), 28 deletions(-)
[pulsar] 01/02: [fix][test] Fix jvm oom on Unit Test broker group 1 (#16542)
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 532d4b8fe65a0d68026c06b513fe1f9cb1fbde8b
Author: lipenghui <pe...@apache.org>
AuthorDate: Tue Jul 12 21:13:56 2022 +0800
[fix][test] Fix jvm oom on Unit Test broker group 1 (#16542)
(cherry picked from commit 3752a1170bd7b362901c3e792548c62352b4b1ad)
---
.../pulsar/client/api/MultiTopicsConsumerTest.java | 32 ++++++++++------------
1 file changed, 15 insertions(+), 17 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
index 29ecb39853a..8bcb1824902 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
@@ -65,11 +65,17 @@ public class MultiTopicsConsumerTest extends ProducerConsumerBase {
super.internalCleanup();
}
- @Override
- protected PulsarClient createNewPulsarClient(ClientBuilder clientBuilder) throws PulsarClientException {
- ClientConfigurationData conf =
- ((ClientBuilderImpl) clientBuilder).getClientConfigurationData();
- return new PulsarClientImpl(conf) {
+ // test that reproduces the issue https://github.com/apache/pulsar/issues/12024
+ // where closing the consumer leads to an endless receive loop
+ @Test
+ public void testMultiTopicsConsumerCloses() throws Exception {
+ String topicNameBase = "persistent://my-property/my-ns/my-topic-consumer-closes-";
+
+ ClientConfigurationData conf = ((ClientBuilderImpl) PulsarClient.builder().serviceUrl(lookupUrl.toString()))
+ .getClientConfigurationData();
+
+ @Cleanup
+ PulsarClientImpl client = new PulsarClientImpl(conf) {
{
ScheduledExecutorService internalExecutorService =
(ScheduledExecutorService) super.getScheduledExecutorProvider().getExecutor();
@@ -85,31 +91,23 @@ public class MultiTopicsConsumerTest extends ProducerConsumerBase {
return internalExecutorServiceDelegate;
}
};
- }
-
- // test that reproduces the issue https://github.com/apache/pulsar/issues/12024
- // where closing the consumer leads to an endless receive loop
- @Test
- public void testMultiTopicsConsumerCloses() throws Exception {
- String topicNameBase = "persistent://my-property/my-ns/my-topic-consumer-closes-";
-
@Cleanup
- Producer<byte[]> producer1 = pulsarClient.newProducer()
+ Producer<byte[]> producer1 = client.newProducer()
.topic(topicNameBase + "1")
.enableBatching(false)
.create();
@Cleanup
- Producer<byte[]> producer2 = pulsarClient.newProducer()
+ Producer<byte[]> producer2 = client.newProducer()
.topic(topicNameBase + "2")
.enableBatching(false)
.create();
@Cleanup
- Producer<byte[]> producer3 = pulsarClient.newProducer()
+ Producer<byte[]> producer3 = client.newProducer()
.topic(topicNameBase + "3")
.enableBatching(false)
.create();
- Consumer<byte[]> consumer = pulsarClient
+ Consumer<byte[]> consumer = client
.newConsumer()
.topics(Lists.newArrayList(topicNameBase + "1", topicNameBase + "2", topicNameBase + "3"))
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
[pulsar] 02/02: [fix][txn] fix ack with txn compute ackedCount error (#17016)
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 91dac3e244e1342110b4199ae38a819501d3df0d
Author: congbo <39...@users.noreply.github.com>
AuthorDate: Thu Sep 1 12:29:04 2022 +0800
[fix][txn] fix ack with txn compute ackedCount error (#17016)
Co-authored-by: congbobo184 <co...@github.com>
(cherry picked from commit 176b0d6e9e0d647c611cfdd359e5088ccb58788c)
---
.../org/apache/pulsar/broker/service/Consumer.java | 25 ++++++------
.../client/impl/TransactionEndToEndTest.java | 46 ++++++++++++++++++++++
2 files changed, 60 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
index 25dd3b908e5..1659d25c74e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -480,25 +480,28 @@ public class Consumer {
LongAdder totalAckCount = new LongAdder();
for (int i = 0; i < ack.getMessageIdsCount(); i++) {
MessageIdData msgId = ack.getMessageIdAt(i);
- PositionImpl position;
+ PositionImpl position = PositionImpl.get(msgId.getLedgerId(), msgId.getEntryId());
+ // acked count at least one
long ackedCount = 0;
- long batchSize = getBatchSize(msgId);
+ long batchSize = 0;
+ if (msgId.hasBatchSize()) {
+ batchSize = msgId.getBatchSize();
+ // ack batch messages set ackeCount = batchSize
+ ackedCount = msgId.getBatchSize();
+ positionsAcked.add(new MutablePair<>(position, msgId.getBatchSize()));
+ } else {
+ // ack no batch message set ackedCount = 1
+ ackedCount = 1;
+ positionsAcked.add(new MutablePair<>(position, (int) batchSize));
+ }
Consumer ackOwnerConsumer = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId());
if (msgId.getAckSetsCount() > 0) {
long[] ackSets = new long[msgId.getAckSetsCount()];
for (int j = 0; j < msgId.getAckSetsCount(); j++) {
ackSets[j] = msgId.getAckSetAt(j);
}
- position = PositionImpl.get(msgId.getLedgerId(), msgId.getEntryId(), ackSets);
+ position.setAckSet(ackSets);
ackedCount = getAckedCountForTransactionAck(batchSize, ackSets);
- } else {
- position = PositionImpl.get(msgId.getLedgerId(), msgId.getEntryId());
- ackedCount = batchSize;
- }
- if (msgId.hasBatchSize()) {
- positionsAcked.add(new MutablePair<>(position, msgId.getBatchSize()));
- } else {
- positionsAcked.add(new MutablePair<>(position, (int) batchSize));
}
addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
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 2d174b9b1fb..c7ae3a53109 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
@@ -1110,6 +1110,52 @@ public class TransactionEndToEndTest extends TransactionTestBase {
}
}
+ @Test
+ public void testAckWithTransactionReduceUnackCountNotInPendingAcks() throws Exception {
+ final String topic = "persistent://" + NAMESPACE1 + "/testAckWithTransactionReduceUnackCountNotInPendingAcks";
+ final String subName = "test";
+ @Cleanup
+ ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topic)
+ .batchingMaxPublishDelay(1, TimeUnit.SECONDS)
+ .sendTimeout(1, TimeUnit.SECONDS)
+ .create();
+
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName(subName)
+ .subscribe();
+
+ // send 5 messages with one batch
+ for (int i = 0; i < 5; i++) {
+ producer.sendAsync((i + "").getBytes(UTF_8));
+ }
+
+ List<MessageId> messageIds = new ArrayList<>();
+
+ // receive the batch messages add to a list
+ for (int i = 0; i < 5; i++) {
+ messageIds.add(consumer.receive().getMessageId());
+ }
+
+ MessageIdImpl messageId = (MessageIdImpl) messageIds.get(0);
+
+
+ // remove the message from the pendingAcks, in fact redeliver will remove the messageId from the pendingAck
+ getPulsarServiceList().get(0).getBrokerService().getTopic(topic, false)
+ .get().get().getSubscription(subName).getConsumers().get(0).getPendingAcks()
+ .remove(messageId.ledgerId, messageId.entryId);
+
+ Transaction txn = getTxn();
+ consumer.acknowledgeAsync(messageIds.get(1), txn).get();
+
+ // ack one message, the unack count is 4
+ assertEquals(getPulsarServiceList().get(0).getBrokerService().getTopic(topic, false)
+ .get().get().getSubscription(subName).getConsumers().get(0).getUnackedMessages(), 4);
+ }
+
@Test
public void testSendTxnAckMessageToDLQ() throws Exception {
String topic = NAMESPACE1 + "/testSendTxnAckMessageToDLQ";