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/11/28 14:58:31 UTC
(pulsar) branch master updated: [fix][client] Fix deadlock of NegativeAcksTracker (#23651)
This is an automated email from the ASF dual-hosted git repository.
lhotari 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 68eb8f29c50 [fix][client] Fix deadlock of NegativeAcksTracker (#23651)
68eb8f29c50 is described below
commit 68eb8f29c50f16f0aa50ecacec16882b82230122
Author: Wenzhi Feng <52...@users.noreply.github.com>
AuthorDate: Thu Nov 28 22:58:24 2024 +0800
[fix][client] Fix deadlock of NegativeAcksTracker (#23651)
---
.../pulsar/client/impl/NegativeAcksTracker.java | 40 ++++++++++++----------
1 file changed, 21 insertions(+), 19 deletions(-)
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 e1724ebb85c..5256ebf04f4 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
@@ -68,36 +68,38 @@ class NegativeAcksTracker implements Closeable {
}
}
- private synchronized void triggerRedelivery(Timeout t) {
- if (nackedMessages.isEmpty()) {
- this.timeout = null;
- return;
- }
-
- // Group all the nacked messages into one single re-delivery request
+ private void triggerRedelivery(Timeout t) {
Set<MessageId> messagesToRedeliver = new HashSet<>();
- long now = System.nanoTime();
- nackedMessages.forEach((ledgerId, entryId, partitionIndex, timestamp) -> {
- if (timestamp < now) {
- MessageId msgId = new MessageIdImpl(ledgerId, entryId,
- // need to covert non-partitioned topic partition index to -1
- (int) (partitionIndex == NON_PARTITIONED_TOPIC_PARTITION_INDEX ? -1 : partitionIndex));
- addChunkedMessageIdsAndRemoveFromSequenceMap(msgId, messagesToRedeliver, this.consumer);
- messagesToRedeliver.add(msgId);
+ synchronized (this) {
+ if (nackedMessages.isEmpty()) {
+ this.timeout = null;
+ return;
}
- });
- if (!messagesToRedeliver.isEmpty()) {
+ long now = System.nanoTime();
+ nackedMessages.forEach((ledgerId, entryId, partitionIndex, timestamp) -> {
+ if (timestamp < now) {
+ MessageId msgId = new MessageIdImpl(ledgerId, entryId,
+ // need to covert non-partitioned topic partition index to -1
+ (int) (partitionIndex == NON_PARTITIONED_TOPIC_PARTITION_INDEX ? -1 : partitionIndex));
+ addChunkedMessageIdsAndRemoveFromSequenceMap(msgId, messagesToRedeliver, this.consumer);
+ messagesToRedeliver.add(msgId);
+ }
+ });
for (MessageId messageId : messagesToRedeliver) {
nackedMessages.remove(((MessageIdImpl) messageId).getLedgerId(),
((MessageIdImpl) messageId).getEntryId());
}
+ this.timeout = timer.newTimeout(this::triggerRedelivery, timerIntervalNanos, TimeUnit.NANOSECONDS);
+ }
+
+ // release the lock of NegativeAcksTracker before calling consumer.redeliverUnacknowledgedMessages,
+ // in which we may acquire the lock of consumer, leading to potential deadlock.
+ if (!messagesToRedeliver.isEmpty()) {
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);
}
public synchronized void add(MessageId messageId) {