PIP-183: Reduce unnecessary REST call in broker

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);
                             }
                         }
                     })
             );