You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by aa...@apache.org on 2021/07/08 21:15:05 UTC
[pulsar] 01/01: Support new topic format for broker admin
healthcheck endpoint.
This is an automated email from the ASF dual-hosted git repository.
aahmed pushed a commit to branch PLSR-2091_3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e286b55044c5492500c2e54a2ad05771ccd59eeb
Author: Ali Ahmed <al...@splunk.com>
AuthorDate: Thu Jun 17 11:44:37 2021 -0700
Support new topic format for broker admin healthcheck endpoint.
* This runs the health check with new topic format
* This help the pulsar-admin tool work correctly
---
.../pulsar/broker/admin/impl/BrokersBase.java | 83 ++++++++++++++--------
.../pulsar/broker/namespace/NamespaceService.java | 35 ++++++++-
.../apache/pulsar/broker/SLAMonitoringTest.java | 2 +-
.../broker/admin/AdminApiHealthCheckTest.java | 83 ++++++++++++++++++++++
.../apache/pulsar/broker/admin/AdminApiTest.java | 4 +-
.../pulsar/broker/admin/v1/V1_AdminApiTest.java | 4 +-
.../org/apache/pulsar/client/admin/Brokers.java | 15 ++++
.../apache/pulsar/common/naming/TopicVersion.java | 24 +++++++
.../pulsar/client/admin/internal/BrokersImpl.java | 20 +++++-
.../pulsar/admin/cli/PulsarAdminToolTest.java | 4 +-
.../org/apache/pulsar/admin/cli/CmdBrokers.java | 6 +-
11 files changed, 239 insertions(+), 41 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
index 2e80dd3..4c03af4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.admin.impl;
import static org.apache.pulsar.broker.service.BrokerService.BROKER_SERVICE_CONFIGURATION_PATH;
import com.google.common.collect.Maps;
import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiParam;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
import java.time.Duration;
@@ -35,11 +36,13 @@ import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
+import javax.ws.rs.QueryParam;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.container.Suspended;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import org.apache.pulsar.PulsarVersion;
+import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.PulsarService.State;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.LeaderBroker;
@@ -55,6 +58,7 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.conf.InternalConfigurationData;
+import org.apache.pulsar.common.naming.TopicVersion;
import org.apache.pulsar.common.policies.data.BrokerInfo;
import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus;
import org.apache.pulsar.common.util.FutureUtil;
@@ -298,13 +302,35 @@ public class BrokersBase extends PulsarWebResource {
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Cluster doesn't exist"),
@ApiResponse(code = 500, message = "Internal server error")})
- public void healthcheck(@Suspended AsyncResponse asyncResponse) throws Exception {
- validateSuperUserAccess();
- String heartbeatNamespace = NamespaceService.getHeartbeatNamespace(
- pulsar().getAdvertisedAddress(), pulsar().getConfiguration());
- String topic = String.format("persistent://%s/healthcheck", heartbeatNamespace);
+ @ApiParam(value = "Topic Version")
+ public void healthcheck(@Suspended AsyncResponse asyncResponse,
+ @QueryParam("topicversion") TopicVersion topicVersion) throws Exception {
+ String topic;
+ PulsarClient client;
+ try {
+ validateSuperUserAccess();
+ String heartbeatNamespace;
+
+ heartbeatNamespace = (topicVersion == TopicVersion.V2)
+ ?
+ NamespaceService.getHeartbeatNamespaceV2(
+ pulsar().getAdvertisedAddress(),
+ pulsar().getConfiguration())
+ :
+ NamespaceService.getHeartbeatNamespace(
+ pulsar().getAdvertisedAddress(),
+ pulsar().getConfiguration());
+
+
+ topic = String.format("persistent://%s/healthcheck", heartbeatNamespace);
- PulsarClient client = pulsar().getClient();
+ LOG.info("Running healthCheck with topic={}", topic);
+
+ client = pulsar().getClient();
+ } catch (Exception e) {
+ LOG.error("Error getting heathcheck topic info", e);
+ throw new PulsarServerException(e);
+ }
String messageStr = UUID.randomUUID().toString();
// create non-partitioned topic manually and close the previous reader if present.
@@ -321,10 +347,11 @@ public class BrokersBase extends PulsarWebResource {
} catch (Exception e) {
LOG.warn("Failed to try to delete subscriptions for health check", e);
}
+
CompletableFuture<Producer<String>> producerFuture =
- client.newProducer(Schema.STRING).topic(topic).createAsync();
+ client.newProducer(Schema.STRING).topic(topic).createAsync();
CompletableFuture<Reader<String>> readerFuture = client.newReader(Schema.STRING)
- .topic(topic).startMessageId(MessageId.latest).createAsync();
+ .topic(topic).startMessageId(MessageId.latest).createAsync();
CompletableFuture<Void> completePromise = new CompletableFuture<>();
@@ -334,7 +361,7 @@ public class BrokersBase extends PulsarWebResource {
completePromise.completeExceptionally(exception);
} else {
producerFuture.thenCompose((producer) -> producer.sendAsync(messageStr))
- .whenComplete((ignore2, exception2) -> {
+ .whenComplete((ignore2, exception2) -> {
if (exception2 != null) {
completePromise.completeExceptionally(exception2);
}
@@ -351,26 +378,26 @@ public class BrokersBase extends PulsarWebResource {
});
completePromise.whenComplete((ignore, exception) -> {
- producerFuture.thenAccept((producer) -> {
- producer.closeAsync().whenComplete((ignore2, exception2) -> {
- if (exception2 != null) {
- LOG.warn("Error closing producer for healthcheck", exception2);
- }
- });
- });
- readerFuture.thenAccept((reader) -> {
- reader.closeAsync().whenComplete((ignore2, exception2) -> {
- if (exception2 != null) {
- LOG.warn("Error closing reader for healthcheck", exception2);
- }
- });
- });
- if (exception != null) {
- asyncResponse.resume(new RestException(exception));
- } else {
- asyncResponse.resume("ok");
- }
+ producerFuture.thenAccept((producer) -> {
+ producer.closeAsync().whenComplete((ignore2, exception2) -> {
+ if (exception2 != null) {
+ LOG.warn("Error closing producer for healthcheck", exception2);
+ }
+ });
});
+ readerFuture.thenAccept((reader) -> {
+ reader.closeAsync().whenComplete((ignore2, exception2) -> {
+ if (exception2 != null) {
+ LOG.warn("Error closing reader for healthcheck", exception2);
+ }
+ });
+ });
+ if (exception != null) {
+ asyncResponse.resume(new RestException(exception));
+ } else {
+ asyncResponse.resume("ok");
+ }
+ });
}
private void healthcheckReadLoop(CompletableFuture<Reader<String>> readerFuture,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index fd6c74e..bf652df 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -140,8 +140,10 @@ public class NamespaceService implements AutoCloseable {
public static final String SLA_NAMESPACE_PROPERTY = "sla-monitor";
public static final Pattern HEARTBEAT_NAMESPACE_PATTERN = Pattern.compile("pulsar/[^/]+/([^:]+:\\d+)");
+ public static final Pattern HEARTBEAT_NAMESPACE_PATTERN_V2 = Pattern.compile("pulsar/([^:]+:\\d+)");
public static final Pattern SLA_NAMESPACE_PATTERN = Pattern.compile(SLA_NAMESPACE_PROPERTY + "/[^/]+/([^:]+:\\d+)");
public static final String HEARTBEAT_NAMESPACE_FMT = "pulsar/%s/%s:%s";
+ public static final String HEARTBEAT_NAMESPACE_FMT_V2 = "pulsar/%s:%s";
public static final String SLA_NAMESPACE_FMT = SLA_NAMESPACE_PROPERTY + "/%s/%s:%s";
public static final String NAMESPACE_ISOLATION_POLICIES = "namespaceIsolationPolicies";
@@ -305,6 +307,12 @@ public class NamespaceService implements AutoCloseable {
LOG.info("added heartbeat namespace name in local cache: ns={}", getHeartbeatNamespace(host, config));
}
+ // ensure that we own the heartbeat namespace
+ if (registerNamespace(getHeartbeatNamespaceV2(host, config), true)) {
+ this.uncountedNamespaces++;
+ LOG.info("added heartbeat namespace name in local cache: ns={}", getHeartbeatNamespaceV2(host, config));
+ }
+
// we may not need strict ownership checking for bootstrap names for now
for (String namespace : config.getBootstrapNamespaces()) {
if (registerNamespace(namespace, false)) {
@@ -471,6 +479,9 @@ public class NamespaceService implements AutoCloseable {
// check if this is Heartbeat or SLAMonitor namespace
candidateBroker = checkHeartbeatNamespace(bundle);
if (candidateBroker == null) {
+ candidateBroker = checkHeartbeatNamespaceV2(bundle);
+ }
+ if (candidateBroker == null) {
String broker = getSLAMonitorBrokerName(bundle);
// checking if the broker is up and running
if (broker != null && isBrokerActive(broker)) {
@@ -1219,7 +1230,6 @@ public class NamespaceService implements AutoCloseable {
return client.getLookup().getTopicsUnderNamespace(namespace, Mode.NON_PERSISTENT);
}
-
public PulsarClientImpl getNamespaceClient(ClusterDataImpl cluster) {
PulsarClientImpl client = namespaceClients.get(cluster);
if (client != null) {
@@ -1295,7 +1305,18 @@ public class NamespaceService implements AutoCloseable {
}
return String.format(HEARTBEAT_NAMESPACE_FMT, config.getClusterName(), host, port);
}
- public static String getSLAMonitorNamespace(String host, ServiceConfiguration config) {
+
+ public static String getHeartbeatNamespaceV2(String host, ServiceConfiguration config) {
+ Integer port = null;
+ if (config.getWebServicePort().isPresent()) {
+ port = config.getWebServicePort().get();
+ } else if (config.getWebServicePortTls().isPresent()) {
+ port = config.getWebServicePortTls().get();
+ }
+ return String.format(HEARTBEAT_NAMESPACE_FMT_V2, host, port);
+ }
+
+ public static String getSLAMonitorNamespace(String host, ServiceConfiguration config) {
Integer port = null;
if (config.getWebServicePort().isPresent()) {
port = config.getWebServicePort().get();
@@ -1315,6 +1336,16 @@ public class NamespaceService implements AutoCloseable {
}
}
+ public static String checkHeartbeatNamespaceV2(ServiceUnitId ns) {
+ Matcher m = HEARTBEAT_NAMESPACE_PATTERN_V2.matcher(ns.getNamespaceObject().toString());
+ if (m.matches()) {
+ LOG.debug("SLAMonitoring namespace matched the lookup namespace {}", ns.getNamespaceObject().toString());
+ return String.format("http://%s", m.group(1));
+ } else {
+ return null;
+ }
+ }
+
public static String getSLAMonitorBrokerName(ServiceUnitId ns) {
Matcher m = SLA_NAMESPACE_PATTERN.matcher(ns.getNamespaceObject().toString());
if (m.matches()) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java
index ce55717..17eceaa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/SLAMonitoringTest.java
@@ -162,7 +162,7 @@ public class SLAMonitoringTest {
Map<String, NamespaceOwnershipStatus> nsMap = pulsarAdmins[i].brokers().getOwnedNamespaces("my-cluster",
list.get(0));
- Assert.assertEquals(nsMap.size(), 2);
+ Assert.assertEquals(nsMap.size(), 3);
}
} catch (Exception e) {
e.printStackTrace();
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
new file mode 100644
index 0000000..e0c887f
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiHealthCheckTest.java
@@ -0,0 +1,83 @@
+/**
+ * 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.broker.admin;
+
+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;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.net.URL;
+
+@Test(groups = "broker")
+@Slf4j
+public class AdminApiHealthCheckTest extends MockedPulsarServiceBaseTest {
+
+ @BeforeMethod
+ @Override
+ public void setup() throws Exception {
+ resetConfig();
+ super.internalSetup();
+ admin.clusters().createCluster("test",
+ ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
+ TenantInfoImpl tenantInfo = new TenantInfoImpl(
+ Sets.newHashSet("role1", "role2"), Sets.newHashSet("test"));
+ admin.tenants().createTenant("pulsar", tenantInfo);
+ admin.namespaces().createNamespace("pulsar/system", Sets.newHashSet("test"));
+ admin.tenants().createTenant("public", tenantInfo);
+ admin.namespaces().createNamespace("public/default", Sets.newHashSet("test"));
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ public void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void testHealthCheckup() throws Exception {
+ admin.brokers().healthcheck();
+ }
+
+ @Test
+ public void testHealthCheckupV1() throws Exception {
+ 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);
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
index 8061b84..5d96389 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
@@ -474,7 +474,7 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
Map<String, NamespaceOwnershipStatus> nsMap = admin.brokers().getOwnedNamespaces("test", list.get(0));
// since sla-monitor ns is not created nsMap.size() == 1 (for HeartBeat Namespace)
- Assert.assertEquals(nsMap.size(), 1);
+ Assert.assertEquals(nsMap.size(), 2);
for (String ns : nsMap.keySet()) {
NamespaceOwnershipStatus nsStatus = nsMap.get(ns);
if (ns.equals(
@@ -490,7 +490,7 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
Assert.assertEquals(parts.length, 2);
Map<String, NamespaceOwnershipStatus> nsMap2 = adminTls.brokers().getOwnedNamespaces("test",
String.format("%s:%d", parts[0], pulsar.getListenPortHTTPS().get()));
- Assert.assertEquals(nsMap2.size(), 1);
+ Assert.assertEquals(nsMap2.size(), 2);
admin.namespaces().deleteNamespace("prop-xyz/ns1");
admin.clusters().deleteCluster("test");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java
index 58ee083..3f67eb6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java
@@ -410,7 +410,7 @@ public class V1_AdminApiTest extends MockedPulsarServiceBaseTest {
Map<String, NamespaceOwnershipStatus> nsMap = admin.brokers().getOwnedNamespaces("use", list.get(0));
// since sla-monitor ns is not created nsMap.size() == 1 (for HeartBeat Namespace)
- Assert.assertEquals(nsMap.size(), 1);
+ Assert.assertEquals(nsMap.size(), 2);
for (String ns : nsMap.keySet()) {
NamespaceOwnershipStatus nsStatus = nsMap.get(ns);
if (ns.equals(
@@ -426,7 +426,7 @@ public class V1_AdminApiTest extends MockedPulsarServiceBaseTest {
Assert.assertEquals(parts.length, 2);
Map<String, NamespaceOwnershipStatus> nsMap2 = adminTls.brokers().getOwnedNamespaces("use",
String.format("%s:%d", parts[0], pulsar.getListenPortHTTPS().get()));
- Assert.assertEquals(nsMap2.size(), 1);
+ Assert.assertEquals(nsMap2.size(), 2);
admin.namespaces().deleteNamespace("prop-xyz/use/ns1");
admin.clusters().deleteCluster("use");
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Brokers.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Brokers.java
index c444b2b..100988c 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Brokers.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Brokers.java
@@ -24,6 +24,7 @@ import java.util.concurrent.CompletableFuture;
import org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException;
import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException;
import org.apache.pulsar.common.conf.InternalConfigurationData;
+import org.apache.pulsar.common.naming.TopicVersion;
import org.apache.pulsar.common.policies.data.BrokerInfo;
import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus;
@@ -268,14 +269,28 @@ public interface Brokers {
*
* @throws PulsarAdminException if the healthcheck fails.
*/
+ @Deprecated
void healthcheck() throws PulsarAdminException;
/**
* Run a healthcheck on the broker asynchronously.
*/
+ @Deprecated
CompletableFuture<Void> healthcheckAsync();
/**
+ * Run a healthcheck on the broker.
+ *
+ * @throws PulsarAdminException if the healthcheck fails.
+ */
+ void healthcheck(TopicVersion topicVersion) throws PulsarAdminException;
+
+ /**
+ * Run a healthcheck on the broker asynchronously.
+ */
+ CompletableFuture<Void> healthcheckAsync(TopicVersion topicVersion);
+
+ /**
* Get version of broker.
* @return version of broker.
*/
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicVersion.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicVersion.java
new file mode 100644
index 0000000..48bed00
--- /dev/null
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicVersion.java
@@ -0,0 +1,24 @@
+/**
+ * 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.naming;
+
+public enum TopicVersion {
+ V1,
+ V2,
+}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java
index 53a3703..b520523 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java
@@ -32,6 +32,7 @@ import org.apache.pulsar.client.admin.Brokers;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.common.conf.InternalConfigurationData;
+import org.apache.pulsar.common.naming.TopicVersion;
import org.apache.pulsar.common.policies.data.BrokerInfo;
import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus;
import org.apache.pulsar.common.util.Codec;
@@ -352,9 +353,21 @@ public class BrokersImpl extends BaseResource implements Brokers {
}
@Override
+ @Deprecated
public void healthcheck() throws PulsarAdminException {
+ healthcheck(TopicVersion.V1);
+ }
+
+ @Override
+ @Deprecated
+ public CompletableFuture<Void> healthcheckAsync() {
+ return healthcheckAsync(TopicVersion.V1);
+ }
+
+ @Override
+ public void healthcheck(TopicVersion topicVersion) throws PulsarAdminException {
try {
- healthcheckAsync().get(this.readTimeoutMs, TimeUnit.MILLISECONDS);
+ healthcheckAsync(topicVersion).get(this.readTimeoutMs, TimeUnit.MILLISECONDS);
} catch (ExecutionException e) {
throw (PulsarAdminException) e.getCause();
} catch (InterruptedException e) {
@@ -366,8 +379,11 @@ public class BrokersImpl extends BaseResource implements Brokers {
}
@Override
- public CompletableFuture<Void> healthcheckAsync() {
+ public CompletableFuture<Void> healthcheckAsync(TopicVersion topicVersion) {
WebTarget path = adminBrokers.path("health");
+ if (topicVersion != null) {
+ path = path.queryParam("topicVersion", topicVersion);
+ }
final CompletableFuture<Void> future = new CompletableFuture<>();
asyncGetRequest(path,
new InvocationCallback<String>() {
diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
index 47e075f..11f5156 100644
--- a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
+++ b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java
@@ -74,11 +74,9 @@ import org.apache.pulsar.common.policies.data.BookiesClusterInfo;
import org.apache.pulsar.common.policies.data.BookiesRackConfiguration;
import org.apache.pulsar.common.policies.data.BundlesData;
import org.apache.pulsar.common.policies.data.ClusterData;
-import org.apache.pulsar.common.policies.data.ClusterDataImpl;
import org.apache.pulsar.common.policies.data.DelayedDeliveryPolicies;
import org.apache.pulsar.common.policies.data.DispatchRate;
import org.apache.pulsar.common.policies.data.FailureDomain;
-import org.apache.pulsar.common.policies.data.FailureDomainImpl;
import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
import org.apache.pulsar.common.policies.data.InactiveTopicPolicies;
import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats.LedgerInfo;
@@ -136,7 +134,7 @@ public class PulsarAdminToolTest {
verify(mockBrokers).getRuntimeConfigurations();
brokers.run(split("healthcheck"));
- verify(mockBrokers).healthcheck();
+ verify(mockBrokers).healthcheck(null);
brokers.run(split("version"));
verify(mockBrokers).getVersion();
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBrokers.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBrokers.java
index 1fe1c29..a9a0dbe 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBrokers.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdBrokers.java
@@ -22,6 +22,7 @@ import org.apache.pulsar.client.admin.PulsarAdmin;
import com.beust.jcommander.Parameter;
import com.beust.jcommander.Parameters;
+import org.apache.pulsar.common.naming.TopicVersion;
import java.util.function.Supplier;
@@ -127,9 +128,12 @@ public class CmdBrokers extends CmdBase {
@Parameters(commandDescription = "Run a health check against the broker")
private class HealthcheckCmd extends CliCommand {
+ @Parameter(names = "--topic-version", description = "topic version V1 is default")
+ private TopicVersion topicVersion;
+
@Override
void run() throws Exception {
- getAdmin().brokers().healthcheck();
+ getAdmin().brokers().healthcheck(topicVersion);
System.out.println("ok");
}