You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/08/09 14:11:57 UTC
(pulsar) branch branch-2.11 updated (6d140af3fdb -> ecbfcdca524)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 6d140af3fdb [improve][client] Add maxConnectionsPerHost and connectionMaxIdleSeconds to PulsarAdminBuilder (#22541)
new 651c8d881f1 [fix][client] fix negative message re-delivery twice issue (#20750)
new 581ad607ced [fix][build] Fix compatibility issue introduced by #20750
new ecbfcdca524 [fix][flaky-test]NegativeAcksTest.testNegativeAcksWithBatchAckEnabled (#17893)
The 3 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/client/impl/NegativeAcksTest.java | 6 +++++-
.../java/org/apache/pulsar/client/impl/ConsumerBase.java | 4 ++++
.../java/org/apache/pulsar/client/impl/ConsumerImpl.java | 16 +++++++++++++++-
.../pulsar/client/impl/MultiTopicsConsumerImpl.java | 3 +++
.../apache/pulsar/client/impl/NegativeAcksTracker.java | 12 +++++++++---
.../apache/pulsar/client/impl/UnAckedMessageTracker.java | 8 ++++++++
6 files changed, 44 insertions(+), 5 deletions(-)
(pulsar) 03/03: [fix][flaky-test]NegativeAcksTest.testNegativeAcksWithBatchAckEnabled (#17893)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ecbfcdca5246ebf66eea7907e725e85540188f89
Author: fengyubiao <99...@qq.com>
AuthorDate: Fri Sep 30 17:32:55 2022 +0800
[fix][flaky-test]NegativeAcksTest.testNegativeAcksWithBatchAckEnabled (#17893)
Fixes: #16864
### Motivation
I think it is a wrong configuration(`ackTimeout 1s`) when writing the code, the original design is set `negativeAckRedeliveryDelay 1s`
The process expects:
- send 10 messages in one batch
- submit a batch.
- receive 10 messages, do negative acknowledge
- after `1s`, will trigger `redelivery`
- receive 10 messages again
The real process:
- send 1 message
- Reach the batch time limit, and submit a batch. return `msgId_1`
- send 9 messages in another batch
- submit a batch. return `msgId_2`
- receive 10 messages, do negative acknowledge
- push the `msgId_1` to `negativeAcksTracker`
- push the `msgId_2` to `unAckedMessageTracker`
- after `1s`, will trigger redelivery `msgId_2` by `unAckedMessageTracker`
- receive 9 messages( `msgId_2` ) again
- after `60s`, will trigger redelivery `msgId_1` by `negativeAcksTracker`. <strong>(High light)</strong> Test execution timeout!
- receive 1 messages( `msgId_1` ) again
### Modifications
- remove conf: `ackTimeout`
- set `negativeAckRedeliveryDelay 1s`
### Documentation
- [x] `doc-not-needed`
### Matching PR in forked repository
PR in forked repository:
- https://github.com/poorbarcode/pulsar/pull/18
(cherry picked from commit 85b1138593a49a29c269266ca3bc1692b00f0971)
---
.../src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
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 a25c4dee242..ea3b53b46a7 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
@@ -329,7 +329,7 @@ public class NegativeAcksTest extends ProducerConsumerBase {
.acknowledgmentGroupTime(0, TimeUnit.SECONDS)
.subscriptionType(SubscriptionType.Shared)
.enableBatchIndexAcknowledgment(true)
- .ackTimeout(1000, TimeUnit.MILLISECONDS)
+ .negativeAckRedeliveryDelay(1, TimeUnit.SECONDS)
.subscribe();
@Cleanup
(pulsar) 02/03: [fix][build] Fix compatibility issue introduced by #20750
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 581ad607ced40e7a157b6b38bfe6f3914984581a
Author: xymeng <19...@qq.com>
AuthorDate: Tue Jan 23 12:46:02 2024 +0800
[fix][build] Fix compatibility issue introduced by #20750
(cherry picked from commit 05ac1f9dda575e668627df966504f2ba422354ea)
---
.../org/apache/pulsar/client/impl/NegativeAcksTest.java | 3 ++-
.../java/org/apache/pulsar/client/impl/ConsumerImpl.java | 15 ++++++++++++++-
2 files changed, 16 insertions(+), 2 deletions(-)
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 576b8778251..a25c4dee242 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
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import static org.junit.Assert.assertTrue;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
@@ -150,7 +151,7 @@ public class NegativeAcksTest extends ProducerConsumerBase {
consumer.negativeAcknowledge(msg);
}
- assertTrue(consumer instanceof ConsumerBase<String>);
+ assertTrue(consumer instanceof ConsumerBase);
assertEquals(((ConsumerBase<String>) consumer).getUnAckedMessageTracker().size(), 0);
Set<String> receivedMessages = new HashSet<>();
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 85cafe644fb..0c3b8da4396 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
@@ -806,7 +806,20 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
negativeAcksTracker.add(message);
// Ensure the message is not redelivered for ack-timeout, since we did receive an "ack"
- unAckedMessageTracker.remove(MessageIdAdvUtils.discardBatch(message.getMessageId()));
+ unAckedMessageTracker.remove(discardBatch(message.getMessageId()));
+ }
+
+ static MessageIdImpl discardBatch(MessageId messageId) {
+ if (messageId instanceof ChunkMessageIdImpl) {
+ return (MessageIdImpl) messageId;
+ }
+ MessageIdImpl msgId;
+ if (messageId instanceof TopicMessageIdImpl) {
+ msgId = (MessageIdImpl) ((TopicMessageIdImpl) messageId).getInnerMessageId();
+ } else {
+ msgId = (MessageIdImpl) messageId;
+ }
+ return new MessageIdImpl(msgId.getLedgerId(), msgId.getEntryId(), msgId.getPartitionIndex());
}
@Override
(pulsar) 01/03: [fix][client] fix negative message re-delivery twice issue (#20750)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 651c8d881f103cd637a9dba6e69a1661fee63278
Author: AloysZhang <lo...@gmail.com>
AuthorDate: Mon Jul 17 15:43:52 2023 +0800
[fix][client] fix negative message re-delivery twice issue (#20750)
(cherry picked from commit ecd16d68e29700cc9521e6d77d124a403c3486c9)
(cherry picked from commit cbc3ea169367e7e14cae3682f36e0da4fc9b587f)
---
.../java/org/apache/pulsar/client/impl/NegativeAcksTest.java | 3 +++
.../java/org/apache/pulsar/client/impl/ConsumerBase.java | 4 ++++
.../java/org/apache/pulsar/client/impl/ConsumerImpl.java | 3 ++-
.../apache/pulsar/client/impl/MultiTopicsConsumerImpl.java | 3 +++
.../org/apache/pulsar/client/impl/NegativeAcksTracker.java | 12 +++++++++---
.../org/apache/pulsar/client/impl/UnAckedMessageTracker.java | 8 ++++++++
6 files changed, 29 insertions(+), 4 deletions(-)
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 30aefd90b80..576b8778251 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
@@ -150,6 +150,9 @@ public class NegativeAcksTest extends ProducerConsumerBase {
consumer.negativeAcknowledge(msg);
}
+ assertTrue(consumer instanceof ConsumerBase<String>);
+ assertEquals(((ConsumerBase<String>) consumer).getUnAckedMessageTracker().size(), 0);
+
Set<String> receivedMessages = new HashSet<>();
// All the messages should be received again
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
index 5d77da8a046..7233480bfd5 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -174,6 +174,10 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
initReceiverQueueSize();
}
+ protected UnAckedMessageTracker getUnAckedMessageTracker() {
+ return unAckedMessageTracker;
+ }
+
protected void triggerBatchReceiveTimeoutTask() {
if (!hasBatchReceiveTimeout() && batchReceivePolicy.getTimeoutMs() > 0) {
batchReceiveTimeout = client.timer().newTimeout(this::pendingBatchReceiveTask,
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 35f47076a9b..85cafe644fb 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
@@ -394,6 +394,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
return connectionHandler;
}
+ @Override
public UnAckedMessageTracker getUnAckedMessageTracker() {
return unAckedMessageTracker;
}
@@ -805,7 +806,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
negativeAcksTracker.add(message);
// Ensure the message is not redelivered for ack-timeout, since we did receive an "ack"
- unAckedMessageTracker.remove(message.getMessageId());
+ unAckedMessageTracker.remove(MessageIdAdvUtils.discardBatch(message.getMessageId()));
}
@Override
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 52d8f56a812..36fdf4427ea 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
@@ -597,6 +597,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
ConsumerImpl<T> consumer = consumers.get(topicMessageId.getTopicPartitionName());
consumer.negativeAcknowledge(topicMessageId.getInnerMessageId());
+ unAckedMessageTracker.remove(messageId);
}
@Override
@@ -607,6 +608,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
ConsumerImpl<T> consumer = consumers.get(topicMessageId.getTopicPartitionName());
consumer.negativeAcknowledge(message);
+ unAckedMessageTracker.remove(messageId);
}
@Override
@@ -883,6 +885,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
return stats;
}
+ @Override
public UnAckedMessageTracker getUnAckedMessageTracker() {
return unAckedMessageTracker;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java
index 6273f4d582e..f24872fb619 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java
@@ -30,8 +30,11 @@ import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.RedeliveryBackoff;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
class NegativeAcksTracker implements Closeable {
+ private static final Logger log = LoggerFactory.getLogger(NegativeAcksTracker.class);
private HashMap<MessageId, Long> nackedMessages = null;
@@ -77,9 +80,12 @@ class NegativeAcksTracker implements Closeable {
}
});
- messagesToRedeliver.forEach(nackedMessages::remove);
- consumer.onNegativeAcksSend(messagesToRedeliver);
- consumer.redeliverUnacknowledgedMessages(messagesToRedeliver);
+ if (!messagesToRedeliver.isEmpty()) {
+ messagesToRedeliver.forEach(nackedMessages::remove);
+ consumer.onNegativeAcksSend(messagesToRedeliver);
+ log.info("[{}] {} messages will be re-delivered", consumer, messagesToRedeliver.size());
+ consumer.redeliverUnacknowledgedMessages(messagesToRedeliver);
+ }
this.timeout = timer.newTimeout(this::triggerRedelivery, timerIntervalNanos, TimeUnit.NANOSECONDS);
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java
index 5a706720dc5..5386f6d4fe5 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java
@@ -192,6 +192,10 @@ public class UnAckedMessageTracker implements Closeable {
}
public boolean add(MessageId messageId) {
+ if (messageId == null) {
+ return false;
+ }
+
writeLock.lock();
try {
HashSet<MessageId> partition = timePartitions.peekLast();
@@ -220,6 +224,10 @@ public class UnAckedMessageTracker implements Closeable {
}
public boolean remove(MessageId messageId) {
+ if (messageId == null) {
+ return false;
+ }
+
writeLock.lock();
try {
boolean removed = false;