Skip to content
New issue

Have a question about this project? # for a free GitHub account to open an issue and contact its maintainers and the community.

#

By clicking “#”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? # to your account

[improve] [broker] replace HashMap with inner implementation ConcurrentLongLongPairHashMap in Negative Ack Tracker. #23582

Merged
merged 6 commits into from
Nov 13, 2024
Merged

[improve] [broker] replace HashMap with inner implementation ConcurrentLongLongPairHashMap in Negative Ack Tracker. #23582

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
0ca887c
replace HashMap with inner implementation ConcurrentLongLongPairHashMap
thetumbled Nov 11, 2024
c2e20c5
set the concurrencyLevel to 1.
thetumbled Nov 12, 2024
9c2cb70
fix check style.
thetumbled Nov 12, 2024
004eff9
fix test code.
thetumbled Nov 12, 2024
8fe0b93
fix.
thetumbled Nov 13, 2024
35863de
fix checkstyle.
thetumbled Nov 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,7 @@

for (int i = 0; i < N; i++) {
Message<String> msg = consumer.receive();
consumer.negativeAcknowledge(msg);

Check failure on line 146 in pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java

View workflow job for this annotation

GitHub Actions / CI - Unit - Brokers - Client Impl

NegativeAcksTest.testNegativeAcks 

Keys and values must be >= 0
}

for (int i = 0; i < N; i++) {
Expand Down Expand Up @@ -311,15 +311,15 @@
// negative topic message id
consumer.negativeAcknowledge(topicMessageId);
NegativeAcksTracker negativeAcksTracker = consumer.getNegativeAcksTracker();
assertEquals(negativeAcksTracker.getNackedMessagesCount().orElse(-1).intValue(), 1);
assertEquals(negativeAcksTracker.getNackedMessagesCount().orElse((long) -1).longValue(), 1L);
assertEquals(unAckedMessageTracker.size(), 0);
negativeAcksTracker.close();
// negative batch message id
unAckedMessageTracker.add(messageId);
consumer.negativeAcknowledge(batchMessageId);
consumer.negativeAcknowledge(batchMessageId2);
consumer.negativeAcknowledge(batchMessageId3);
assertEquals(negativeAcksTracker.getNackedMessagesCount().orElse(-1).intValue(), 1);
assertEquals(negativeAcksTracker.getNackedMessagesCount().orElse((long) -1).longValue(), 1L);
assertEquals(unAckedMessageTracker.size(), 0);
negativeAcksTracker.close();
}
Expand Down
26 changes: 18 additions & 8 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,22 +23,23 @@
import io.netty.util.Timeout;
import io.netty.util.Timer;
import java.io.Closeable;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.api.RedeliveryBackoff;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap;
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;
private ConcurrentLongLongPairHashMap nackedMessages = null;

private final ConsumerBase<?> consumer;
private final Timer timer;
Expand Down Expand Up @@ -75,15 +76,19 @@ private synchronized void triggerRedelivery(Timeout t) {
// Group all the nacked messages into one single re-delivery request
Set<MessageId> messagesToRedeliver = new HashSet<>();
long now = System.nanoTime();
nackedMessages.forEach((msgId, timestamp) -> {
nackedMessages.forEach((ledgerId, entryId, partitionIndex, timestamp) -> {
if (timestamp < now) {
MessageId msgId = new MessageIdImpl(ledgerId, entryId, (int) partitionIndex);
addChunkedMessageIdsAndRemoveFromSequenceMap(msgId, messagesToRedeliver, this.consumer);
messagesToRedeliver.add(msgId);
}
});

if (!messagesToRedeliver.isEmpty()) {
messagesToRedeliver.forEach(nackedMessages::remove);
for (MessageId messageId : messagesToRedeliver) {
nackedMessages.remove(((MessageIdImpl) messageId).getLedgerId(),
((MessageIdImpl) messageId).getEntryId());
}
consumer.onNegativeAcksSend(messagesToRedeliver);
log.info("[{}] {} messages will be re-delivered", consumer, messagesToRedeliver.size());
consumer.redeliverUnacknowledgedMessages(messagesToRedeliver);
Expand All @@ -102,7 +107,10 @@ public synchronized void add(Message<?> message) {

private synchronized void add(MessageId messageId, int redeliveryCount) {
if (nackedMessages == null) {
nackedMessages = new HashMap<>();
nackedMessages = ConcurrentLongLongPairHashMap.newBuilder()
.autoShrink(true)
.concurrencyLevel(1)
.build();
}

long backoffNs;
Expand All @@ -111,7 +119,9 @@ private synchronized void add(MessageId messageId, int redeliveryCount) {
} else {
backoffNs = nackDelayNanos;
}
nackedMessages.put(MessageIdAdvUtils.discardBatch(messageId), System.nanoTime() + backoffNs);
MessageIdAdv messageIdAdv = MessageIdAdvUtils.discardBatch(messageId);
nackedMessages.put(messageIdAdv.getLedgerId(), messageIdAdv.getEntryId(),
messageIdAdv.getPartitionIndex(), System.nanoTime() + backoffNs);

if (this.timeout == null) {
// Schedule a task and group all the redeliveries for same period. Leave a small buffer to allow for
Expand All @@ -121,8 +131,8 @@ private synchronized void add(MessageId messageId, int redeliveryCount) {
}

@VisibleForTesting
Optional<Integer> getNackedMessagesCount() {
return Optional.ofNullable(nackedMessages).map(HashMap::size);
Optional<Long> getNackedMessagesCount() {
return Optional.ofNullable(nackedMessages).map(ConcurrentLongLongPairHashMap::size);
}

@Override
Expand Down
Loading