[DISSCUSS]PIP-181: Reduce unnecessary REST call in broker

[Anon Hxy]

Hi Pulsar community:

I open a pip to discuss "Reduce unnecessary REST call in broker"

Proposal Link: https://github.com/apache/pulsar/issues/16422

---

## Motivation

The design of admin API now is such that: when handle a partitioned topic
request, the broker will query the topic's partition meta, and then use the
internal admin client to query all the non-partitioned topics (I.e. the
suffix of  the topic name is `-partition-`),
even if the non-partitioned topic is owned by the broker, which will cause
unnecessary  REST call in the broker.

we can call the methods directlly,  who handle the non-partitioned topic,
 to reduce the unnecessary  REST call.

## Goal

* Try to call the methods directlly if the non-partitioned topic is owned
by the broker

## Implementation

* We need to check all the place where
`org.apache.pulsar.broker.PulsarService#getAdminClient` is invoked in
`org.apache.pulsar.broker.admin.impl.PersistentTopicsBase`
* take `internalGetPartitionedStats` for example:

  *  Original:
```
   for (int i = 0; i < partitionMetadata.partitions; i++) {
                try {
                    topicStatsFutureList

.add(pulsar().getAdminClient().topics().getStatsAsync(
                                    (topicName.getPartition(i).toString()),
getPreciseBacklog, subscriptionBacklogSize,
                                    getEarliestTimeInBacklog));
                } catch (PulsarServerException e) {
                    asyncResponse.resume(new RestException(e));
                    return;
                }
            }
  ```

   *  Suggest to do like this:
   ```
 for (int i = 0; i < partitionMetadata.partitions; i++) {
                TopicName topicNamePartition = topicName.getPartition(i);
                topicStatsFutureList.add(

pulsar().getNamespaceService().isServiceUnitOwnedAsync(topicName)
                        .thenCompose(owned -> {
                            if (owned) {
                                // local call
                                return
getTopicReferenceAsync(topicNamePartition)
                                    .thenCompose(topic ->

topic.asyncGetStats(getPreciseBacklog, subscriptionBacklogSize,
                                            getEarliestTimeInBacklog));
                            } else {
                                // call from admin client
                                try {

pulsar().getAdminClient().topics().getStatsAsync(topicNamePartition.toString()),
                                        getPreciseBacklog,
subscriptionBacklogSize, getEarliestTimeInBacklog)
                                } catch (PulsarServerException e) {
                                    throw new RestException(e);
                                }
                            }
                        })
                );
   ```

Thanks,
Xiaoyu Hou