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/22 19:54:03 UTC
(pulsar) branch branch-4.1 updated (35afad62604 -> 2e7ae813ef7)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 35afad62604 [improve][build] Upgrade SpotBugs to a version that supports JDK25 (#24768)
new c3f6f591ed6 [fix][broker] First entry will be skipped if opening NonDurableCursor while trimmed ledger is adding first entry. (#24738)
new d38a22d3db5 [improve][broker] If there is a deadlock in the service, the probe should return a failure because the service may be unavailable (#23634)
new 9d3e7d044fc [improve][build] Upgrade Lombok to 1.18.42 to fully support JDK25 (#24763)
new 67425640b74 [fix][client] Exclude io.prometheus:simpleclient_caffeine from client-side dependencies (#24761)
new 2c920b66061 [improve][build] Upgrade Mockito, AssertJ and ByteBuddy to fully support JDK25 (#24764)
new 2e7ae813ef7 [fix][test] Flaky-test: BrokerServiceTest.testShutDownWithMaxConcurrentUnload (#24769)
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:
bouncy-castle/bc/pom.xml | 6 +
bouncy-castle/bcfips/pom.xml | 6 +
buildtools/pom.xml | 4 +-
distribution/server/src/assemble/LICENSE.bin.txt | 2 +-
distribution/shell/src/assemble/LICENSE.bin.txt | 2 -
.../mledger/impl/NonDurableCursorImpl.java | 5 +-
.../mledger/impl/NonDurableCursorTest.java | 58 ++++++++
pom.xml | 44 ++----
pulsar-broker-common/pom.xml | 6 +
.../pulsar/common/configuration/VipStatus.java | 99 +++++++++++--
.../pulsar/common/configuration/VipStatusTest.java | 161 +++++++++++++++++++++
.../pulsar/broker/service/BrokerServiceTest.java | 3 +-
pulsar-client-messagecrypto-bc/pom.xml | 6 +
pulsar-client/pom.xml | 6 +
pulsar-functions/instance/pom.xml | 9 +-
pulsar-functions/utils/pom.xml | 6 +
pulsar-io/debezium/core/pom.xml | 6 +
pulsar-io/kafka-connect-adaptor/pom.xml | 6 +
pulsar-websocket/pom.xml | 6 +
19 files changed, 384 insertions(+), 57 deletions(-)
create mode 100644 pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/VipStatusTest.java
(pulsar) 04/06: [fix][client] Exclude io.prometheus:simpleclient_caffeine from client-side dependencies (#24761)
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.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 67425640b74b18c7ab94f4eea485f7a23f01afa7
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Sep 19 18:53:24 2025 +0300
[fix][client] Exclude io.prometheus:simpleclient_caffeine from client-side dependencies (#24761)
(cherry picked from commit b11f05feb7fcb08c8ad2c5f4ad301ef14d518490)
---
bouncy-castle/bc/pom.xml | 6 +++++
bouncy-castle/bcfips/pom.xml | 6 +++++
distribution/shell/src/assemble/LICENSE.bin.txt | 2 --
pom.xml | 34 +++++--------------------
pulsar-client-messagecrypto-bc/pom.xml | 6 +++++
pulsar-client/pom.xml | 6 +++++
pulsar-functions/instance/pom.xml | 9 ++++---
pulsar-functions/utils/pom.xml | 6 +++++
pulsar-io/debezium/core/pom.xml | 6 +++++
pulsar-io/kafka-connect-adaptor/pom.xml | 6 +++++
pulsar-websocket/pom.xml | 6 +++++
11 files changed, 59 insertions(+), 34 deletions(-)
diff --git a/bouncy-castle/bc/pom.xml b/bouncy-castle/bc/pom.xml
index c5b24f836f3..c3e2fee7309 100644
--- a/bouncy-castle/bc/pom.xml
+++ b/bouncy-castle/bc/pom.xml
@@ -37,6 +37,12 @@
<artifactId>pulsar-common</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient_caffeine</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
diff --git a/bouncy-castle/bcfips/pom.xml b/bouncy-castle/bcfips/pom.xml
index c3715906921..14203adae91 100644
--- a/bouncy-castle/bcfips/pom.xml
+++ b/bouncy-castle/bcfips/pom.xml
@@ -37,6 +37,12 @@
<artifactId>pulsar-common</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient_caffeine</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt
index aeff29513dd..5dd6d6edd7f 100644
--- a/distribution/shell/src/assemble/LICENSE.bin.txt
+++ b/distribution/shell/src/assemble/LICENSE.bin.txt
@@ -324,8 +324,6 @@ The Apache Software License, Version 2.0
- jackson-datatype-jdk8-2.17.2.jar
- jackson-datatype-jsr310-2.17.2.jar
- jackson-module-parameter-names-2.17.2.jar
- * Caffeine -- caffeine-3.2.1.jar
- * simpleclient_caffeine-0.16.0.jar
* Conscrypt -- conscrypt-openjdk-uber-2.5.2.jar
* Gson
- gson-2.8.9.jar
diff --git a/pom.xml b/pom.xml
index e2eb1460dff..046883223a1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1102,38 +1102,16 @@ flexible messaging model and an intuitive client API.</description>
<dependency>
<groupId>io.prometheus</groupId>
- <artifactId>simpleclient</artifactId>
- <version>${prometheus.version}</version>
- </dependency>
-
- <dependency>
- <groupId>io.prometheus</groupId>
- <artifactId>simpleclient_hotspot</artifactId>
- <version>${prometheus.version}</version>
- </dependency>
-
- <dependency>
- <groupId>io.prometheus</groupId>
- <artifactId>simpleclient_log4j2</artifactId>
- <version>${prometheus.version}</version>
- </dependency>
-
- <dependency>
- <groupId>io.prometheus</groupId>
- <artifactId>simpleclient_servlet</artifactId>
- <version>${prometheus.version}</version>
- </dependency>
-
- <dependency>
- <groupId>io.prometheus</groupId>
- <artifactId>simpleclient_jetty</artifactId>
+ <artifactId>simpleclient_bom</artifactId>
<version>${prometheus.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
</dependency>
<dependency>
- <groupId>io.prometheus</groupId>
- <artifactId>simpleclient_caffeine</artifactId>
- <version>${prometheus.version}</version>
+ <groupId>io.prometheus.jmx</groupId>
+ <artifactId>collector</artifactId>
+ <version>${prometheus-jmx.version}</version>
</dependency>
<dependency>
diff --git a/pulsar-client-messagecrypto-bc/pom.xml b/pulsar-client-messagecrypto-bc/pom.xml
index 4cedca6a81a..72f85f60978 100644
--- a/pulsar-client-messagecrypto-bc/pom.xml
+++ b/pulsar-client-messagecrypto-bc/pom.xml
@@ -39,6 +39,12 @@
<artifactId>pulsar-common</artifactId>
<version>${project.parent.version}</version>
<scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient_caffeine</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
diff --git a/pulsar-client/pom.xml b/pulsar-client/pom.xml
index 4e37d694a42..4f035dfa50d 100644
--- a/pulsar-client/pom.xml
+++ b/pulsar-client/pom.xml
@@ -42,6 +42,12 @@
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-common</artifactId>
<version>${project.parent.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient_caffeine</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
diff --git a/pulsar-functions/instance/pom.xml b/pulsar-functions/instance/pom.xml
index 7065a4d7743..43ee5c595af 100644
--- a/pulsar-functions/instance/pom.xml
+++ b/pulsar-functions/instance/pom.xml
@@ -175,27 +175,28 @@
<dependency>
<groupId>io.prometheus</groupId>
<artifactId>simpleclient</artifactId>
- <version>${prometheus.version}</version>
</dependency>
<!-- Hotspot JVM metrics-->
<dependency>
<groupId>io.prometheus</groupId>
<artifactId>simpleclient_hotspot</artifactId>
- <version>${prometheus.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient_caffeine</artifactId>
</dependency>
<!-- Exposition HTTPServer-->
<dependency>
<groupId>io.prometheus</groupId>
<artifactId>simpleclient_httpserver</artifactId>
- <version>${prometheus.version}</version>
</dependency>
<dependency>
<groupId>io.prometheus.jmx</groupId>
<artifactId>collector</artifactId>
- <version>${prometheus-jmx.version}</version>
</dependency>
<dependency>
diff --git a/pulsar-functions/utils/pom.xml b/pulsar-functions/utils/pom.xml
index cb8b1753efb..2206e3be039 100644
--- a/pulsar-functions/utils/pom.xml
+++ b/pulsar-functions/utils/pom.xml
@@ -36,6 +36,12 @@
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-common</artifactId>
<version>${project.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient_caffeine</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
diff --git a/pulsar-io/debezium/core/pom.xml b/pulsar-io/debezium/core/pom.xml
index 13821980f11..3154f4b1d5e 100644
--- a/pulsar-io/debezium/core/pom.xml
+++ b/pulsar-io/debezium/core/pom.xml
@@ -54,6 +54,12 @@
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-common</artifactId>
<version>${project.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient_caffeine</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
diff --git a/pulsar-io/kafka-connect-adaptor/pom.xml b/pulsar-io/kafka-connect-adaptor/pom.xml
index 4d2f7016d69..9931d1aa48c 100644
--- a/pulsar-io/kafka-connect-adaptor/pom.xml
+++ b/pulsar-io/kafka-connect-adaptor/pom.xml
@@ -44,6 +44,12 @@
<artifactId>pulsar-common</artifactId>
<version>${project.version}</version>
<scope>compile</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>io.prometheus</groupId>
+ <artifactId>simpleclient_caffeine</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
diff --git a/pulsar-websocket/pom.xml b/pulsar-websocket/pom.xml
index d3e4d876c11..b17b1b0fbd0 100644
--- a/pulsar-websocket/pom.xml
+++ b/pulsar-websocket/pom.xml
@@ -38,6 +38,12 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-client-original</artifactId>
(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-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c3f6f591ed68d3645c0f8054969dca098488d308
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 | 58 ++++++++++++++++++++++
2 files changed, 61 insertions(+), 2 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 d1fc54a9d6d..69c30fdaca9 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
@@ -45,11 +45,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<Position, 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 a7fe289f56c..aced11d3b3d 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,6 +19,7 @@
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;
@@ -50,6 +51,11 @@ import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
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;
@@ -103,6 +109,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
+ Position startCursorPosition = PositionFactory.create(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)
+ );
+ Position 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) 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-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d38a22d3db50acf79e310b59b8342859235e0871
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 8563419b804..2ebc0b8e473 100644
--- a/pulsar-broker-common/pom.xml
+++ b/pulsar-broker-common/pom.xml
@@ -94,6 +94,12 @@
<artifactId>rest-assured</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();
+ }
+ }
+ }
+ }
+}
(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-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9d3e7d044fc1735cc9a6d092e572c71ab6aadb80
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 7c421792880..e2eb1460dff 100644
--- a/pom.xml
+++ b/pom.xml
@@ -276,7 +276,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) 06/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-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2e7ae813ef799a26bb8b57232155e851aabee996
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 c73acfe9ee8..49f5d7c5c36 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
@@ -179,10 +179,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) 05/06: [improve][build] Upgrade Mockito, AssertJ and ByteBuddy to fully support JDK25 (#24764)
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.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2c920b66061822a0f7e1e09886885b4bdd341cd4
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Sep 19 19:41:04 2025 +0300
[improve][build] Upgrade Mockito, AssertJ and ByteBuddy to fully support JDK25 (#24764)
(cherry picked from commit e94755d58b563cde3a06d9daae50e9e0ca86334c)
---
buildtools/pom.xml | 4 ++--
distribution/server/src/assemble/LICENSE.bin.txt | 2 +-
pom.xml | 8 ++++----
3 files changed, 7 insertions(+), 7 deletions(-)
diff --git a/buildtools/pom.xml b/buildtools/pom.xml
index 075f947a98d..e115d0cd40c 100644
--- a/buildtools/pom.xml
+++ b/buildtools/pom.xml
@@ -60,8 +60,8 @@
<guava.version>33.4.8-jre</guava.version>
<ant.version>1.10.12</ant.version>
<snakeyaml.version>2.0</snakeyaml.version>
- <mockito.version>5.17.0</mockito.version>
- <byte-buddy.version>1.15.11</byte-buddy.version>
+ <mockito.version>5.19.0</mockito.version>
+ <byte-buddy.version>1.17.7</byte-buddy.version>
<wagon-ssh-external.version>3.5.3</wagon-ssh-external.version>
<!-- required for running tests on JDK11+ -->
<test.additional.args>
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index ec0d444daa1..8d9c88e64e7 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -460,7 +460,7 @@ The Apache Software License, Version 2.0
- net.jodah-typetools-0.5.0.jar
- dev.failsafe-failsafe-3.3.2.jar
* Byte Buddy
- - net.bytebuddy-byte-buddy-1.15.11.jar
+ - net.bytebuddy-byte-buddy-1.17.7.jar
* zt-zip
- org.zeroturnaround-zt-zip-1.17.jar
* Apache Avro
diff --git a/pom.xml b/pom.xml
index 046883223a1..074feef8663 100644
--- a/pom.xml
+++ b/pom.xml
@@ -216,8 +216,8 @@ flexible messaging model and an intuitive client API.</description>
<docker-maven.version>0.45.1</docker-maven.version>
<docker.verbose>true</docker.verbose>
<typetools.version>0.5.0</typetools.version>
- <!-- match the byte-buddy version used by the mockito-core pom -->
- <byte-buddy.version>1.15.11</byte-buddy.version>
+ <!-- match the byte-buddy version (or newer) used by the mockito-core pom -->
+ <byte-buddy.version>1.17.7</byte-buddy.version>
<zt-zip.version>1.17</zt-zip.version>
<protobuf3.version>3.25.5</protobuf3.version>
<protoc3.version>${protobuf3.version}</protoc3.version>
@@ -275,7 +275,7 @@ flexible messaging model and an intuitive client API.</description>
<jline3.version>3.21.0</jline3.version>
<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>
+ <assertj-core.version>3.27.5</assertj-core.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>
@@ -324,7 +324,7 @@ flexible messaging model and an intuitive client API.</description>
<restassured.version>5.4.0</restassured.version>
<kerby.version>1.1.1</kerby.version>
<testng.version>7.7.1</testng.version>
- <mockito.version>5.17.0</mockito.version>
+ <mockito.version>5.19.0</mockito.version>
<javassist.version>3.25.0-GA</javassist.version>
<skyscreamer.version>1.5.0</skyscreamer.version>
<objenesis.version>3.3</objenesis.version> <!-- should match mockito-core objenesis version -->