There is a conflict in PIP numbers. The issue has moved to PIP-183. Also there is not only one method that should be modified, I will use this PIP to track all of them.
Thanks, Xiaoyu Hou Anon Hxy <anonhx...@gmail.com> 于2022年7月6日周三 22:12写道： > 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 > > >
