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 25b51493cf066..bb04c1b597e1f 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 @@ -43,6 +43,7 @@ import javax.ws.rs.core.Response.Status; import org.apache.pulsar.broker.PulsarService.State; import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.loadbalance.LeaderBroker; import org.apache.pulsar.broker.loadbalance.LoadManager; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.service.BrokerService; @@ -56,6 +57,7 @@ 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.policies.data.BrokerInfo; import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -92,6 +94,31 @@ public Set getActiveBrokers(@PathParam("cluster") String cluster) throws } } + @GET + @Path("/leaderBroker") + @ApiOperation( + value = "Get the information of the leader broker.", + response = BrokerInfo.class) + @ApiResponses( + value = { + @ApiResponse(code = 401, message = "Authentication required"), + @ApiResponse(code = 403, message = "This operation requires super-user access"), + @ApiResponse(code = 404, message = "Leader broker not found") }) + public BrokerInfo getLeaderBroker() throws Exception { + validateSuperUserAccess(); + + try { + LeaderBroker leaderBroker = pulsar().getLeaderElectionService().getCurrentLeader() + .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Couldn't find leader broker")); + BrokerInfo brokerInfo = new BrokerInfo(); + brokerInfo.setServiceUrl(leaderBroker.getServiceUrl()); + return brokerInfo; + } catch (Exception e) { + LOG.error("[{}] Failed to get the information of the leader broker.", clientAppId(), e); + throw new RestException(e); + } + } + @GET @Path("/{clusterName}/{broker-webserviceurl}/ownedNamespaces") @ApiOperation(value = "Get the list of namespaces served by the specific broker", 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 98e4aac018fac..c63f72eca027a 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 @@ -66,6 +66,7 @@ import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; +import org.apache.pulsar.broker.loadbalance.LeaderBroker; import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl; import org.apache.pulsar.broker.namespace.NamespaceEphemeralData; import org.apache.pulsar.broker.namespace.NamespaceService; @@ -109,6 +110,7 @@ import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType; import org.apache.pulsar.common.policies.data.BacklogQuota.RetentionPolicy; import org.apache.pulsar.common.policies.data.BrokerAssignment; +import org.apache.pulsar.common.policies.data.BrokerInfo; import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.ConsumerStats; import org.apache.pulsar.common.policies.data.NamespaceIsolationData; @@ -441,6 +443,9 @@ public void brokers() throws Exception { Assert.assertNotNull(list2); Assert.assertEquals(list2.size(), 1); + BrokerInfo leaderBroker = admin.brokers().getLeaderBroker(); + Assert.assertEquals(leaderBroker.getServiceUrl(), pulsar.getLeaderElectionService().getCurrentLeader().map(LeaderBroker::getServiceUrl).get()); + Map 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); @@ -634,7 +639,7 @@ public void properties() throws PulsarAdminException { } catch (PulsarAdminException e) { assertTrue(e instanceof NotFoundException); } - + Set allowedClusters = Sets.newHashSet("test"); TenantInfo tenantInfo = new TenantInfo(Sets.newHashSet("role1", "role2"), allowedClusters); admin.tenants().updateTenant("prop-xyz", tenantInfo); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java index 341821892605e..31ca1f236a5ec 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTest.java @@ -70,6 +70,7 @@ import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.authentication.AuthenticationDataHttps; import org.apache.pulsar.broker.cache.ConfigurationCacheService; +import org.apache.pulsar.broker.loadbalance.LeaderBroker; import org.apache.pulsar.broker.web.PulsarWebResource; import org.apache.pulsar.broker.web.RestException; import org.apache.pulsar.common.conf.InternalConfigurationData; @@ -79,6 +80,7 @@ import org.apache.pulsar.common.policies.data.AutoFailoverPolicyData; import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType; import org.apache.pulsar.common.policies.data.BundlesData; +import org.apache.pulsar.common.policies.data.BrokerInfo; import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.NamespaceIsolationData; import org.apache.pulsar.common.policies.data.Policies; @@ -621,6 +623,9 @@ public void brokers() throws Exception { Set activeBrokers = brokers.getActiveBrokers("use"); assertEquals(activeBrokers.size(), 1); assertEquals(activeBrokers, Sets.newHashSet(pulsar.getAdvertisedAddress() + ":" + pulsar.getListenPortHTTP().get())); + + BrokerInfo leaderBroker = brokers.getLeaderBroker(); + assertEquals(leaderBroker.getServiceUrl(), pulsar.getLeaderElectionService().getCurrentLeader().map(LeaderBroker::getServiceUrl).get()); } @Test 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 d0bb9b9db4f90..11f7fd7c74af1 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 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.policies.data.BrokerInfo; import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus; /** @@ -72,6 +73,40 @@ public interface Brokers { */ CompletableFuture> getActiveBrokersAsync(String cluster); + /** + * Get the information of the leader broker. + *

+ * Get the information of the leader broker. + *

+ * Response Example: + * + * 

+     * {serviceUrl:"prod1-broker1.messaging.use.example.com:8080"}
+     *
+ * + * @return the information of the leader broker. + * @throws PulsarAdminException + * Unexpected error + */ + BrokerInfo getLeaderBroker() throws PulsarAdminException; + + /** + * Get the service url of the leader broker asynchronously. + *

+ * Get the service url of the leader broker. + *

+ * Response Example: + * + * 

+     * {serviceUrl:"prod1-broker1.messaging.use.example.com:8080"}
+     *
+ * + * @return the service url of the leader broker + * @throws PulsarAdminException + * Unexpected error + */ + CompletableFuture getLeaderBrokerAsync() throws PulsarAdminException; + /** * Get the map of owned namespaces and their status from a single broker in the cluster. *

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 3db8dfe4fe65d..53b9e075709f6 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.PulsarAdminException; import org.apache.pulsar.client.api.Authentication; import org.apache.pulsar.common.conf.InternalConfigurationData; +import org.apache.pulsar.common.policies.data.BrokerInfo; import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus; import org.apache.pulsar.common.util.Codec; @@ -76,6 +77,39 @@ public void failed(Throwable throwable) { return future; } + @Override + public BrokerInfo getLeaderBroker() throws PulsarAdminException { + try { + return getLeaderBrokerAsync().get(this.readTimeoutMs, TimeUnit.MILLISECONDS); + } catch (ExecutionException e) { + throw (PulsarAdminException) e.getCause(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new PulsarAdminException(e); + } catch (TimeoutException e) { + throw new PulsarAdminException.TimeoutException(e); + } + } + + @Override + public CompletableFuture getLeaderBrokerAsync() { + WebTarget path = adminBrokers.path("leaderBroker"); + final CompletableFuture future = new CompletableFuture<>(); + asyncGetRequest(path, + new InvocationCallback() { + @Override + public void completed(BrokerInfo leaderBroker) { + future.complete(leaderBroker); + } + + @Override + public void failed(Throwable throwable) { + future.completeExceptionally(getApiException(throwable.getCause())); + } + }); + return future; + } + @Override public Map getOwnedNamespaces(String cluster, String brokerUrl) throws PulsarAdminException { 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 3e9dbc310674c..8c311c78d55a1 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 @@ -75,6 +75,9 @@ public void brokers() throws Exception { brokers.run(split("list use")); verify(mockBrokers).getActiveBrokers("use"); + brokers.run(split("leader-broker")); + verify(mockBrokers).getLeaderBroker(); + brokers.run(split("namespaces use --url http://my-service.url:8080")); verify(mockBrokers).getOwnedNamespaces("use", "http://my-service.url:8080"); 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 4ce74100dc0f2..707629c5db1f9 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 @@ -40,6 +40,15 @@ void run() throws Exception { } } + @Parameters(commandDescription = "Get the information of the leader broker") + private class LeaderBroker extends CliCommand { + + @Override + void run() throws Exception { + print(getAdmin().brokers().getLeaderBroker()); + } + } + @Parameters(commandDescription = "List namespaces owned by the broker") private class Namespaces extends CliCommand { @Parameter(description = "cluster-name\n", required = true) @@ -77,7 +86,7 @@ void run() throws Exception { getAdmin().brokers().deleteDynamicConfiguration(configName); } } - + @Parameters(commandDescription = "Get all overridden dynamic-configuration values") private class GetAllConfigurationsCmd extends CliCommand { @@ -86,7 +95,7 @@ void run() throws Exception { print(getAdmin().brokers().getAllDynamicConfigurations()); } } - + @Parameters(commandDescription = "Get list of updatable configuration name") private class GetUpdatableConfigCmd extends CliCommand { @@ -140,6 +149,7 @@ void run() throws Exception { public CmdBrokers(Supplier admin) { super("brokers", admin); jcommander.addCommand("list", new List()); + jcommander.addCommand("leader-broker", new LeaderBroker()); jcommander.addCommand("namespaces", new Namespaces()); jcommander.addCommand("update-dynamic-config", new UpdateConfigurationCmd()); jcommander.addCommand("delete-dynamic-config", new DeleteConfigurationCmd()); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/BrokerInfo.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/BrokerInfo.java new file mode 100644 index 0000000000000..c194a7867e8e9 --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/BrokerInfo.java @@ -0,0 +1,33 @@ +/** + * 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.policies.data; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * Broker Information + */ +@Data +@AllArgsConstructor +@NoArgsConstructor +public class BrokerInfo { + private String serviceUrl; +} diff --git a/site2/docs/admin-api-brokers.md b/site2/docs/admin-api-brokers.md index 01ed0053f79a0..e01f84100677a 100644 --- a/site2/docs/admin-api-brokers.md +++ b/site2/docs/admin-api-brokers.md @@ -48,6 +48,34 @@ admin.brokers().getActiveBrokers(clusterName) +### Get the information of the leader broker + +Fetch the information of the leader broker, for example, the service url. + + + + +```shell +$ pulsar-admin brokers leader-broker +``` + +``` +BrokerInfo(serviceUrl=broker1.use.org.com:8080) +``` + + + +{@inject: endpoint|GET|/admin/v2/brokers/leaderBroker?version=[[pulsar:version_number]]} + + + +```java +admin.brokers().getLeaderBroker() +``` +For the detail of the code above, see [here](https://github.com/apache/pulsar/blob/master/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java#L80) + + + #### list of namespaces owned by a given broker It finds all namespaces which are owned and served by a given broker. diff --git a/site2/docs/reference-pulsar-admin.md b/site2/docs/reference-pulsar-admin.md index 8228c5fdccd8f..11bc87f95b6ef 100644 --- a/site2/docs/reference-pulsar-admin.md +++ b/site2/docs/reference-pulsar-admin.md @@ -135,6 +135,14 @@ Usage $ pulsar-admin brokers list cluster-name ``` +### `leader-broker` +Get the information of the leader broker + +Usage +```bash +$ pulsar-admin brokers leader-broker +``` + ### `namespaces` List namespaces owned by the broker