poorbarcode commented on code in PR #24833:
URL: https://github.com/apache/pulsar/pull/24833#discussion_r2476710868
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -2610,6 +2581,93 @@ protected void
handleGetTopicsOfNamespace(CommandGetTopicsOfNamespace commandGet
}
}
+ private void internalHandleGetTopicsOfNamespace(String namespace,
NamespaceName namespaceName, long requestId,
+
CommandGetTopicsOfNamespace.Mode mode,
+ Optional<String>
topicsPattern, Optional<String> topicsHash,
+ Semaphore lookupSemaphore)
{
+ BooleanSupplier isPermitRequestCancelled = () ->
!ctx().channel().isActive();
+ TopicListSizeResultCache.ResultHolder
+ listSizeHolder =
service.getTopicListSizeResultCache().getTopicListSize(namespaceName.toString(),
mode);
+ listSizeHolder.getSizeAsync().thenAccept(initialSize -> {
+ maxTopicListInFlightLimiter.withAcquiredPermits(initialSize,
+ AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY,
isPermitRequestCancelled, initialPermits -> {
+ return
getBrokerService().pulsar().getNamespaceService()
+ .getListOfUserTopics(namespaceName, mode)
+ .thenAccept(topics -> {
+ long actualSize =
TopicListMemoryLimiter.estimateTopicListSize(topics);
+ listSizeHolder.updateSize(actualSize);
+
maxTopicListInFlightLimiter.withUpdatedPermits(initialPermits, actualSize,
Review Comment:
No, the implementation is like follows
- update
- call `internalAcquire` if `realSize > estimatedSize`, see also
https://github.com/apache/pulsar/pull/24833/files#diff-1daf74806e4d1f752d9576bd63d71678a7a46cbc7a1cf56f71ba88719f87d346R173-R175
- `internalAcquire` will pending the current thread or throws errors, see
also
https://github.com/apache/pulsar/pull/24833/files#diff-1daf74806e4d1f752d9576bd63d71678a7a46cbc7a1cf56f71ba88719f87d346R117-R122
- **method update**
```java
public CompletableFuture<AsyncSemaphorePermit>
update(AsyncSemaphorePermit permit, long newPermits,
BooleanSupplier
isCancelled) {
long oldPermits = permit.getPermits();
long additionalPermits = newPermits - oldPermits;
if (additionalPermits > 0) {
return internalAcquire(newPermits, additionalPermits,
isCancelled);
}
}
```
- **pending requests**
```java
private CompletableFuture<AsyncSemaphorePermit> internalAcquire(long
permits, long acquirePermits,
BooleanSupplier isCancelled) {
validatePermits(permits);
PendingRequest request = new PendingRequest(permits, acquirePermits,
future, isCancelled);
if (!queue.offer(request)) {
future.completeExceptionally(new PermitAcquireQueueFullException(
"Semaphore queue is full"));
return future;
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
