You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by he...@apache.org on 2025/04/24 21:00:35 UTC
(pulsar) branch master updated: [fix][broker] set broker healthcheck timeout correctly (#24198)
This is an automated email from the ASF dual-hosted git repository.
heesung pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new dd58ec9dd15 [fix][broker] set broker healthcheck timeout correctly (#24198)
dd58ec9dd15 is described below
commit dd58ec9dd1510a1003fcc70d7601a33d1ffd32b3
Author: Heesung Sohn <10...@users.noreply.github.com>
AuthorDate: Thu Apr 24 14:00:29 2025 -0700
[fix][broker] set broker healthcheck timeout correctly (#24198)
---
.../loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
index f20fb70cffd..3e2a9c7468d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -1278,6 +1278,8 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
if (!cleanupJobs.isEmpty() && cleanupJobs.containsKey(broker)) {
healthCheckBrokerAsync(broker)
+ .orTimeout(MAX_BROKER_HEALTH_CHECK_DELAY_IN_MILLIS * (MAX_BROKER_HEALTH_CHECK_RETRY + 1)
+ , MILLISECONDS)
.thenAccept(__ -> {
CompletableFuture<Void> future = cleanupJobs.remove(broker);
if (future != null) {
@@ -1528,11 +1530,13 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
return;
}
+
// if not gracefully, verify the broker is inactive by health-check.
if (!gracefully) {
try {
healthCheckBrokerAsync(broker).get(
- pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds(), SECONDS);
+ MAX_BROKER_HEALTH_CHECK_DELAY_IN_MILLIS * (MAX_BROKER_HEALTH_CHECK_RETRY + 1)
+ , MILLISECONDS);
log.warn("Found that the broker to clean is healthy. Skip the broker:{}'s orphan bundle cleanup",
broker);
return;