Healthcheck v2 failed (#13525)
* fix healthcheck v2
* fix failed testHealthCheckupV2 because of error web port
Co-authored-by: gavingaozhangmin <gavingaozhangmin@didiglobal.com>
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 666673e..76d2d6c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -265,6 +265,7 @@
private volatile CompletableFuture<Void> closeFuture;
// key is listener name , value is pulsar address and pulsar ssl address
private Map<String, AdvertisedListener> advertisedListeners;
+ private NamespaceName heartbeatNamespaceV2;
public PulsarService(ServiceConfiguration config) {
this(config, Optional.empty(), (exitCode) -> {
@@ -671,6 +672,7 @@
this.addWebServerHandlers(webService, metricsServlet, this.config);
this.webService.start();
+ heartbeatNamespaceV2 = NamespaceService.getHeartbeatNamespaceV2(this.advertisedAddress, this.config);
// Refresh addresses and update configuration, since the port might have been dynamically assigned
if (config.getBrokerServicePort().equals(Optional.of(0))) {
@@ -684,6 +686,7 @@
this.brokerServiceUrl = brokerUrl(config);
this.brokerServiceUrlTls = brokerUrlTls(config);
+
if (null != this.webSocketService) {
ClusterDataImpl clusterData = ClusterDataImpl.builder()
.serviceUrl(webServiceAddress)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
index d2fccad..2a97fe6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -68,6 +68,7 @@
import org.apache.pulsar.common.api.proto.CommandSubscribe.InitialPosition;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.KeySharedMeta;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats.CursorStats;
@@ -523,6 +524,10 @@
if (!name.isGlobal()) {
return CompletableFuture.completedFuture(null);
}
+ NamespaceName heartbeatNamespace = brokerService.pulsar().getHeartbeatNamespaceV2();
+ if (name.getNamespaceObject().equals(heartbeatNamespace)) {
+ return CompletableFuture.completedFuture(null);
+ }
if (log.isDebugEnabled()) {
log.debug("[{}] Checking replication status", name);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index a3d2ff3..5c6f99a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -125,6 +125,7 @@
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.api.proto.TxnAction;
import org.apache.pulsar.common.events.EventsTopicNames;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType;
@@ -1373,6 +1374,10 @@
if (!name.isGlobal()) {
return CompletableFuture.completedFuture(null);
}
+ NamespaceName heartbeatNamespace = brokerService.pulsar().getHeartbeatNamespaceV2();
+ if (name.getNamespaceObject().equals(heartbeatNamespace)) {
+ return CompletableFuture.completedFuture(null);
+ }
if (log.isDebugEnabled()) {
log.debug("[{}] Checking replication status", name);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
index e3ef2b3..b243a99 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
@@ -738,6 +738,11 @@
if (!namespace.isGlobal()) {
return CompletableFuture.completedFuture(null);
}
+ NamespaceName heartbeatNamespace = pulsarService.getHeartbeatNamespaceV2();
+ if (namespace.equals(heartbeatNamespace)) {
+ return CompletableFuture.completedFuture(null);
+ }
+
final CompletableFuture<ClusterDataImpl> validationFuture = new CompletableFuture<>();
final String localCluster = pulsarService.getConfiguration().getClusterName();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiHealthCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiHealthCheckTest.java
index e0c887f..7c568da 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiHealthCheckTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiHealthCheckTest.java
@@ -21,7 +21,6 @@
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
-import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.common.naming.TopicVersion;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
@@ -29,8 +28,6 @@
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import java.net.URL;
-
@Test(groups = "broker")
@Slf4j
public class AdminApiHealthCheckTest extends MockedPulsarServiceBaseTest {
@@ -66,18 +63,8 @@
admin.brokers().healthcheck(TopicVersion.V1);
}
- @Test(expectedExceptions = PulsarAdminException.class)
- public void testHealthCheckupV2Error() throws Exception {
- admin.brokers().healthcheck(TopicVersion.V2);
- }
-
@Test
public void testHealthCheckupV2() throws Exception {
- final URL pulsarWebAddress = new URL(pulsar.getWebServiceAddress());
- final String targetNameSpace = "pulsar/" +
- pulsarWebAddress.getHost() + ":" + pulsarWebAddress.getPort();
- log.info("Target namespace for broker admin healthcheck V2 endpoint is {}", targetNameSpace);
- admin.namespaces().createNamespace(targetNameSpace);
admin.brokers().healthcheck(TopicVersion.V2);
}
}
