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 20:28:47 UTC
(pulsar) branch branch-4.0 updated (86c31c6baca -> 2c05137289a)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 86c31c6baca [improve][broker] Decouple pulsar_storage_backlog_age_seconds metric with backlogQuota check (#23619)
new 142897c1a12 [fix][client] Fix deadlock of NegativeAcksTracker (#23651)
new 2c05137289a [improve] Upgrade oxia-java to 0.4.10 and fix closing of OxiaMetadataStore (#23653)
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:
distribution/server/src/assemble/LICENSE.bin.txt | 4 +--
pom.xml | 2 +-
.../pulsar/client/impl/NegativeAcksTracker.java | 40 ++++++++++++----------
.../metadata/impl/oxia/OxiaMetadataStore.java | 8 +++--
4 files changed, 29 insertions(+), 25 deletions(-)
(pulsar) 02/02: [improve] Upgrade oxia-java to 0.4.10 and fix closing of OxiaMetadataStore (#23653)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2c05137289af91f4b748953ebb0aa80f9b077443
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Nov 28 18:01:17 2024 +0200
[improve] Upgrade oxia-java to 0.4.10 and fix closing of OxiaMetadataStore (#23653)
(cherry picked from commit 7e6fa554012b237308db1bebf3c625e103bf0db5)
---
distribution/server/src/assemble/LICENSE.bin.txt | 4 ++--
pom.xml | 2 +-
.../org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java | 8 +++++---
3 files changed, 8 insertions(+), 6 deletions(-)
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index 7c66460c216..fd393cfec9b 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -481,8 +481,8 @@ The Apache Software License, Version 2.0
* Prometheus
- io.prometheus-simpleclient_httpserver-0.16.0.jar
* Oxia
- - io.streamnative.oxia-oxia-client-api-0.4.9.jar
- - io.streamnative.oxia-oxia-client-0.4.9.jar
+ - io.streamnative.oxia-oxia-client-api-0.4.10.jar
+ - io.streamnative.oxia-oxia-client-0.4.10.jar
* OpenHFT
- net.openhft-zero-allocation-hashing-0.16.jar
* Java JSON WebTokens
diff --git a/pom.xml b/pom.xml
index ee0a0a27fcf..1f0dd764422 100644
--- a/pom.xml
+++ b/pom.xml
@@ -251,7 +251,7 @@ flexible messaging model and an intuitive client API.</description>
<apache-http-client.version>4.5.13</apache-http-client.version>
<apache-httpcomponents.version>4.4.15</apache-httpcomponents.version>
<jetcd.version>0.7.7</jetcd.version>
- <oxia.version>0.4.9</oxia.version>
+ <oxia.version>0.4.10</oxia.version>
<snakeyaml.version>2.0</snakeyaml.version>
<ant.version>1.10.12</ant.version>
<seancfoley.ipaddress.version>5.5.0</seancfoley.ipaddress.version>
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java
index 27cd4a2d2f6..4073eb8dcca 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java
@@ -297,10 +297,12 @@ public class OxiaMetadataStore extends AbstractMetadataStore {
@Override
public void close() throws Exception {
- if (client != null) {
- client.close();
+ if (isClosed.compareAndSet(false, true)) {
+ if (client != null) {
+ client.close();
+ }
+ super.close();
}
- super.close();
}
public Optional<MetadataEventSynchronizer> getMetadataEventSynchronizer() {
(pulsar) 01/02: [fix][client] Fix deadlock of NegativeAcksTracker (#23651)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 142897c1a1239ad8e119b008d898afcb430d550e
Author: Wenzhi Feng <52...@users.noreply.github.com>
AuthorDate: Thu Nov 28 22:58:24 2024 +0800
[fix][client] Fix deadlock of NegativeAcksTracker (#23651)
(cherry picked from commit 68eb8f29c50f16f0aa50ecacec16882b82230122)
---
.../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) {