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 2025/09/23 10:37:39 UTC
(pulsar) branch branch-3.0 updated (ddfd46a0b15 -> ae9be46ecec)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from ddfd46a0b15 [fix][ml]Fix EOFException after enabled topics offloading (#24753)
new 0e4e54bb3bd [fix][broker] First entry will be skipped if opening NonDurableCursor while trimmed ledger is adding first entry. (#24738)
new cb223f7e454 [improve][broker] If there is a deadlock in the service, the probe should return a failure because the service may be unavailable (#23634)
new a95abb2d7a1 [improve][build] Upgrade Lombok to 1.18.42 to fully support JDK25 (#24763)
new 0f2fe525aad [fix][ci] Fix CI for Java 25 including upgrade of Gradle Develocity Maven extension (#24767)
new 2d68d84b89f [fix][test] Flaky-test: BrokerServiceTest.testShutDownWithMaxConcurrentUnload (#24769)
new ae9be46ecec [fix][misc] Fix compareTo contract violation for NamespaceBundleStats, TimeAverageMessageData and ResourceUnitRanking (#24772)
The 6 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:
.github/workflows/ci-owasp-dependency-check.yaml | 2 +-
.github/workflows/pulsar-ci-flaky.yaml | 2 +-
.github/workflows/pulsar-ci.yaml | 2 +-
.mvn/extensions.xml | 4 +-
.../mledger/impl/NonDurableCursorImpl.java | 5 +-
.../mledger/impl/NonDurableCursorTest.java | 63 +++++++-
pom.xml | 2 +-
pulsar-broker-common/pom.xml | 6 +
.../pulsar/common/configuration/VipStatus.java | 99 +++++++++++--
.../pulsar/common/configuration/VipStatusTest.java | 161 +++++++++++++++++++++
.../loadbalance/impl/SimpleLoadManagerImpl.java | 2 +-
.../extensions/models/TopKBundlesTest.java | 25 ++++
.../pulsar/broker/service/BrokerServiceTest.java | 3 +-
.../data/loadbalancer/NamespaceBundleStats.java | 50 +++----
.../loadbalancer/NamespaceBundleStatsTest.java | 48 ++++++
.../org/apache/pulsar/common/util/CompareUtil.java | 61 ++++++++
.../data/loadbalancer/ResourceUnitRanking.java | 10 +-
.../data/loadbalancer/TimeAverageMessageData.java | 38 ++++-
.../apache/pulsar/common/util/CompareUtilTest.java | 87 +++++++++++
.../loadbalancer/TimeAverageMessageDataTest.java | 44 ++++++
20 files changed, 657 insertions(+), 57 deletions(-)
create mode 100644 pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/VipStatusTest.java
create mode 100644 pulsar-client-admin-api/src/test/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStatsTest.java
create mode 100644 pulsar-common/src/main/java/org/apache/pulsar/common/util/CompareUtil.java
create mode 100644 pulsar-common/src/test/java/org/apache/pulsar/common/util/CompareUtilTest.java
create mode 100644 pulsar-common/src/test/java/org/apache/pulsar/policies/data/loadbalancer/TimeAverageMessageDataTest.java
(pulsar) 05/06: [fix][test] Flaky-test: BrokerServiceTest.testShutDownWithMaxConcurrentUnload (#24769)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2d68d84b89f9a55dcf8a5858ca563dbfec574221
Author: Yike Xiao <km...@live.com>
AuthorDate: Sat Sep 20 18:46:19 2025 +0800
[fix][test] Flaky-test: BrokerServiceTest.testShutDownWithMaxConcurrentUnload (#24769)
(cherry picked from commit 0c6ba1cf72fb12b9a97e3f8438d5c34a5918fb69)
---
.../test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index 9be0c67f199..0a22c47d532 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -176,10 +176,11 @@ public class BrokerServiceTest extends BrokerTestBase {
assertEquals(bundlesData.getNumBundles(), bundleNum);
List<String> list = admin.brokers().getActiveBrokers("test");
assertEquals(list.size(), 1);
+ BrokerService brokerService = pulsar.getBrokerService();
admin.brokers().shutDownBrokerGracefully(1, false);
//We can only unload one bundle per second, so it takes at least 2 seconds.
Awaitility.await().atLeast(bundleNum - 1, TimeUnit.SECONDS).untilAsserted(() -> {
- assertEquals(pulsar.getBrokerService().getTopics().size(), 0);
+ assertEquals(brokerService.getTopics().size(), 0);
});
Awaitility.await().atMost(60, TimeUnit.SECONDS).untilAsserted(() -> {
assertNull(pulsar.getBrokerService());
(pulsar) 01/06: [fix][broker] First entry will be skipped if opening NonDurableCursor while trimmed ledger is adding first entry. (#24738)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0e4e54bb3bddafd6fcaa31bc602fde62d4b16729
Author: zhouyifan279 <88...@users.noreply.github.com>
AuthorDate: Fri Sep 19 02:16:51 2025 +0800
[fix][broker] First entry will be skipped if opening NonDurableCursor while trimmed ledger is adding first entry. (#24738)
Co-authored-by: Yunze Xu <xy...@163.com>
Co-authored-by: Lari Hotari <lh...@users.noreply.github.com>
(cherry picked from commit 7a120f310c286eed78701e10c7e6985bc3b6d001)
---
.../mledger/impl/NonDurableCursorImpl.java | 5 +-
.../mledger/impl/NonDurableCursorTest.java | 63 ++++++++++++++++++++--
2 files changed, 62 insertions(+), 6 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
index 7dbf5400da1..333ec00f52a 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
@@ -43,11 +43,12 @@ public class NonDurableCursorImpl extends ManagedCursorImpl {
// Compare with "latest" position marker by using only the ledger id. Since the C++ client is using 48bits to
// store the entryId, it's not able to pass a Long.max() as entryId. In this case there's no point to require
// both ledgerId and entryId to be Long.max()
- if (startCursorPosition == null || startCursorPosition.compareTo(ledger.lastConfirmedEntry) > 0) {
+ Pair<PositionImpl, Long> lastPositionCounter = ledger.getLastPositionAndCounter();
+ if (startCursorPosition == null || startCursorPosition.compareTo(lastPositionCounter.getLeft()) > 0) {
// Start from last entry
switch (initialPosition) {
case Latest:
- initializeCursorPosition(ledger.getLastPositionAndCounter());
+ initializeCursorPosition(lastPositionCounter);
break;
case Earliest:
initializeCursorPosition(ledger.getFirstPositionAndCounter());
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorTest.java
index 82141bfd0ee..d2cc796a7d1 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorTest.java
@@ -19,15 +19,15 @@
package org.apache.bookkeeper.mledger.impl;
import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.ENTRIES_ADDED_COUNTER_UPDATER;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
-
import com.google.common.collect.Iterables;
-
+import io.netty.buffer.ByteBuf;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@@ -37,8 +37,6 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
-
-import io.netty.buffer.ByteBuf;
import lombok.Cleanup;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
@@ -52,6 +50,11 @@ import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.common.api.proto.CommandSubscribe;
+import org.awaitility.Awaitility;
+import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
@@ -105,6 +108,58 @@ public class NonDurableCursorTest extends MockedBookKeeperTestCase {
ledger.close();
}
+ @Test(timeOut = 20000)
+ void testOpenNonDurableCursorWhileLedgerIsAddingFirstEntryAfterTrimmed() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig().setMaxEntriesPerLedger(1)
+ .setRetentionTime(0, TimeUnit.MILLISECONDS);
+ config.setMinimumRolloverTime(0, TimeUnit.MILLISECONDS);
+ @Cleanup
+ ManagedLedgerImpl ledgerSpy =
+ Mockito.spy((ManagedLedgerImpl) factory.open("non_durable_cursor_while_ledger_trimmed", config));
+
+ ledgerSpy.addEntry("message1".getBytes());
+
+ ledgerSpy.rollCurrentLedgerIfFull();
+ Awaitility.waitAtMost(5, TimeUnit.SECONDS).until(() ->
+ ledgerSpy.getLedgersInfoAsList().size() > 1
+ );
+ CompletableFuture<Void> trimFuture = new CompletableFuture<>();
+ ledgerSpy.trimConsumedLedgersInBackground(trimFuture);
+ trimFuture.join();
+
+ // Use (currentLedgerId, -1) as startCursorPosition after ledger was trimmed
+ PositionImpl startCursorPosition = PositionImpl.get(ledgerSpy.getCurrentLedger().getId(), -1);
+ assertTrue(startCursorPosition.compareTo(ledgerSpy.lastConfirmedEntry) > 0);
+
+ CountDownLatch getLastPositionLatch = new CountDownLatch(1);
+ CountDownLatch newNonDurableCursorLatch = new CountDownLatch(1);
+ Mockito.when(ledgerSpy.getLastPositionAndCounter()).then((Answer<Pair<Position, Long>>) invocation -> {
+ newNonDurableCursorLatch.countDown();
+ getLastPositionLatch.await();
+ return Pair.of(ledgerSpy.lastConfirmedEntry, ENTRIES_ADDED_COUNTER_UPDATER.get(ledgerSpy));
+ });
+
+ CompletableFuture<ManagedCursor> cursorFuture = new CompletableFuture<ManagedCursor>()
+ .completeAsync(() ->
+ new NonDurableCursorImpl(bkc, ledgerSpy, "my_test_cursor",
+ startCursorPosition, CommandSubscribe.InitialPosition.Latest, false)
+ );
+ PositionImpl oldLastConfirmedEntry = ledgerSpy.lastConfirmedEntry;
+
+ // Wait until NonDurableCursorImpl constructor invokes ManagedLedgerImpl.getLastPositionAndCounter
+ newNonDurableCursorLatch.await();
+ // Add first entry after ledger was trimmed
+ ledgerSpy.addEntry("message2".getBytes());
+ assertTrue(oldLastConfirmedEntry.compareTo(ledgerSpy.lastConfirmedEntry) < 0);
+
+ // Unblock NonDurableCursorImpl constructor
+ getLastPositionLatch.countDown();
+
+ // cursor should read from lastConfirmedEntry
+ ManagedCursor cursor = cursorFuture.join();
+ assertEquals(cursor.getReadPosition(), ledgerSpy.lastConfirmedEntry);
+ }
+
@Test(timeOut = 20000)
void testZNodeBypassed() throws Exception {
ManagedLedger ledger = factory.open("my_test_ledger");
(pulsar) 03/06: [improve][build] Upgrade Lombok to 1.18.42 to fully support JDK25 (#24763)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a95abb2d7a189486b5aab004d08f12aa766cbdd3
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Sep 19 16:14:55 2025 +0300
[improve][build] Upgrade Lombok to 1.18.42 to fully support JDK25 (#24763)
(cherry picked from commit e45f3aff510fe2d7444f6627c949f0211d769c57)
---
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pom.xml b/pom.xml
index 1339b7ba61d..2ef4701f7f0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -233,7 +233,7 @@ flexible messaging model and an intuitive client API.</description>
<hppc.version>0.9.1</hppc.version>
<spark-streaming_2.10.version>2.1.0</spark-streaming_2.10.version>
<assertj-core.version>3.24.2</assertj-core.version>
- <lombok.version>1.18.38</lombok.version>
+ <lombok.version>1.18.42</lombok.version>
<jakarta.annotation-api.version>1.3.5</jakarta.annotation-api.version>
<jaxb-api>2.3.1</jaxb-api>
<jakarta.activation.version>1.2.2</jakarta.activation.version>
(pulsar) 04/06: [fix][ci] Fix CI for Java 25 including upgrade of Gradle Develocity Maven extension (#24767)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0f2fe525aad8bba02c270f533155fda60dc07f42
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Sep 19 23:25:24 2025 +0300
[fix][ci] Fix CI for Java 25 including upgrade of Gradle Develocity Maven extension (#24767)
(cherry picked from commit 735c920532f2b7a8ecfcce78667a4a0dc238b28b)
---
.github/workflows/ci-owasp-dependency-check.yaml | 2 +-
.github/workflows/pulsar-ci-flaky.yaml | 2 +-
.github/workflows/pulsar-ci.yaml | 2 +-
.mvn/extensions.xml | 4 ++--
4 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/.github/workflows/ci-owasp-dependency-check.yaml b/.github/workflows/ci-owasp-dependency-check.yaml
index 4a2e12a0637..9f6c90d359d 100644
--- a/.github/workflows/ci-owasp-dependency-check.yaml
+++ b/.github/workflows/ci-owasp-dependency-check.yaml
@@ -24,7 +24,7 @@ on:
workflow_dispatch:
env:
- MAVEN_OPTS: -Xss1500k -Xmx2048m -XX:+UnlockDiagnosticVMOptions -XX:GCLockerRetryAllocationCount=100 -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000
+ MAVEN_OPTS: -Xss1500k -Xmx2048m -XX:+UnlockDiagnosticVMOptions -XX:+IgnoreUnrecognizedVMOptions -XX:GCLockerRetryAllocationCount=100 -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven. [...]
JDK_DISTRIBUTION: corretto
NIST_NVD_API_KEY: ${{ secrets.NIST_NVD_API_KEY }}
diff --git a/.github/workflows/pulsar-ci-flaky.yaml b/.github/workflows/pulsar-ci-flaky.yaml
index 6b6c43e0e1d..d3dfd2ebf4d 100644
--- a/.github/workflows/pulsar-ci-flaky.yaml
+++ b/.github/workflows/pulsar-ci-flaky.yaml
@@ -37,7 +37,7 @@ concurrency:
cancel-in-progress: true
env:
- MAVEN_OPTS: -Xss1500k -Xmx2048m -XX:+UnlockDiagnosticVMOptions -XX:GCLockerRetryAllocationCount=100 -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000
+ MAVEN_OPTS: -Xss1500k -Xmx2048m -XX:+UnlockDiagnosticVMOptions -XX:+IgnoreUnrecognizedVMOptions -XX:GCLockerRetryAllocationCount=100 -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven. [...]
# defines the retention period for the intermediate build artifacts needed for rerunning a failed build job
# it's possible to rerun individual failed jobs when the build artifacts are available
# if the artifacts have already been expired, the complete workflow can be rerun by closing and reopening the PR or by rebasing the PR
diff --git a/.github/workflows/pulsar-ci.yaml b/.github/workflows/pulsar-ci.yaml
index e0276c803e4..a2ac42c2428 100644
--- a/.github/workflows/pulsar-ci.yaml
+++ b/.github/workflows/pulsar-ci.yaml
@@ -37,7 +37,7 @@ concurrency:
cancel-in-progress: true
env:
- MAVEN_OPTS: -Xss1500k -Xmx2048m -XX:+UnlockDiagnosticVMOptions -XX:GCLockerRetryAllocationCount=100 -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000
+ MAVEN_OPTS: -Xss1500k -Xmx2048m -XX:+UnlockDiagnosticVMOptions -XX:+IgnoreUnrecognizedVMOptions -XX:GCLockerRetryAllocationCount=100 -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven. [...]
# defines the retention period for the intermediate build artifacts needed for rerunning a failed build job
# it's possible to rerun individual failed jobs when the build artifacts are available
# if the artifacts have already been expired, the complete workflow can be rerun by closing and reopening the PR or by rebasing the PR
diff --git a/.mvn/extensions.xml b/.mvn/extensions.xml
index b31376eb797..12e764a4097 100644
--- a/.mvn/extensions.xml
+++ b/.mvn/extensions.xml
@@ -24,11 +24,11 @@
<extension>
<groupId>com.gradle</groupId>
<artifactId>develocity-maven-extension</artifactId>
- <version>1.23.2</version>
+ <version>2.2</version>
</extension>
<extension>
<groupId>com.gradle</groupId>
<artifactId>common-custom-user-data-maven-extension</artifactId>
- <version>2.0.2</version>
+ <version>2.0.6</version>
</extension>
</extensions>
(pulsar) 06/06: [fix][misc] Fix compareTo contract violation for NamespaceBundleStats, TimeAverageMessageData and ResourceUnitRanking (#24772)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ae9be46ececa153379bdcedb7fda9f9a404e0d9a
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Sep 23 01:48:44 2025 +0300
[fix][misc] Fix compareTo contract violation for NamespaceBundleStats, TimeAverageMessageData and ResourceUnitRanking (#24772)
(cherry picked from commit 75045380de82d0f14f80dfa49e2946ef03b30d60)
---
.../loadbalance/impl/SimpleLoadManagerImpl.java | 2 +-
.../extensions/models/TopKBundlesTest.java | 25 +++++++
.../data/loadbalancer/NamespaceBundleStats.java | 50 ++++++-------
.../loadbalancer/NamespaceBundleStatsTest.java | 48 ++++++++++++
.../org/apache/pulsar/common/util/CompareUtil.java | 61 +++++++++++++++
.../data/loadbalancer/ResourceUnitRanking.java | 10 ++-
.../data/loadbalancer/TimeAverageMessageData.java | 38 +++++++++-
.../apache/pulsar/common/util/CompareUtilTest.java | 87 ++++++++++++++++++++++
.../loadbalancer/TimeAverageMessageDataTest.java | 44 +++++++++++
9 files changed, 333 insertions(+), 32 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
index 8bb27e52298..6ec87b98f3e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
@@ -820,7 +820,7 @@ public class SimpleLoadManagerImpl implements LoadManager, Consumer<Notification
minLoadPercentage = loadPercentage;
} else {
if ((unboundedRanks ? ranking.compareMessageRateTo(selectedRanking)
- : ranking.compareTo(selectedRanking)) < 0) {
+ : ranking.compareToOtherRanking(selectedRanking)) < 0) {
minLoadPercentage = loadPercentage;
selectedRU = candidate;
selectedRanking = ranking;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundlesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundlesTest.java
index 472d44df890..603c00d566c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundlesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundlesTest.java
@@ -271,4 +271,29 @@ public class TopKBundlesTest {
}
}
}
+
+ // Issue https://github.com/apache/pulsar/issues/24754
+ @Test
+ public void testPartitionSortCompareToContractViolationIssue() {
+ Random rnd = new Random(0);
+ ArrayList<NamespaceBundleStats> stats = new ArrayList<>();
+ for (int i = 0; i < 1000; ++i) {
+ NamespaceBundleStats s = new NamespaceBundleStats();
+ s.msgThroughputIn = 4 * 75000 * rnd.nextDouble(); // Just above threshold (1e5)
+ s.msgThroughputOut = 75000000 - (4 * (75000 * rnd.nextDouble()));
+ s.msgRateIn = 4 * 75 * rnd.nextDouble();
+ s.msgRateOut = 75000 - (4 * 75 * rnd.nextDouble());
+ s.topics = i;
+ s.consumerCount = i;
+ s.producerCount = 4 * rnd.nextInt(375);
+ s.cacheSize = 75000000 - (rnd.nextInt(4 * 75000));
+ stats.add(s);
+ }
+ List<Map.Entry<String, ? extends Comparable>> bundleEntries = new ArrayList<>();
+
+ for (NamespaceBundleStats s : stats) {
+ bundleEntries.add(Map.entry("bundle-" + s.msgThroughputIn, s));
+ }
+ TopKBundles.partitionSort(bundleEntries, 100);
+ }
}
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStats.java
index cd364990842..d3c894524bf 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStats.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStats.java
@@ -37,14 +37,14 @@ public class NamespaceBundleStats implements Comparable<NamespaceBundleStats>, S
public long topics;
public long cacheSize;
- // Consider the throughput equal if difference is less than 100 KB/s
- private static final double throughputDifferenceThreshold = 1e5;
- // Consider the msgRate equal if the difference is less than 100
- private static final double msgRateDifferenceThreshold = 100;
- // Consider the total topics/producers/consumers equal if the difference is less than 500
- private static final long topicConnectionDifferenceThreshold = 500;
- // Consider the cache size equal if the difference is less than 100 kb
- private static final long cacheSizeDifferenceThreshold = 100000;
+ // When comparing throughput, uses a resolution of 100 KB/s, effectively rounding values before comparison
+ private static final double throughputComparisonResolution = 1e5;
+ // When comparing message rate, uses a resolution of 100, effectively rounding values before comparison
+ private static final double msgRateComparisonResolution = 100;
+ // When comparing total topics/producers/consumers, uses a resolution/rounding of 500
+ private static final long topicConnectionComparisonResolution = 500;
+ // When comparing cache size, uses a resolution/rounding of 100kB
+ private static final long cacheSizeComparisonResolution = 100000;
public NamespaceBundleStats() {
reset();
@@ -89,39 +89,33 @@ public class NamespaceBundleStats implements Comparable<NamespaceBundleStats>, S
public int compareByMsgRate(NamespaceBundleStats other) {
double thisMsgRate = this.msgRateIn + this.msgRateOut;
double otherMsgRate = other.msgRateIn + other.msgRateOut;
- if (Math.abs(thisMsgRate - otherMsgRate) > msgRateDifferenceThreshold) {
- return Double.compare(thisMsgRate, otherMsgRate);
- }
- return 0;
+ return compareDoubleWithResolution(thisMsgRate, otherMsgRate, msgRateComparisonResolution);
+ }
+
+ private static int compareDoubleWithResolution(double v1, double v2, double resolution) {
+ return Long.compare(Math.round(v1 / resolution), Math.round(v2 / resolution));
+ }
+
+ private static int compareLongWithResolution(long v1, long v2, long resolution) {
+ return Long.compare(v1 / resolution, v2 / resolution);
}
public int compareByTopicConnections(NamespaceBundleStats other) {
long thisTopicsAndConnections = this.topics + this.consumerCount + this.producerCount;
long otherTopicsAndConnections = other.topics + other.consumerCount + other.producerCount;
- if (Math.abs(thisTopicsAndConnections - otherTopicsAndConnections) > topicConnectionDifferenceThreshold) {
- return Long.compare(thisTopicsAndConnections, otherTopicsAndConnections);
- }
- return 0;
+ return compareLongWithResolution(thisTopicsAndConnections, otherTopicsAndConnections,
+ topicConnectionComparisonResolution);
}
public int compareByCacheSize(NamespaceBundleStats other) {
- if (Math.abs(this.cacheSize - other.cacheSize) > cacheSizeDifferenceThreshold) {
- return Long.compare(this.cacheSize, other.cacheSize);
- }
- return 0;
+ return compareLongWithResolution(cacheSize, other.cacheSize, cacheSizeComparisonResolution);
}
public int compareByBandwidthIn(NamespaceBundleStats other) {
- if (Math.abs(this.msgThroughputIn - other.msgThroughputIn) > throughputDifferenceThreshold) {
- return Double.compare(this.msgThroughputIn, other.msgThroughputIn);
- }
- return 0;
+ return compareDoubleWithResolution(msgThroughputIn, other.msgThroughputIn, throughputComparisonResolution);
}
public int compareByBandwidthOut(NamespaceBundleStats other) {
- if (Math.abs(this.msgThroughputOut - other.msgThroughputOut) > throughputDifferenceThreshold) {
- return Double.compare(this.msgThroughputOut, other.msgThroughputOut);
- }
- return 0;
+ return compareDoubleWithResolution(msgThroughputOut, other.msgThroughputOut, throughputComparisonResolution);
}
}
diff --git a/pulsar-client-admin-api/src/test/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStatsTest.java b/pulsar-client-admin-api/src/test/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStatsTest.java
new file mode 100644
index 00000000000..7b8d120e629
--- /dev/null
+++ b/pulsar-client-admin-api/src/test/java/org/apache/pulsar/policies/data/loadbalancer/NamespaceBundleStatsTest.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.policies.data.loadbalancer;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import org.testng.annotations.Test;
+
+public class NamespaceBundleStatsTest {
+
+ @Test
+ public void testCompareToContract() {
+ Random rnd = new Random();
+ List<NamespaceBundleStats> stats = new ArrayList<>();
+ for (int i = 0; i < 1000; ++i) {
+ NamespaceBundleStats s = new NamespaceBundleStats();
+ s.msgThroughputIn = 4 * 75000 * rnd.nextDouble();
+ s.msgThroughputOut = 75000000 - (4 * (75000 * rnd.nextDouble()));
+ s.msgRateIn = 4 * 75 * rnd.nextDouble();
+ s.msgRateOut = 75000 - (4 * 75 * rnd.nextDouble());
+ s.topics = i;
+ s.consumerCount = i;
+ s.producerCount = 4 * rnd.nextInt(375);
+ s.cacheSize = 75000000 - (rnd.nextInt(4 * 75000));
+ stats.add(s);
+ }
+ // this would throw "java.lang.IllegalArgumentException: Comparison method violates its general contract!"
+ // if compareTo() is not implemented correctly.
+ stats.sort(NamespaceBundleStats::compareTo);
+ }
+}
\ No newline at end of file
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/CompareUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/CompareUtil.java
new file mode 100644
index 00000000000..b43c42ff1c2
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/CompareUtil.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import lombok.experimental.UtilityClass;
+
+/**
+ * Utility class for comparing values.
+ */
+@UtilityClass
+public class CompareUtil {
+
+ /**
+ * Compare two double values with a given resolution.
+ * @param v1 first value to compare
+ * @param v2 second value to compare
+ * @param resolution resolution to compare with
+ * @return -1 if v1 < v2, 0 if v1 == v2, 1 if v1 > v2
+ */
+ public static int compareDoubleWithResolution(double v1, double v2, double resolution) {
+ return Long.compare((long) (v1 / resolution), (long) (v2 / resolution));
+ }
+
+ /**
+ * Compare two long values with a given resolution.
+ * @param v1 first value to compare
+ * @param v2 second value to compare
+ * @param resolution resolution to compare with
+ * @return -1 if v1 < v2, 0 if v1 == v2, 1 if v1 > v2
+ */
+ public static int compareLongWithResolution(long v1, long v2, long resolution) {
+ return Long.compare(v1 / resolution, v2 / resolution);
+ }
+
+ /**
+ * Compare two int values with a given resolution.
+ * @param v1 first value to compare
+ * @param v2 second value to compare
+ * @param resolution resolution to compare with
+ * @return -1 if v1 < v2, 0 if v1 == v2, 1 if v1 > v2
+ */
+ public static int compareIntegerWithResolution(int v1, int v2, int resolution) {
+ return Integer.compare(v1 / resolution, v2 / resolution);
+ }
+}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/ResourceUnitRanking.java b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/ResourceUnitRanking.java
index 3a9d7de09e1..9b83be08205 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/ResourceUnitRanking.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/ResourceUnitRanking.java
@@ -26,7 +26,7 @@ import org.apache.pulsar.common.policies.data.ResourceQuota;
* The class containing information about system resources, allocated quota, and loaded bundles.
*/
@EqualsAndHashCode
-public class ResourceUnitRanking implements Comparable<ResourceUnitRanking> {
+public class ResourceUnitRanking {
private static final long KBITS_TO_BYTES = 1024 / 8;
private static final double PERCENTAGE_DIFFERENCE_THRESHOLD = 5.0;
@@ -129,7 +129,13 @@ public class ResourceUnitRanking implements Comparable<ResourceUnitRanking> {
}
- public int compareTo(ResourceUnitRanking other) {
+ /**
+ * Compares to another ranking. Please note that this cannot be used to sort the rankings since the results
+ * of this method don't satify the contract of {@link Comparable#compareTo(Object)}
+ * @param other other ranking to compare to
+ * @return negative if this is less than other, 0 if they are equal, positive if this is greater than other
+ */
+ public int compareToOtherRanking(ResourceUnitRanking other) {
if (Math.abs(this.estimatedLoadPercentage - other.estimatedLoadPercentage) > PERCENTAGE_DIFFERENCE_THRESHOLD) {
return Double.compare(this.estimatedLoadPercentage, other.estimatedLoadPercentage);
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/TimeAverageMessageData.java b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/TimeAverageMessageData.java
index 777a6684ce8..5264c5058fb 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/TimeAverageMessageData.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/policies/data/loadbalancer/TimeAverageMessageData.java
@@ -18,10 +18,14 @@
*/
package org.apache.pulsar.policies.data.loadbalancer;
+import static org.apache.pulsar.common.util.CompareUtil.compareDoubleWithResolution;
+import lombok.EqualsAndHashCode;
+
/**
* Data class comprising the average message data over a fixed period of time.
*/
-public class TimeAverageMessageData {
+@EqualsAndHashCode
+public class TimeAverageMessageData implements Comparable<TimeAverageMessageData> {
// The maximum number of samples this data will consider.
private int maxSamples;
@@ -41,6 +45,11 @@ public class TimeAverageMessageData {
// The average message rate out per second.
private double msgRateOut;
+ // When comparing throughput, uses a resolution of 100 KB/s, effectively rounding values before comparison
+ private static final double throughputComparisonResolution = 1e5;
+ // When comparing message rate, uses a resolution of 100, effectively rounding values before comparison
+ private static final double msgRateComparisonResolution = 100;
+
// For JSON only.
public TimeAverageMessageData() {
}
@@ -177,4 +186,31 @@ public class TimeAverageMessageData {
public double totalMsgThroughput() {
return msgThroughputIn + msgThroughputOut;
}
+
+ @Override
+ public int compareTo(TimeAverageMessageData other) {
+ int result = this.compareByBandwidthIn(other);
+
+ if (result == 0) {
+ result = this.compareByBandwidthOut(other);
+ }
+ if (result == 0) {
+ result = this.compareByMsgRate(other);
+ }
+ return result;
+ }
+
+ public int compareByMsgRate(TimeAverageMessageData other) {
+ double thisMsgRate = this.msgRateIn + this.msgRateOut;
+ double otherMsgRate = other.msgRateIn + other.msgRateOut;
+ return compareDoubleWithResolution(thisMsgRate, otherMsgRate, msgRateComparisonResolution);
+ }
+
+ public int compareByBandwidthIn(TimeAverageMessageData other) {
+ return compareDoubleWithResolution(msgThroughputIn, other.msgThroughputIn, throughputComparisonResolution);
+ }
+
+ public int compareByBandwidthOut(TimeAverageMessageData other) {
+ return compareDoubleWithResolution(msgThroughputOut, other.msgThroughputOut, throughputComparisonResolution);
+ }
}
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/CompareUtilTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/CompareUtilTest.java
new file mode 100644
index 00000000000..1c08ee4b2d0
--- /dev/null
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/CompareUtilTest.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import static org.testng.Assert.assertEquals;
+import org.testng.annotations.Test;
+
+public class CompareUtilTest {
+ @Test
+ public void testCompareDoubleWithResolution_bucketedComparison() {
+ // Same truncated bucket when dividing by resolution
+ assertEquals(CompareUtil.compareDoubleWithResolution(0.01, 0.49, 0.5), 0);
+ assertEquals(CompareUtil.compareDoubleWithResolution(0.51, 0.99, 0.5), 0);
+
+ // Different truncated buckets
+ assertEquals(CompareUtil.compareDoubleWithResolution(0.51, 0.49, 0.5), 1);
+ assertEquals(CompareUtil.compareDoubleWithResolution(0.49, 0.51, 0.5), -1);
+ assertEquals(CompareUtil.compareDoubleWithResolution(1.01, 0.49, 0.5), 1);
+
+ // Larger numbers
+ assertEquals(CompareUtil.compareDoubleWithResolution(19.99, 10.01, 10.0), 0);
+ assertEquals(CompareUtil.compareDoubleWithResolution(19.99, 20.01, 10.0), -1);
+ assertEquals(CompareUtil.compareDoubleWithResolution(10.00, 9.99, 10.0), 1);
+ }
+
+ @Test
+ public void testCompareLongWithResolution_exactComparison() {
+ // resolution = 1 -> behave like Long.compare
+ assertEquals(CompareUtil.compareLongWithResolution(1L, 2L, 1L), -1);
+ assertEquals(CompareUtil.compareLongWithResolution(2L, 1L, 1L), 1);
+ assertEquals(CompareUtil.compareLongWithResolution(2L, 2L, 1L), 0);
+ }
+
+ @Test
+ public void testCompareLongWithResolution_bucketedComparison() {
+ // Same bucket when divided by resolution
+ assertEquals(CompareUtil.compareLongWithResolution(8L, 9L, 10L), 0);
+ assertEquals(CompareUtil.compareLongWithResolution(10L, 19L, 10L), 0);
+
+ // Different buckets
+ assertEquals(CompareUtil.compareLongWithResolution(9L, 20L, 10L), -1);
+ assertEquals(CompareUtil.compareLongWithResolution(21L, 10L, 10L), 1);
+
+ // Larger resolution
+ assertEquals(CompareUtil.compareLongWithResolution(100L, 175L, 100L), 0);
+ assertEquals(CompareUtil.compareLongWithResolution(199L, 201L, 100L), -1);
+ }
+
+ @Test
+ public void testCompareIntegerWithResolution_exactComparison() {
+ // resolution = 1 -> behave like Integer.compare
+ assertEquals(CompareUtil.compareIntegerWithResolution(1, 2, 1), -1);
+ assertEquals(CompareUtil.compareIntegerWithResolution(2, 1, 1), 1);
+ assertEquals(CompareUtil.compareIntegerWithResolution(2, 2, 1), 0);
+ }
+
+ @Test
+ public void testCompareIntegerWithResolution_bucketedComparison() {
+ // Same bucket
+ assertEquals(CompareUtil.compareIntegerWithResolution(3, 4, 5), 0);
+ assertEquals(CompareUtil.compareIntegerWithResolution(5, 9, 5), 0);
+
+ // Different buckets
+ assertEquals(CompareUtil.compareIntegerWithResolution(4, 10, 5), -1);
+ assertEquals(CompareUtil.compareIntegerWithResolution(11, 5, 5), 1);
+
+ // Larger resolution
+ assertEquals(CompareUtil.compareIntegerWithResolution(51, 75, 50), 0);
+ assertEquals(CompareUtil.compareIntegerWithResolution(49, 101, 50), -1);
+ }
+}
\ No newline at end of file
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/policies/data/loadbalancer/TimeAverageMessageDataTest.java b/pulsar-common/src/test/java/org/apache/pulsar/policies/data/loadbalancer/TimeAverageMessageDataTest.java
new file mode 100644
index 00000000000..ff7f3083014
--- /dev/null
+++ b/pulsar-common/src/test/java/org/apache/pulsar/policies/data/loadbalancer/TimeAverageMessageDataTest.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.policies.data.loadbalancer;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import org.testng.annotations.Test;
+
+public class TimeAverageMessageDataTest {
+ @Test
+ public void testCompareToContract() {
+ Random rnd = new Random();
+ List<TimeAverageMessageData> list = new ArrayList<>();
+ for (int i = 0; i < 1000; ++i) {
+ TimeAverageMessageData data = new TimeAverageMessageData(1);
+ double msgThroughputIn = 4 * 75000 * rnd.nextDouble();
+ double msgThroughputOut = 75000000 - (4 * (75000 * rnd.nextDouble()));
+ double msgRateIn = 4 * 75 * rnd.nextDouble();
+ double msgRateOut = 75000 - (4 * 75 * rnd.nextDouble());
+ data.update(msgThroughputIn, msgThroughputOut, msgRateIn, msgRateOut);
+ list.add(data);
+ }
+ // this would throw "java.lang.IllegalArgumentException: Comparison method violates its general contract!"
+ // if compareTo() is not implemented correctly.
+ list.sort(TimeAverageMessageData::compareTo);
+ }
+}
\ No newline at end of file
(pulsar) 02/06: [improve][broker] If there is a deadlock in the service, the probe should return a failure because the service may be unavailable (#23634)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit cb223f7e4540fee31bc230fab23e26d4e6d16103
Author: yangyijun <10...@qq.com>
AuthorDate: Fri Sep 19 18:53:34 2025 +0800
[improve][broker] If there is a deadlock in the service, the probe should return a failure because the service may be unavailable (#23634)
Co-authored-by: Lari Hotari <lh...@apache.org>
Co-authored-by: Lari Hotari <lh...@users.noreply.github.com>
(cherry picked from commit d833b8bef21cb9e85f0f313eb9d49c7ca550fbbd)
---
pulsar-broker-common/pom.xml | 6 +
.../pulsar/common/configuration/VipStatus.java | 99 +++++++++++--
.../pulsar/common/configuration/VipStatusTest.java | 161 +++++++++++++++++++++
3 files changed, 254 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker-common/pom.xml b/pulsar-broker-common/pom.xml
index ff8c7c033af..32f1b518e99 100644
--- a/pulsar-broker-common/pom.xml
+++ b/pulsar-broker-common/pom.xml
@@ -83,6 +83,12 @@
<artifactId>awaitility</artifactId>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.glassfish.jersey.core</groupId>
+ <artifactId>jersey-server</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/common/configuration/VipStatus.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/common/configuration/VipStatus.java
index 5e6a31b323f..aa4ec1109a6 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/common/configuration/VipStatus.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/common/configuration/VipStatus.java
@@ -18,8 +18,15 @@
*/
package org.apache.pulsar.common.configuration;
+import com.google.common.annotations.VisibleForTesting;
import java.io.File;
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
+import java.time.Clock;
+import java.util.Arrays;
import java.util.function.Supplier;
+import java.util.stream.Collectors;
import javax.servlet.ServletContext;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
@@ -27,6 +34,7 @@ import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.Response.Status;
import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.common.util.ThreadDumpUtil;
/**
* Web resource used by the VIP service to check to availability of the service instance.
@@ -38,25 +46,92 @@ public class VipStatus {
public static final String ATTRIBUTE_STATUS_FILE_PATH = "statusFilePath";
public static final String ATTRIBUTE_IS_READY_PROBE = "isReadyProbe";
+ // log a full thread dump when a deadlock is detected in status check once every 10 minutes
+ // to prevent excessive logging
+ private static final long LOG_THREADDUMP_INTERVAL_WHEN_DEADLOCK_DETECTED = 600000L;
+ // Rate limit status checks to every 500ms to prevent DoS
+ private static final long CHECK_STATUS_INTERVAL = 500L;
+
+ private static volatile long lastCheckStatusTimestamp;
+ private static volatile long lastPrintThreadDumpTimestamp;
+ private static volatile boolean lastCheckStatusResult;
+
+ private long printThreadDumpIntervalMs;
+ private Clock clock;
+
@Context
protected ServletContext servletContext;
+ public VipStatus() {
+ this.clock = Clock.systemUTC();
+ this.printThreadDumpIntervalMs = LOG_THREADDUMP_INTERVAL_WHEN_DEADLOCK_DETECTED;
+ }
+
+ @VisibleForTesting
+ public VipStatus(ServletContext servletContext, long printThreadDumpIntervalMs) {
+ this.servletContext = servletContext;
+ this.printThreadDumpIntervalMs = printThreadDumpIntervalMs;
+ this.clock = Clock.systemUTC();
+ }
+
+ @VisibleForTesting
+ static void reset() {
+ lastCheckStatusTimestamp = 0L;
+ lastPrintThreadDumpTimestamp = 0L;
+ lastCheckStatusResult = false;
+ }
+
@GET
public String checkStatus() {
- String statusFilePath = (String) servletContext.getAttribute(ATTRIBUTE_STATUS_FILE_PATH);
- @SuppressWarnings("unchecked")
- Supplier<Boolean> isReadyProbe = (Supplier<Boolean>) servletContext.getAttribute(ATTRIBUTE_IS_READY_PROBE);
+ // Locking classes to avoid deadlock detection in multi-thread concurrent requests.
+ synchronized (VipStatus.class) {
+ if (clock.millis() - lastCheckStatusTimestamp < CHECK_STATUS_INTERVAL) {
+ if (lastCheckStatusResult) {
+ return "OK";
+ } else {
+ throw new WebApplicationException(Status.SERVICE_UNAVAILABLE);
+ }
+ }
+ lastCheckStatusTimestamp = clock.millis();
- boolean isReady = isReadyProbe != null ? isReadyProbe.get() : true;
+ String statusFilePath = (String) servletContext.getAttribute(ATTRIBUTE_STATUS_FILE_PATH);
+ @SuppressWarnings("unchecked")
+ Supplier<Boolean> isReadyProbe = (Supplier<Boolean>) servletContext.getAttribute(ATTRIBUTE_IS_READY_PROBE);
+ boolean isReady = isReadyProbe != null ? isReadyProbe.get() : true;
- if (statusFilePath != null) {
- File statusFile = new File(statusFilePath);
- if (isReady && statusFile.exists() && statusFile.isFile()) {
- return "OK";
+ if (statusFilePath != null) {
+ File statusFile = new File(statusFilePath);
+ if (isReady && statusFile.exists() && statusFile.isFile()) {
+ // check deadlock
+ ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();
+ long[] threadIds = threadBean.findDeadlockedThreads();
+ if (threadIds != null && threadIds.length > 0) {
+ ThreadInfo[] threadInfos = threadBean.getThreadInfo(threadIds, false,
+ false);
+ String threadNames = Arrays.stream(threadInfos)
+ .map(threadInfo -> threadInfo.getThreadName()
+ + "(tid=" + threadInfo.getThreadId() + ")")
+ .collect(Collectors.joining(", "));
+ if (clock.millis() - lastPrintThreadDumpTimestamp > printThreadDumpIntervalMs) {
+ String diagnosticResult = ThreadDumpUtil.buildThreadDiagnosticString();
+ log.error("Deadlocked threads detected. {}. Service may be unavailable, "
+ + "thread stack details are as follows:\n{}", threadNames, diagnosticResult);
+ lastPrintThreadDumpTimestamp = clock.millis();
+ } else {
+ log.error("Deadlocked threads detected. {}", threadNames);
+ }
+ lastCheckStatusResult = false;
+ throw new WebApplicationException(Status.SERVICE_UNAVAILABLE);
+ } else {
+ lastCheckStatusResult = true;
+ return "OK";
+ }
+ }
}
+ lastCheckStatusResult = false;
+ log.warn("Status file '{}' doesn't exist or ready probe value ({}) isn't true. The service is not ready",
+ statusFilePath, isReady);
+ throw new WebApplicationException(Status.NOT_FOUND);
}
- log.warn("Failed to access \"status.html\". The service is not ready");
- throw new WebApplicationException(Status.NOT_FOUND);
}
-
-}
+}
\ No newline at end of file
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/VipStatusTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/VipStatusTest.java
new file mode 100644
index 00000000000..36542d237b5
--- /dev/null
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/VipStatusTest.java
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.configuration;
+
+import static org.testng.Assert.assertEquals;
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Phaser;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Supplier;
+import javax.servlet.ServletContext;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.Response;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.assertj.core.util.Files;
+import org.mockito.Mockito;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class VipStatusTest {
+
+ public static final String ATTRIBUTE_STATUS_FILE_PATH = "statusFilePath";
+ public static final String ATTRIBUTE_IS_READY_PROBE = "isReadyProbe";
+ private static final long LOG_THREADDUMP_INTERVAL_WHEN_DEADLOCK_DETECTED = 10000L;
+ // Rate limit status checks to every 500ms to prevent DoS
+ private static final long CHECK_STATUS_INTERVAL = 500L;
+
+ private ServletContext mockServletContext;
+ private VipStatus vipStatus;
+ private File file;
+
+ @BeforeMethod
+ public void setup() throws IOException {
+ file = Files.newTemporaryFile();
+ Supplier<Boolean> isReadyProbe = () -> true;
+ mockServletContext = Mockito.mock(ServletContext.class);
+ Mockito.when(mockServletContext.getAttribute(ATTRIBUTE_STATUS_FILE_PATH)).thenReturn(file.getAbsolutePath());
+ Mockito.when(mockServletContext.getAttribute(ATTRIBUTE_IS_READY_PROBE)).thenReturn(isReadyProbe);
+ vipStatus = new VipStatus(mockServletContext, LOG_THREADDUMP_INTERVAL_WHEN_DEADLOCK_DETECTED);
+ VipStatus.reset();
+ }
+
+ @Test
+ public void testVipStatusCheckStatus() {
+ // No deadlocks
+ testVipStatusCheckStatusWithoutDeadlock();
+ // There is a deadlock
+ testVipStatusCheckStatusWithDeadlock();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void release() throws IOException {
+ if (file != null) {
+ file.delete();
+ file = null;
+ }
+ }
+
+ @Test
+ public void testVipStatusCheckStatusWithoutDeadlock() {
+ assertEquals(vipStatus.checkStatus(), "OK");
+ }
+
+ @Test
+ public void testVipStatusCheckStatusWithDeadlock() {
+ MockDeadlock mockDeadlock = new MockDeadlock();
+ boolean asExpected = true;
+ try {
+ mockDeadlock.startDeadlock();
+ vipStatus.checkStatus();
+ asExpected = false;
+ System.out.println("Simulated deadlock, no deadlock detected, not as expected.");
+ } catch (Exception wae) {
+ System.out.println("Simulated deadlock and detected it, as expected.");
+ } finally {
+ mockDeadlock.close();
+ }
+
+ if (!asExpected) {
+ throw new WebApplicationException(Response.Status.SERVICE_UNAVAILABLE);
+ }
+ }
+
+ static class MockDeadlock implements Closeable {
+ private ExecutorService executorService = Executors.newCachedThreadPool();
+ private ReentrantLock lockA = new ReentrantLock();
+ private ReentrantLock lockB = new ReentrantLock();
+ private Phaser phaser = new Phaser(2);
+
+ @SneakyThrows
+ public void startDeadlock() {
+ executorService.execute(new TaskOne());
+ executorService.execute(new TaskTwo());
+ Thread.sleep(CHECK_STATUS_INTERVAL);
+ }
+
+ @Override
+ public void close() {
+ executorService.shutdownNow();
+ }
+
+ private class TaskOne implements Runnable {
+ @Override
+ public void run() {
+ try {
+ lockA.lock();
+ System.out.println("ThreadOne acquired lockA");
+ phaser.arriveAndAwaitAdvance();
+ while (!lockB.tryLock(1, TimeUnit.SECONDS)) {
+ System.out.println("ThreadOne acquired lockB");
+ }
+ } catch (InterruptedException e) {
+ //e.printStackTrace();
+ } finally {
+ lockA.unlock();
+ }
+ }
+ }
+
+ private class TaskTwo implements Runnable {
+ @Override
+ public void run() {
+ try {
+ lockB.lock();
+ System.out.println("ThreadOne acquired lockB");
+ phaser.arriveAndAwaitAdvance();
+ while (!lockA.tryLock(1, TimeUnit.SECONDS)) {
+ System.out.println("ThreadOne acquired lockA");
+ }
+ } catch (InterruptedException e) {
+ //e.printStackTrace();
+ } finally {
+ lockB.unlock();
+ }
+ }
+ }
+ }
+}