You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2026/10/16 14:55:29 UTC
[PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
lhotari opened a new pull request, #26945:
URL: https://github.com/apache/pulsar/pull/26945
Fixes #26937
PIP: [PIP-442](https://github.com/apache/pulsar/blob/master/pip/pip-442.md), [PIP-452](https://github.com/apache/pulsar/blob/master/pip/pip-452.md)
### Motivation
Topic listing memory limits (PIP-442, `maxTopicListInFlight*`) are applied inconsistently across REST endpoints, binary protocol requests, proxy requests and topic list watchers. Cancellation and oversized topic lists can also leave requests without a response or leak permits, listeners or command buffers.
**Cancellation**
- A permit request is checked for cancellation only while it waits in the queue. When the client connection closes after the permits are granted, the topic listing still runs and the response is still serialized.
- The REST namespace topic listing treated a request as cancelled only when its `CompletionCallback` received a throwable. Permits granted after the response had completed without a throwable, for example after an async response timeout, were stored after the callback had run and never released.
- Watcher success responses and updates were cancelled only when the connection became inactive, not when the watcher was closed.
**REST API and PIP-452**
- Only the namespace topic listing applied the limit. The PIP-452 listing (`GET /admin/v2/persistent/{tenant}/{namespace}?properties=...`) calls the `PulsarResourcesExtended` plugin without the limit, and the partitioned, non-persistent, bundle and scalable topic listings weren't limited either.
- The size estimates were kept per (namespace, mode), so listings with different PIP-452 properties overwrote each other's estimate on the broker and in the proxy.
**Topic lists larger than the limit**
- The limiter rejects a request larger than the limit by throwing `IllegalArgumentException` synchronously. `ServerCnx` acquired the initial permits inside `getSizeAsync().thenAccept(...)`, so after one listing of a namespace whose list exceeds `maxTopicListInFlightHeapMemSizeMB`, every later `CommandGetTopicsOfNamespace` for it got no response and leaked a broker-wide lookup semaphore permit. The proxy didn't respond either and leaked the command buffer.
**Topic list watchers**
- A duplicate `CommandWatchTopicList` waits for the watcher's update. When the watcher was closed or the update failed, the request never completed and leaked a lookup semaphore permit.
- Only permit timeouts and full queues were retried for watcher updates. On other failures, the watcher stopped sending updates and never recovered.
- When sending the success response failed, the watcher was only closed: it stayed registered as a listener and the request got no response.
### Modifications
**Permits and cancellation (`pulsar-common`)**
- `AsyncDualMemoryLimiterUtil.withPermitsFuture` and `acquireDirectMemoryPermitsAndWriteAndFlush` check cancellation again after the grant. A cancelled request releases the permits and calls the error handler with `PermitAcquireCancelledException` without running the operation or serializing the response.
- `withAcquiredPermits`, `withUpdatedPermits` and `acquireDirectMemoryPermitsAndWriteAndFlush` pass a request larger than the limit to the error handler as the new `PermitAcquireLimitExceededException` instead of throwing. `AsyncSemaphoreImpl` is unchanged.
**REST API**
- Add `TopicListResponseMemoryLimiter`. `listTopics` acquires heap permits for the size estimate before listing, updates them to the actual size, and holds them until the `AsyncResponse` has completed, since the list is referenced until the response has been written. A response that is done counts as cancelled, and a permit granted after completion is released right away. Permit acquisition failures, including a list larger than the limit, fail the request with 429.
- Apply the limit to all REST topic listings:
- `GET /admin/v2/namespaces/{tenant}/{namespace}/topics`
- `GET /admin/v2/persistent/{tenant}/{namespace}`, including PIP-452 `properties`. With `?bundle=`, the request is charged for the namespace's topic list before the bundle filter is applied.
- `GET /admin/v2/{persistent|non-persistent}/{tenant}/{namespace}/partitioned`
- `GET /admin/v2/non-persistent/{tenant}/{namespace}` and `.../{bundle}`
- `GET /admin/v2/scalable/{tenant}/{namespace}`, including the property filters
- The non-persistent listings (`GET /admin/v2/non-persistent/{tenant}/{namespace}` and the namespace listing in the `NON_PERSISTENT` and `ALL` modes) collect the topics from the bundle listings of the bundle owners, which can be the same broker. They use `listTopicsFromOtherListings`, which acquires permits for the collected list after collecting it, so that the listing doesn't hold permits that a bundle listing on the same broker needs. In the `ALL` mode, the persistent topics are listed first under their own permits, which are updated to the persistent list's size before the non-persistent topics are collected.
**Size estimates**
- Add a `TopicListing` record (namespace, mode, `Type`, bundle range) with the types namespace topics, owned bundle topics, partitioned topics and scalable topics. `TopicListSizeResultCache` is keyed by the `TopicListing` and a murmur3-128 hash of the PIP-452 properties, so listings that return different lists don't share an estimate, and client-provided property values aren't retained. `ServerCnx` and the proxy include the properties in the key.
- A listing whose cached estimate is larger than the heap memory limit fails right away (`TooManyRequests` for the binary protocol, 429 for the REST API) without queuing. Requesting all permits instead would make the request wait until no other topic list is in flight, delaying the listings of all other namespaces. For each cached listing, at most one request per minute gets the limit as its estimate so that it can list the topics and measure the list again, which lets a namespace whose list has become smaller be listed again.
**Topic list watchers (`TopicListService`)**
- Closing a watcher completes the requests waiting for its update with an error and releases their lookup permits. A failed update retries with backoff unless it was cancelled, and the requests waiting for it get an error response on the first failure. When the topic list is larger than the heap memory limit, the watcher is deleted instead of retrying. The client isn't notified, since the protocol has no command for it.
- `PulsarCommandSender.sendWatchTopicListSuccess` and `sendWatchTopicListUpdate` take an `isCancelled` supplier (an internal broker interface). A removed watcher gets no success response or topic updates.
- When the success response can't be sent, the watch request gets an error response while the connection is active (`TooManyRequests` for a response larger than the direct memory limit), and the watcher is deleted unless it had already been closed. A topic update larger than the direct memory limit isn't retried; the failure is logged and the watcher continues.
- A watcher listing failure resets a size estimate that is still initializing, so that concurrent requests don't wait for it.
**Logging and proxy**
- `ServerCnx`, the proxy and `TopicListService` log cancelled requests at debug level and don't write error responses to closed connections.
- The proxy releases the command buffer when the initial heap permits can't be acquired.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- `TopicListRestMemoryLimitTest` (new):
- with all heap permits taken, seven REST listing call variants fail with 429, and they succeed after the permits are released
- the non-persistent listings, including the `ALL` mode, succeed repeatedly when the size estimates of the outer and the nested bundle listing fit within the limit separately but not together
- `TopicListResponseMemoryLimiterTest` (new):
- permits are held until the response completes
- the listing is skipped for a response that is already done
- a response that completes while waiting for permits, during the listing, or concurrently with the grant
- 429 for permit failures and lists larger than the limit
- an oversized listing fails without waiting behind other listings
- `listTopicsFromOtherListings` holds permits only after collecting the topics
- `ServerCnxTest.testGetTopicsOfNamespaceLargerThanHeapMemoryLimit`: repeated requests for a namespace larger than the limit get `TooManyRequests`, and the lookup permits are released.
- `TopicListServiceTest`:
- closing a watcher completes a duplicate watch request
- a failed update is retried, and a waiting request gets an error on the first failure
- a watcher whose list is larger than the limit is closed
- watch sends are cancelled when the watcher is closed
- the request gets an error when the success response can't be sent or is cancelled because the watcher was closed
- a cancelled watcher request resets the size estimate
- `AsyncDualMemoryLimiterUtilTest`: cancellation after the grant, and requests larger than the limit going to the error handler.
- `TopicListSizeResultCacheTest`:
- separate estimates per properties, listing type and bundle
- oversized estimates are re-measured once per interval
- Existing tests:
- `AdminApi2Test`, `PersistentTopicsTest`, `NamespacesTest`, `AdminTest`, `ServerCnxTest`
- `ScalableTopicsListByPropertyTest`, `ScalableTopicsAuthZTest`
- `CustomizedPulsarResourcesExtendedTest`
- the broker and proxy pattern consumer back pressure tests
Each new test was also checked to fail when the fix it covers is reverted.
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [x] The binary protocol
- No command or error code changes. `CommandGetTopicsOfNamespace` for a topic list larger than the limit now gets a `TooManyRequests` error instead of no response. A pending `CommandWatchTopicList` gets an error response when the watcher is closed, its update fails or its success response can't be sent. A watcher whose topic list is larger than the limit is closed on the broker.
- [x] The REST endpoints
- The topic listing endpoints listed above apply the topic list heap memory limit (`maxTopicListInFlightHeapMemSizeMB`). They return 429 when permits can't be acquired within `maxTopicListInFlightHeapMemSizePermitsAcquireTimeoutMillis`, the permit queue is full, or the topic list is larger than the limit. Previously, all of them except the namespace topic listing returned the list without a limit.
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26945:
URL: https://github.com/apache/pulsar/pull/26945#discussion_r4297415405
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TopicListResponseMemoryLimiter.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.broker.admin.impl;
+
+import jakarta.ws.rs.container.AsyncResponse;
+import jakarta.ws.rs.container.CompletionCallback;
+import jakarta.ws.rs.core.Response.Status;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.function.Supplier;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.topiclistlimit.TopicListMemoryLimiter;
+import org.apache.pulsar.broker.topiclistlimit.TopicListSizeResultCache;
+import org.apache.pulsar.broker.topiclistlimit.TopicListing;
+import org.apache.pulsar.broker.web.RestException;
+import org.apache.pulsar.common.semaphore.AsyncDualMemoryLimiter;
+import org.apache.pulsar.common.semaphore.AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * Applies the topic list heap memory limit (PIP-442) to REST topic listing requests.
+ *
+ * <p>The heap memory permits are held until the {@link AsyncResponse} has completed, since the topic list is
+ * referenced until the response has been written. The request is treated as cancelled once the response is done
+ * (for example, after the async response has timed out), so that a request that is no longer waited for doesn't
+ * list the topics, and permits that are granted after the response has completed are released immediately.
+ * A response can combine several listings, each of them holds its own permits until the response completes.
+ */
+public final class TopicListResponseMemoryLimiter {
+ private TopicListResponseMemoryLimiter() {
+ }
+
+ /**
+ * Lists topics while holding heap memory permits for the estimated size of the topic list.
+ *
+ * @param brokerService the broker service
+ * @param response the async response of the REST request, the permits are released when it completes
+ * @param listing the topic listing, which identifies the topic list size estimate
+ * @param properties the topic listing properties (PIP-452), null when not provided
+ * @param topicsSupplier supplier that performs the topic listing
+ * @return a future that completes with the topic list
+ */
+ public static CompletableFuture<List<String>> listTopics(BrokerService brokerService, AsyncResponse response,
+ TopicListing listing,
+ @Nullable Map<String, String> properties,
+ Supplier<CompletableFuture<List<String>>>
+ topicsSupplier) {
+ TopicListSizeResultCache.ResultHolder listSizeHolder = brokerService.getTopicListSizeResultCache()
+ .getTopicListSize(listing, properties);
+ ResponsePermits permits =
+ new ResponsePermits(brokerService.getMaxTopicListInFlightLimiter(), response, listSizeHolder);
+ response.register(permits);
+ return listSizeHolder.getSizeAsync()
+ .thenCompose(initialSize -> permits.acquire(initialSize))
+ .thenCompose(__ -> topicsSupplier.get())
+ .thenCompose(topicList -> {
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(topicList);
+ listSizeHolder.updateSize(actualSize);
+ return permits.update(actualSize).thenApply(___ -> topicList);
+ })
+ .whenComplete((__, t) -> {
+ if (t != null) {
+ // let concurrent requests waiting for the size estimate proceed
+ listSizeHolder.resetIfInitializing();
+ }
+ });
+ }
+
+ /**
+ * Lists topics by collecting them from other topic listings, and then holds heap memory permits for the
+ * collected topic list until the response completes.
+ *
+ * <p>Use this for a listing that gets its topics from topic listings that apply the topic list memory limit
+ * themselves, such as the bundle topic listings of the bundle owners, which can be this broker. Unlike
+ * {@link #listTopics}, no permits are held while the topics are collected: the collected topic lists are
+ * covered by the permits of the other listings until they have responded, and holding permits for an
+ * estimate here could leave too few permits for a listing on this broker that this request waits for.
+ *
+ * @param brokerService the broker service
+ * @param response the async response of the REST request, the permits are released when it completes
+ * @param topicsSupplier supplier that collects the topics from the other listings
+ * @return a future that completes with the topic list
+ */
+ public static CompletableFuture<List<String>> listTopicsFromOtherListings(
+ BrokerService brokerService, AsyncResponse response,
+ Supplier<CompletableFuture<List<String>>> topicsSupplier) {
+ ResponsePermits permits = new ResponsePermits(brokerService.getMaxTopicListInFlightLimiter(), response, null);
+ response.register(permits);
+ if (permits.isCancelled()) {
+ return CompletableFuture.failedFuture(ResponsePermits.cancelled());
+ }
+ return topicsSupplier.get().thenCompose(topicList ->
+ permits.acquire(TopicListMemoryLimiter.estimateTopicListSize(topicList)).thenApply(__ -> topicList));
+ }
+
+ /**
+ * Holds the heap memory permits of a single REST request until its {@link AsyncResponse} completes.
+ */
+ private static final class ResponsePermits implements CompletionCallback {
+ private final AsyncDualMemoryLimiter limiter;
+ private final AsyncResponse response;
+ private final TopicListSizeResultCache.ResultHolder listSizeHolder;
+ // guarded by this
+ private boolean completed;
+ // guarded by this
+ private AsyncDualMemoryLimiterPermit permit;
+
+ ResponsePermits(AsyncDualMemoryLimiter limiter, AsyncResponse response,
+ TopicListSizeResultCache.ResultHolder listSizeHolder) {
+ this.limiter = limiter;
+ this.response = response;
+ this.listSizeHolder = listSizeHolder;
+ }
+
+ boolean isCancelled() {
+ synchronized (this) {
+ if (completed) {
+ return true;
+ }
+ }
+ // the response is resumed only after the topic list has been retrieved, so a response that is
+ // already done has been cancelled or has timed out
+ return response.isDone();
+ }
+
+ CompletableFuture<Void> acquire(long size) {
+ return holdOrFail(() -> limiter.acquire(size, AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY,
+ this::isCancelled));
+ }
+
+ CompletableFuture<Void> update(long size) {
+ AsyncDualMemoryLimiterPermit currentPermit;
+ synchronized (this) {
+ currentPermit = permit;
+ }
+ if (currentPermit == null) {
+ // the response has already completed and the permits have been released
+ return CompletableFuture.failedFuture(cancelled());
+ }
+ return holdOrFail(() -> limiter.update(currentPermit, size, this::isCancelled));
+ }
+
+ private CompletableFuture<Void> holdOrFail(
+ Supplier<CompletableFuture<AsyncDualMemoryLimiterPermit>> permitFutureSupplier) {
+ CompletableFuture<AsyncDualMemoryLimiterPermit> permitFuture;
+ try {
Review Comment:
Done in 27b3538fefd: `holdOrFail` uses `AsyncDualMemoryLimiterUtil.requestPermits`. Agreed on the longer-term fix in `AsyncSemaphoreImpl`; I'd do that after #26929 to avoid conflicting with it.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #26945:
URL: https://github.com/apache/pulsar/pull/26945#discussion_r4297813875
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java:
##########
@@ -208,29 +278,32 @@ synchronized boolean prepareUpdateTopics(Runnable afterUpdateCompletionCallback)
}
}
- synchronized void updateTopics(List<String> topics) {
- if (closed) {
- return;
- }
- matchingTopics.clear();
- TopicList.filterTopicsToStream(topics, topicsPattern).forEach(matchingTopics::add);
- updatingTopics = false;
- if (disconnected) {
- handleNewAndDeletedTopicsWhileDisconnected();
- matchingTopicsBeforeDisconnected = null;
- disconnected = false;
- }
- for (Runnable callback : updateCallbacks) {
- try {
- callback.run();
- } catch (Exception e) {
- log.warn()
- .attr("callback", callback)
- .exception(e)
- .log("Error executing topic list update callback");
+ void updateTopics(List<String> topics) {
+ List<BiConsumer<Collection<String>, Throwable>> callbacks;
+ Collection<String> snapshot;
+ synchronized (this) {
+ if (closed) {
+ return;
+ }
+ matchingTopics.clear();
+ TopicList.filterTopicsToStream(topics, topicsPattern).forEach(matchingTopics::add);
+ updatingTopics = false;
+ if (disconnected) {
+ handleNewAndDeletedTopicsWhileDisconnected();
+ matchingTopicsBeforeDisconnected = null;
+ disconnected = false;
}
+ callbacks = new ArrayList<>(updateCallbacks);
+ updateCallbacks.clear();
+ // the waiting requests respond with the result of this update, also if the matching topics change
+ // after the lock has been released, for example when another update starts
+ snapshot = callbacks.isEmpty() ? null : new ArrayList<>(matchingTopics);
+ }
+ // the callbacks run without holding the lock, since a callback can close the watcher, for example when
+ // sending the response of a waiting request fails
+ for (BiConsumer<Collection<String>, Throwable> callback : callbacks) {
+ runUpdateCallback(callback, snapshot, null);
}
- updateCallbacks.clear();
sendingCompleted();
Review Comment:
`sendingCompleted()` now runs after the lock has been released. If another update starts in between (`prepareUpdateTopics` from a duplicate watch request or a session event sets `sendingInProgress` and clears the queue), this call resets `sendingInProgress` to false while that update is in progress, and topic events during the update are sent right away, with a hash of the partially rebuilt matching topics. Would returning early from `sendingCompleted()` while `updatingTopics` is true work? The update calls `sendingCompleted()` again when it completes, and that would also cover a success response that completes during an update.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java:
##########
@@ -418,47 +501,104 @@ public void handleWatchTopicList(NamespaceName namespaceName, long watcherId, lo
.attr("requestId", requestId)
.log("Received WatchTopicList");
sendTopicListSuccessWithPermitAcquiringRetries(watcherId, requestId, topicList, hash,
- watcher::sendingCompleted, watcher::close);
+ watcher::sendingCompleted, failure -> {
+ if (!isCancellation(failure)) {
+ // the watcher can't be used without the success response
+ deleteTopicListWatcher(watcherId);
+ }
+ // the request is waiting for a response unless the connection was closed
+ if (connection.isActive()) {
+ ServerError error =
+ failure instanceof AsyncSemaphore.PermitAcquireLimitExceededException
+ ? ServerError.TooManyRequests
+ : BrokerServiceException.getClientErrorCode(
+ new BrokerServiceException.ServerMetadataException(
+ failure));
+ connection.getCommandSender().sendErrorResponse(requestId, error,
+ failure.getMessage());
+ }
+ });
lookupSemaphore.release();
})
.exceptionally(ex -> {
- log.warn()
- .attr("connection", connection)
- .attr("namespace", namespaceName)
- .attr("requestId", requestId)
- .exceptionMessage(ex)
- .log("Error WatchTopicList");
- connection.getCommandSender().sendErrorResponse(requestId,
- BrokerServiceException.getClientErrorCode(
- new BrokerServiceException.ServerMetadataException(ex)), ex.getMessage());
- watchers.remove(watcherId, finalWatcherFuture);
+ if (connection.isActive()) {
+ log.warn()
+ .attr("connection", connection)
+ .attr("namespace", namespaceName)
+ .attr("requestId", requestId)
+ .exceptionMessage(ex)
+ .log("Error WatchTopicList");
+ connection.getCommandSender().sendErrorResponse(requestId,
+ BrokerServiceException.getClientErrorCode(
+ new BrokerServiceException.ServerMetadataException(ex)), ex.getMessage());
+ } else {
+ // the connection was closed, there's no client to respond to
+ log.debug()
+ .attr("connection", connection)
+ .attr("namespace", namespaceName)
+ .attr("requestId", requestId)
+ .exceptionMessage(ex)
+ .log("WatchTopicList was cancelled since the connection was closed");
+ }
+ if (createdWatcherFuture != null) {
+ watchers.remove(watcherId, createdWatcherFuture);
Review Comment:
For a new watcher, this future can now fail after the watcher was created and registered as a listener: `whenMatchingTopicsAvailable` waits for an update that started after the watcher was created (e.g. a `ConnectionLost` session event during initialization, or a full update queue), and when that update fails with a retriable error such as a permit timeout, `failUpdateCallbacks` fails this request. That's likely right after a metadata store reconnect, since all watchers refresh at the same time.
Removing only the map entry leaves the watcher registered in `TopicResources.topicListeners` without closing it. Once it's out of `watchers`, `inactivate()` doesn't see it and its update retries stop as cancelled, so it stays registered until the broker restarts, receiving topic events and queueing up to `topicListUpdateMaxQueueSize` update tasks. Before this change, this handler ran only for a failed initialization, which had already closed and deregistered the watcher.
Could this deregister and close the watcher that it removes? For example:
```java
if (createdWatcherFuture != null && watchers.remove(watcherId, createdWatcherFuture)) {
createdWatcherFuture.thenAccept(w -> {
topicResources.deregisterPersistentTopicListener(w);
w.close();
});
}
```
A test that starts an update before the watcher future completes and fails it would cover this.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -186,76 +180,40 @@ protected CompletableFuture<List<String>> internalGetListOfTopics(AsyncResponse
CommandGetTopicsOfNamespace.Mode mode) {
// Use maxTopicListInFlightLimiter to limit inflight get topic listing responses
// to avoid OOME caused by a lot of clients using HTTP service lookups to list topics
- AsyncDualMemoryLimiterImpl maxTopicListInFlightLimiter =
- pulsar().getBrokerService().getMaxTopicListInFlightLimiter();
- TopicListSizeResultCache.ResultHolder listSizeHolder = pulsar().getBrokerService().getTopicListSizeResultCache()
- .getTopicListSize(namespaceName.toString(), mode);
- // setup the permit cancellation function
- AtomicBoolean permitRequestCancelled = new AtomicBoolean(false);
- BooleanSupplier isPermitRequestCancelled = permitRequestCancelled::get;
- // add callback that releases permits when the response completes
- AtomicReference<AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit> initialPermitsRef =
- new AtomicReference<>();
- AtomicReference<AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit> permitsRef = new AtomicReference<>();
- response.register(new CompletionCallback() {
- @Override
- public void onComplete(Throwable throwable) {
- if (throwable != null) {
- // for failed request
- // handle resetting the TopicListSizeResultCache.ResultHolder
- listSizeHolder.resetIfInitializing();
- // cancel any pending permit request
- permitRequestCancelled.set(true);
- }
- AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit initialPermit = initialPermitsRef.get();
- if (initialPermit != null) {
- maxTopicListInFlightLimiter.release(initialPermit);
- }
- AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit permits = permitsRef.get();
- if (permits != null) {
- maxTopicListInFlightLimiter.release(permits);
- }
- }
- });
- return listSizeHolder.getSizeAsync().thenCompose(initialSize -> maxTopicListInFlightLimiter.acquire(initialSize,
- AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled).exceptionally(t -> {
- throw new CompletionException(
- new RestException(Status.TOO_MANY_REQUESTS, "Failed due to heap memory limit exceeded"));
- }).thenCompose(initialPermits -> {
- initialPermitsRef.set(initialPermits);
- // perform the actual get list of topics operation
- return doInternalGetListOfTopics(policies, mode).thenCompose(topicList -> {
- long actualSize = TopicListMemoryLimiter.estimateTopicListSize(topicList);
- listSizeHolder.updateSize(actualSize);
- return maxTopicListInFlightLimiter.update(initialPermits, actualSize, isPermitRequestCancelled)
- .exceptionally(t -> {
- throw new CompletionException(new RestException(Status.TOO_MANY_REQUESTS,
- "Failed due to heap memory limit exceeded"));
- }).thenApply(permits -> {
- permitsRef.set(permits);
- initialPermitsRef.set(null);
- return topicList;
- });
- });
- }));
- }
-
- private CompletableFuture<List<String>> doInternalGetListOfTopics(Policies policies,
- CommandGetTopicsOfNamespace.Mode mode) {
switch (mode) {
case ALL:
- return pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName)
- .thenCombine(internalGetNonPersistentTopics(policies),
- (persistentTopics, nonPersistentTopics) ->
- ListUtils.union(persistentTopics, nonPersistentTopics));
+ // The non-persistent topics are collected first from the bundle listings, which acquire permits
+ // from the same limiter, possibly on this broker, so no permits are held while waiting for them.
+ // The listing then acquires permits once, for the whole topic list, with the estimate of the ALL
+ // listing, so that it doesn't hold permits for one part while waiting for permits for the other.
+ return internalGetNonPersistentTopics(policies)
Review Comment:
With this order, an `ALL` listing whose cached estimate is larger than the limit, or that can't get permits, still collects the non-persistent topics from all bundle owners before it fails with 429. For `HttpLookupService` pattern consumers that poll an oversized namespace, that's one bundle listing per bundle per poll. Could it fail fast on an oversized cached `ALL` estimate before collecting? Not blocking.
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/topiclistlimit/TopicListSizeResultCache.java:
##########
@@ -54,23 +177,70 @@ record CacheKey(String namespaceName, CommandGetTopicsOfNamespace.Mode mode) {
public static class ResultHolder {
private final AtomicReference<CompletableFuture<Long>> topicListSizeFuture =
new AtomicReference<>(null);
+ private final OversizedRemeasurePolicy policy;
+ private final long sizeLimit;
+ private final long initialEstimate;
+ // the earliest time when a request for an oversized listing may measure the topic list again
+ private volatile long nextRemeasureNanos;
+ // the earliest time when the topic list is measured again even if the limiter is busy
+ private volatile long nextForcedRemeasureNanos;
+
+ public ResultHolder() {
+ this(OversizedRemeasurePolicy.UNLIMITED, 0);
+ }
+
+ private ResultHolder(OversizedRemeasurePolicy policy, long knownSizeEstimate) {
+ this.policy = policy;
+ this.sizeLimit = policy.sizeLimit();
+ this.initialEstimate = Math.max(Math.min(INITIAL_TOPIC_LIST_HEAP_SIZE, sizeLimit), knownSizeEstimate);
+ long now = System.nanoTime();
+ // a listing that starts from an oversized known estimate hasn't been measured yet
+ this.nextRemeasureNanos = now;
Review Comment:
Each properties-keyed holder gets its own re-measure schedule. Since the properties come from the client, the number of forced re-measures, each requesting the whole limit while other listings are in flight, still grows with the number of distinct property maps for an oversized listing on a busy broker. Would a broker-wide minimum interval between forced re-measures (for example kept with the gate) be worth it? Minor, since it needs an oversized listing.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26945:
URL: https://github.com/apache/pulsar/pull/26945#discussion_r4298343210
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java:
##########
@@ -418,47 +501,104 @@ public void handleWatchTopicList(NamespaceName namespaceName, long watcherId, lo
.attr("requestId", requestId)
.log("Received WatchTopicList");
sendTopicListSuccessWithPermitAcquiringRetries(watcherId, requestId, topicList, hash,
- watcher::sendingCompleted, watcher::close);
+ watcher::sendingCompleted, failure -> {
+ if (!isCancellation(failure)) {
+ // the watcher can't be used without the success response
+ deleteTopicListWatcher(watcherId);
+ }
+ // the request is waiting for a response unless the connection was closed
+ if (connection.isActive()) {
+ ServerError error =
+ failure instanceof AsyncSemaphore.PermitAcquireLimitExceededException
+ ? ServerError.TooManyRequests
+ : BrokerServiceException.getClientErrorCode(
+ new BrokerServiceException.ServerMetadataException(
+ failure));
+ connection.getCommandSender().sendErrorResponse(requestId, error,
+ failure.getMessage());
+ }
+ });
lookupSemaphore.release();
})
.exceptionally(ex -> {
- log.warn()
- .attr("connection", connection)
- .attr("namespace", namespaceName)
- .attr("requestId", requestId)
- .exceptionMessage(ex)
- .log("Error WatchTopicList");
- connection.getCommandSender().sendErrorResponse(requestId,
- BrokerServiceException.getClientErrorCode(
- new BrokerServiceException.ServerMetadataException(ex)), ex.getMessage());
- watchers.remove(watcherId, finalWatcherFuture);
+ if (connection.isActive()) {
+ log.warn()
+ .attr("connection", connection)
+ .attr("namespace", namespaceName)
+ .attr("requestId", requestId)
+ .exceptionMessage(ex)
+ .log("Error WatchTopicList");
+ connection.getCommandSender().sendErrorResponse(requestId,
+ BrokerServiceException.getClientErrorCode(
+ new BrokerServiceException.ServerMetadataException(ex)), ex.getMessage());
+ } else {
+ // the connection was closed, there's no client to respond to
+ log.debug()
+ .attr("connection", connection)
+ .attr("namespace", namespaceName)
+ .attr("requestId", requestId)
+ .exceptionMessage(ex)
+ .log("WatchTopicList was cancelled since the connection was closed");
+ }
+ if (createdWatcherFuture != null) {
+ watchers.remove(watcherId, createdWatcherFuture);
Review Comment:
Good catch, thanks. Fixed in c3d36f5dcdd as you suggested: when the failed request removes the watcher that it created, it also deregisters the listener and closes the watcher. `testFailedWatchRequestClosesTheWatcherItCreated` starts an update (a `ConnectionLost` session event) while the watcher is being created, fails it, and checks the error response, the lookup permit, the map entry and the deregistration.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java:
##########
@@ -208,29 +278,32 @@ synchronized boolean prepareUpdateTopics(Runnable afterUpdateCompletionCallback)
}
}
- synchronized void updateTopics(List<String> topics) {
- if (closed) {
- return;
- }
- matchingTopics.clear();
- TopicList.filterTopicsToStream(topics, topicsPattern).forEach(matchingTopics::add);
- updatingTopics = false;
- if (disconnected) {
- handleNewAndDeletedTopicsWhileDisconnected();
- matchingTopicsBeforeDisconnected = null;
- disconnected = false;
- }
- for (Runnable callback : updateCallbacks) {
- try {
- callback.run();
- } catch (Exception e) {
- log.warn()
- .attr("callback", callback)
- .exception(e)
- .log("Error executing topic list update callback");
+ void updateTopics(List<String> topics) {
+ List<BiConsumer<Collection<String>, Throwable>> callbacks;
+ Collection<String> snapshot;
+ synchronized (this) {
+ if (closed) {
+ return;
+ }
+ matchingTopics.clear();
+ TopicList.filterTopicsToStream(topics, topicsPattern).forEach(matchingTopics::add);
+ updatingTopics = false;
+ if (disconnected) {
+ handleNewAndDeletedTopicsWhileDisconnected();
+ matchingTopicsBeforeDisconnected = null;
+ disconnected = false;
}
+ callbacks = new ArrayList<>(updateCallbacks);
+ updateCallbacks.clear();
+ // the waiting requests respond with the result of this update, also if the matching topics change
+ // after the lock has been released, for example when another update starts
+ snapshot = callbacks.isEmpty() ? null : new ArrayList<>(matchingTopics);
+ }
+ // the callbacks run without holding the lock, since a callback can close the watcher, for example when
+ // sending the response of a waiting request fails
+ for (BiConsumer<Collection<String>, Throwable> callback : callbacks) {
+ runUpdateCallback(callback, snapshot, null);
}
- updateCallbacks.clear();
sendingCompleted();
Review Comment:
Right, thanks. Fixed in c3d36f5dcdd: `sendingCompleted()` returns early while `updatingTopics` is true, and the update calls it again when it completes. `testTopicChangesAreQueuedWhileUpdateIsInProgress` covers a send that completes during an update.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -186,76 +180,40 @@ protected CompletableFuture<List<String>> internalGetListOfTopics(AsyncResponse
CommandGetTopicsOfNamespace.Mode mode) {
// Use maxTopicListInFlightLimiter to limit inflight get topic listing responses
// to avoid OOME caused by a lot of clients using HTTP service lookups to list topics
- AsyncDualMemoryLimiterImpl maxTopicListInFlightLimiter =
- pulsar().getBrokerService().getMaxTopicListInFlightLimiter();
- TopicListSizeResultCache.ResultHolder listSizeHolder = pulsar().getBrokerService().getTopicListSizeResultCache()
- .getTopicListSize(namespaceName.toString(), mode);
- // setup the permit cancellation function
- AtomicBoolean permitRequestCancelled = new AtomicBoolean(false);
- BooleanSupplier isPermitRequestCancelled = permitRequestCancelled::get;
- // add callback that releases permits when the response completes
- AtomicReference<AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit> initialPermitsRef =
- new AtomicReference<>();
- AtomicReference<AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit> permitsRef = new AtomicReference<>();
- response.register(new CompletionCallback() {
- @Override
- public void onComplete(Throwable throwable) {
- if (throwable != null) {
- // for failed request
- // handle resetting the TopicListSizeResultCache.ResultHolder
- listSizeHolder.resetIfInitializing();
- // cancel any pending permit request
- permitRequestCancelled.set(true);
- }
- AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit initialPermit = initialPermitsRef.get();
- if (initialPermit != null) {
- maxTopicListInFlightLimiter.release(initialPermit);
- }
- AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit permits = permitsRef.get();
- if (permits != null) {
- maxTopicListInFlightLimiter.release(permits);
- }
- }
- });
- return listSizeHolder.getSizeAsync().thenCompose(initialSize -> maxTopicListInFlightLimiter.acquire(initialSize,
- AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled).exceptionally(t -> {
- throw new CompletionException(
- new RestException(Status.TOO_MANY_REQUESTS, "Failed due to heap memory limit exceeded"));
- }).thenCompose(initialPermits -> {
- initialPermitsRef.set(initialPermits);
- // perform the actual get list of topics operation
- return doInternalGetListOfTopics(policies, mode).thenCompose(topicList -> {
- long actualSize = TopicListMemoryLimiter.estimateTopicListSize(topicList);
- listSizeHolder.updateSize(actualSize);
- return maxTopicListInFlightLimiter.update(initialPermits, actualSize, isPermitRequestCancelled)
- .exceptionally(t -> {
- throw new CompletionException(new RestException(Status.TOO_MANY_REQUESTS,
- "Failed due to heap memory limit exceeded"));
- }).thenApply(permits -> {
- permitsRef.set(permits);
- initialPermitsRef.set(null);
- return topicList;
- });
- });
- }));
- }
-
- private CompletableFuture<List<String>> doInternalGetListOfTopics(Policies policies,
- CommandGetTopicsOfNamespace.Mode mode) {
switch (mode) {
case ALL:
- return pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName)
- .thenCombine(internalGetNonPersistentTopics(policies),
- (persistentTopics, nonPersistentTopics) ->
- ListUtils.union(persistentTopics, nonPersistentTopics));
+ // The non-persistent topics are collected first from the bundle listings, which acquire permits
+ // from the same limiter, possibly on this broker, so no permits are held while waiting for them.
+ // The listing then acquires permits once, for the whole topic list, with the estimate of the ALL
+ // listing, so that it doesn't hold permits for one part while waiting for permits for the other.
+ return internalGetNonPersistentTopics(policies)
Review Comment:
Done in c3d36f5dcdd: the `ALL` listing now gets its size estimate before collecting, and an estimate larger than the limit fails right away without collecting (a re-measuring request still collects). The permits are still acquired only after collecting. This is in `TopicListResponseMemoryLimiter.listTopicsAfterCollecting`, covered by `testListTopicsAfterCollectingSkipsCollectingWhenEstimateIsLargerThanLimit`.
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/topiclistlimit/TopicListSizeResultCache.java:
##########
@@ -54,23 +177,70 @@ record CacheKey(String namespaceName, CommandGetTopicsOfNamespace.Mode mode) {
public static class ResultHolder {
private final AtomicReference<CompletableFuture<Long>> topicListSizeFuture =
new AtomicReference<>(null);
+ private final OversizedRemeasurePolicy policy;
+ private final long sizeLimit;
+ private final long initialEstimate;
+ // the earliest time when a request for an oversized listing may measure the topic list again
+ private volatile long nextRemeasureNanos;
+ // the earliest time when the topic list is measured again even if the limiter is busy
+ private volatile long nextForcedRemeasureNanos;
+
+ public ResultHolder() {
+ this(OversizedRemeasurePolicy.UNLIMITED, 0);
+ }
+
+ private ResultHolder(OversizedRemeasurePolicy policy, long knownSizeEstimate) {
+ this.policy = policy;
+ this.sizeLimit = policy.sizeLimit();
+ this.initialEstimate = Math.max(Math.min(INITIAL_TOPIC_LIST_HEAP_SIZE, sizeLimit), knownSizeEstimate);
+ long now = System.nanoTime();
+ // a listing that starts from an oversized known estimate hasn't been measured yet
+ this.nextRemeasureNanos = now;
Review Comment:
Agreed, done in c3d36f5dcdd: forced re-measures (the ones that don't wait for an idle limiter) are now limited for the whole broker, at most one per forced interval (10 minutes), in addition to the per-listing schedule and the gate that allows one re-measure at a time. Re-measures while the limiter is idle aren't limited by it, since they don't make other listings wait. `testForcedRemeasuresAreLimitedForTheWholeCache` covers it with two properties-keyed listings.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26945:
URL: https://github.com/apache/pulsar/pull/26945#issuecomment-6221047025
Thanks for the detailed review @codelipenghui! I pushed 27b3538fefd with the changes, and replied to the inline comments.
1. Properties-keyed estimates: a new properties key starts from the estimate without properties.
2. `ALL` mode: collects the non-persistent topics without holding permits and then acquires permits once for the whole list.
3. Oversized re-measure: I'd rather keep the full reservation than let the estimate lapse to 10 KB, since that would list a namespace known to exceed the limit with only 10 KB reserved. The re-measure is now only granted while the limiter is idle, for one listing at a time, starting one interval after the oversized measurement, with a forced re-measure after 10 minutes so that a broker that is never idle can recover. Details in the inline reply.
Smaller items:
- The watch request whose watcher is removed before the first attempt now gets an error response (and the callbacks no longer run twice after retries or under the watcher lock).
- 429 and the cancellation 503 aren't logged at ERROR anymore.
- `Fixes #26937`: you're right, the non-persistent listing and `HttpLookupService` gaps remain. I changed it to "Main Issue: #26937" so the issue stays open for those.
- Release note: agreed, the 429 responses of these REST endpoints need a release note.
- Backport: agreed, a minimal backport for branch-4.0 and branch-4.1 (the `ServerCnx`/proxy response fix, checking cancellation after the grant and the watcher request completion) is better than cherry-picking this PR, in separate PRs.
Test gaps:
- Added: a new properties key with an existing estimate without properties, `ALL` mode with a persistent topic list of 60% of the limit, and `ServerCnxTest.testGetTopicsOfNamespaceLargerThanHeapMemoryLimitFailsFastAfterMeasuring`, which uses a cache configured like the broker's (rejection at the update, then failing fast without listing). The re-measure itself is covered by `TopicListSizeResultCacheTest`, since the interval is one minute in the broker.
- Not added: a proxy test for the oversized and cancelled paths. There's no unit-test harness for `LookupProxyHandler`; the shared paths (`withAcquiredPermits`/`withUpdatedPermits`/`acquireDirectMemoryPermitsAndWriteAndFlush` passing the rejection and the cancellation to the error handler) are covered by `AsyncDualMemoryLimiterUtilTest`.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #26945:
URL: https://github.com/apache/pulsar/pull/26945#discussion_r4296926739
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/topiclistlimit/TopicListSizeResultCache.java:
##########
@@ -54,23 +130,59 @@ record CacheKey(String namespaceName, CommandGetTopicsOfNamespace.Mode mode) {
public static class ResultHolder {
private final AtomicReference<CompletableFuture<Long>> topicListSizeFuture =
new AtomicReference<>(null);
+ private final long sizeLimit;
+ private final long oversizedRemeasureIntervalNanos;
+ // the earliest time when a request for an oversized listing may measure the topic list again
+ private final AtomicLong nextOversizedRemeasureNanos;
+
+ public ResultHolder() {
+ this(Long.MAX_VALUE, 0);
+ }
+
+ ResultHolder(long sizeLimit, long oversizedRemeasureIntervalNanos) {
+ this.sizeLimit = sizeLimit;
+ this.oversizedRemeasureIntervalNanos = oversizedRemeasureIntervalNanos;
+ this.nextOversizedRemeasureNanos = new AtomicLong(System.nanoTime());
+ }
/**
* Get the topic list size estimate. The first request will return the initial estimate
* and update the estimate based on the returned size of the topic list. Other concurrent requests
* will wait for the first request to complete and use the estimate of the first request.
* Subsequent requests will use estimate which gets updated based on the returned size of the topic list
* of each request.
+ * When the estimate is larger than the size limit, it's returned as-is so that the request fails without
+ * waiting for permits, except for one request per interval, which gets the size limit so that it can
+ * measure the topic list again.
* @return a future that will return the topic list size estimate
*/
public CompletableFuture<Long> getSizeAsync() {
+ CompletableFuture<Long> estimate;
if (topicListSizeFuture.compareAndSet(null, new CompletableFuture<>())) {
// let the first request proceed with the initial estimate
- return CompletableFuture.completedFuture(INITIAL_TOPIC_LIST_HEAP_SIZE);
+ estimate = CompletableFuture.completedFuture(Math.min(INITIAL_TOPIC_LIST_HEAP_SIZE, sizeLimit));
} else {
// all other requests wait for the first one to complete
- return topicListSizeFuture.get();
+ estimate = topicListSizeFuture.get();
}
+ return sizeLimit == Long.MAX_VALUE ? estimate : estimate.thenApply(this::limitOversizedEstimate);
+ }
+
+ private long limitOversizedEstimate(long estimate) {
+ if (estimate <= sizeLimit) {
+ return estimate;
+ }
+ long now = System.nanoTime();
+ long next = nextOversizedRemeasureNanos.get();
+ if (now - next >= 0
+ && nextOversizedRemeasureNanos.compareAndSet(next, now + oversizedRemeasureIntervalNanos)) {
+ // a request that can be granted, so that the topic list gets measured again
+ return sizeLimit;
Review Comment:
Returning `sizeLimit` makes the re-measuring request ask for the whole limit. `AsyncSemaphoreImpl.internalProcessQueue` is FIFO (it stops at a head request that doesn't fit), so this request waits until all topic list permits in flight (binary protocol, REST and watchers) have been released, and every later request on the broker waits behind it, up to `maxTopicListInFlightHeapMemSizePermitsAcquireTimeoutMillis` (25 s by default). Then it holds the whole limit while listing a namespace that is most likely still larger than the limit. This happens once per minute for each oversized listing, and with the properties in the key, the number of listings depends on what clients send.
Have you considered re-measuring with the normal initial estimate instead, i.e. letting the oversized estimate lapse after the interval? That accepts the same exposure as the first listing of any namespace without making all other listings wait.
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/topiclistlimit/TopicListSizeResultCache.java:
##########
@@ -54,23 +130,59 @@ record CacheKey(String namespaceName, CommandGetTopicsOfNamespace.Mode mode) {
public static class ResultHolder {
private final AtomicReference<CompletableFuture<Long>> topicListSizeFuture =
new AtomicReference<>(null);
+ private final long sizeLimit;
+ private final long oversizedRemeasureIntervalNanos;
+ // the earliest time when a request for an oversized listing may measure the topic list again
+ private final AtomicLong nextOversizedRemeasureNanos;
+
+ public ResultHolder() {
+ this(Long.MAX_VALUE, 0);
+ }
+
+ ResultHolder(long sizeLimit, long oversizedRemeasureIntervalNanos) {
+ this.sizeLimit = sizeLimit;
+ this.oversizedRemeasureIntervalNanos = oversizedRemeasureIntervalNanos;
+ this.nextOversizedRemeasureNanos = new AtomicLong(System.nanoTime());
Review Comment:
Since the deadline starts when the holder is created, the first request after the first oversized measurement gets the limit right away, so a listing that has just become oversized is listed twice in a row (`testOversizedEstimateIsRemeasuredOncePerInterval` asserts this). Setting the deadline when `updateSize()` records a size larger than the limit would make the first re-measure happen one interval later.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java:
##########
@@ -135,7 +135,7 @@ public void getList(
@Parameter(description = "properties for customized topic listing plugin, format: k1=v1,k2=v2")
@QueryParam("properties") String propertiesStr) {
validateNamespaceName(tenant, namespace);
- internalGetListAsync(Optional.ofNullable(bundle), parseProperties(propertiesStr))
+ internalGetListAsync(asyncResponse, Optional.ofNullable(bundle), parseProperties(propertiesStr))
Review Comment:
429 is now an expected response of this endpoint, but it goes through the `exceptionally` handler below, which logs at ERROR with a stack trace (`isNot307And404Exception` returns true for 429). While the limiter rejects requests, that's one ERROR with a stack trace per rejected request. The same applies to `getPartitionedTopicList`, the non-persistent and scalable topic listings, and `Namespaces.getTopics`. Could 429 (and the 503 "Request was cancelled") be logged at debug level, or without the stack trace?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java:
##########
@@ -702,14 +817,16 @@ private void performOperationWithPermitAcquiringRetries(long watcherId, String o
AtomicReference<Supplier<CompletableFuture<Void>>> operationRef = new AtomicReference<>();
// create the error handler for the operation
Function<Throwable, CompletableFuture<Void>> permitAcquireErrorHandler =
- createPermitAcquireErrorHandler(watcherId, operationName, () -> operationRef.get().get());
+ createPermitAcquireErrorHandler(watcherId, operationName, failOnCancellation,
+ () -> operationRef.get().get());
// create the async operation using the factory function. Pass the error handler to the factory function.
Supplier<CompletableFuture<Void>> asyncOperation = asyncOperationFactory.apply(permitAcquireErrorHandler);
// set the operation to run into the operation reference
operationRef.set(() -> {
if (!connection.isActive() || !watchers.containsKey(watcherId)) {
// do nothing if the connection has already been closed or the watcher has been removed
- return CompletableFuture.completedFuture(null);
+ return failOnCancellation ? CompletableFuture.failedFuture(newWatcherClosedException(watcherId))
Review Comment:
On the first attempt, this failed future is ignored: `performOperationWithPermitAcquiringRetries` calls `operationRef.get().get()` without using the result, and the `whenComplete` that calls `failedCompletionCallback` is part of `asyncOperation`, which isn't called here. If the watcher is removed before the success response is sent the first time (e.g. by a concurrent `CommandWatchTopicListClose`, or by another request's failure path that calls `deleteTopicListWatcher`), the watch request gets no response although the connection is active. Retries are fine, since `FutureUtil.completeAfter` passes the failure on. Returning the future and handling the completion in the caller would cover both cases.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TopicListResponseMemoryLimiter.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.broker.admin.impl;
+
+import jakarta.ws.rs.container.AsyncResponse;
+import jakarta.ws.rs.container.CompletionCallback;
+import jakarta.ws.rs.core.Response.Status;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.function.Supplier;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.topiclistlimit.TopicListMemoryLimiter;
+import org.apache.pulsar.broker.topiclistlimit.TopicListSizeResultCache;
+import org.apache.pulsar.broker.topiclistlimit.TopicListing;
+import org.apache.pulsar.broker.web.RestException;
+import org.apache.pulsar.common.semaphore.AsyncDualMemoryLimiter;
+import org.apache.pulsar.common.semaphore.AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * Applies the topic list heap memory limit (PIP-442) to REST topic listing requests.
+ *
+ * <p>The heap memory permits are held until the {@link AsyncResponse} has completed, since the topic list is
+ * referenced until the response has been written. The request is treated as cancelled once the response is done
+ * (for example, after the async response has timed out), so that a request that is no longer waited for doesn't
+ * list the topics, and permits that are granted after the response has completed are released immediately.
+ * A response can combine several listings, each of them holds its own permits until the response completes.
+ */
+public final class TopicListResponseMemoryLimiter {
+ private TopicListResponseMemoryLimiter() {
+ }
+
+ /**
+ * Lists topics while holding heap memory permits for the estimated size of the topic list.
+ *
+ * @param brokerService the broker service
+ * @param response the async response of the REST request, the permits are released when it completes
+ * @param listing the topic listing, which identifies the topic list size estimate
+ * @param properties the topic listing properties (PIP-452), null when not provided
+ * @param topicsSupplier supplier that performs the topic listing
+ * @return a future that completes with the topic list
+ */
+ public static CompletableFuture<List<String>> listTopics(BrokerService brokerService, AsyncResponse response,
+ TopicListing listing,
+ @Nullable Map<String, String> properties,
+ Supplier<CompletableFuture<List<String>>>
+ topicsSupplier) {
+ TopicListSizeResultCache.ResultHolder listSizeHolder = brokerService.getTopicListSizeResultCache()
+ .getTopicListSize(listing, properties);
+ ResponsePermits permits =
+ new ResponsePermits(brokerService.getMaxTopicListInFlightLimiter(), response, listSizeHolder);
+ response.register(permits);
+ return listSizeHolder.getSizeAsync()
+ .thenCompose(initialSize -> permits.acquire(initialSize))
+ .thenCompose(__ -> topicsSupplier.get())
+ .thenCompose(topicList -> {
+ long actualSize = TopicListMemoryLimiter.estimateTopicListSize(topicList);
+ listSizeHolder.updateSize(actualSize);
+ return permits.update(actualSize).thenApply(___ -> topicList);
+ })
+ .whenComplete((__, t) -> {
+ if (t != null) {
+ // let concurrent requests waiting for the size estimate proceed
+ listSizeHolder.resetIfInitializing();
+ }
+ });
+ }
+
+ /**
+ * Lists topics by collecting them from other topic listings, and then holds heap memory permits for the
+ * collected topic list until the response completes.
+ *
+ * <p>Use this for a listing that gets its topics from topic listings that apply the topic list memory limit
+ * themselves, such as the bundle topic listings of the bundle owners, which can be this broker. Unlike
+ * {@link #listTopics}, no permits are held while the topics are collected: the collected topic lists are
+ * covered by the permits of the other listings until they have responded, and holding permits for an
+ * estimate here could leave too few permits for a listing on this broker that this request waits for.
+ *
+ * @param brokerService the broker service
+ * @param response the async response of the REST request, the permits are released when it completes
+ * @param topicsSupplier supplier that collects the topics from the other listings
+ * @return a future that completes with the topic list
+ */
+ public static CompletableFuture<List<String>> listTopicsFromOtherListings(
+ BrokerService brokerService, AsyncResponse response,
+ Supplier<CompletableFuture<List<String>>> topicsSupplier) {
+ ResponsePermits permits = new ResponsePermits(brokerService.getMaxTopicListInFlightLimiter(), response, null);
+ response.register(permits);
+ if (permits.isCancelled()) {
+ return CompletableFuture.failedFuture(ResponsePermits.cancelled());
+ }
+ return topicsSupplier.get().thenCompose(topicList ->
+ permits.acquire(TopicListMemoryLimiter.estimateTopicListSize(topicList)).thenApply(__ -> topicList));
+ }
+
+ /**
+ * Holds the heap memory permits of a single REST request until its {@link AsyncResponse} completes.
+ */
+ private static final class ResponsePermits implements CompletionCallback {
+ private final AsyncDualMemoryLimiter limiter;
+ private final AsyncResponse response;
+ private final TopicListSizeResultCache.ResultHolder listSizeHolder;
+ // guarded by this
+ private boolean completed;
+ // guarded by this
+ private AsyncDualMemoryLimiterPermit permit;
+
+ ResponsePermits(AsyncDualMemoryLimiter limiter, AsyncResponse response,
+ TopicListSizeResultCache.ResultHolder listSizeHolder) {
+ this.limiter = limiter;
+ this.response = response;
+ this.listSizeHolder = listSizeHolder;
+ }
+
+ boolean isCancelled() {
+ synchronized (this) {
+ if (completed) {
+ return true;
+ }
+ }
+ // the response is resumed only after the topic list has been retrieved, so a response that is
+ // already done has been cancelled or has timed out
+ return response.isDone();
+ }
+
+ CompletableFuture<Void> acquire(long size) {
+ return holdOrFail(() -> limiter.acquire(size, AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY,
+ this::isCancelled));
+ }
+
+ CompletableFuture<Void> update(long size) {
+ AsyncDualMemoryLimiterPermit currentPermit;
+ synchronized (this) {
+ currentPermit = permit;
+ }
+ if (currentPermit == null) {
+ // the response has already completed and the permits have been released
+ return CompletableFuture.failedFuture(cancelled());
+ }
+ return holdOrFail(() -> limiter.update(currentPermit, size, this::isCancelled));
+ }
+
+ private CompletableFuture<Void> holdOrFail(
+ Supplier<CompletableFuture<AsyncDualMemoryLimiterPermit>> permitFutureSupplier) {
+ CompletableFuture<AsyncDualMemoryLimiterPermit> permitFuture;
+ try {
Review Comment:
nit: this duplicates `AsyncDualMemoryLimiterUtil.requestPermits`, with a broader catch. Reusing it would keep a single conversion of the synchronous rejection. Longer term, if `AsyncSemaphoreImpl.acquire`/`update` returned a failed future for permits larger than the limit (CODING.md: a method returning `CompletableFuture` must not throw synchronously), these wrappers wouldn't be needed.
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/topiclistlimit/TopicListSizeResultCache.java:
##########
@@ -116,7 +228,34 @@ public void resetIfInitializing() {
*/
public ResultHolder getTopicListSize(String namespaceName,
CommandGetTopicsOfNamespace.Mode mode) {
+ return getTopicListSize(namespaceName, mode, null);
+ }
+
+ /**
+ * Get the topic list size result holder for the given namespace, mode and topic listing properties.
+ * The properties are passed to a customized topic listing implementation (PIP-452) which can return
+ * a different topic list for different properties. Therefore, the estimate is kept separately for each
+ * distinct set of properties.
+ * @param namespaceName the namespace name in the format of "tenant/namespace"
+ * @param mode the mode of the topic list request (PERSISTENT, NON_PERSISTENT, ALL)
+ * @param properties the topic listing properties, null or empty when not provided
+ * @return the topic list size result holder
+ */
+ public ResultHolder getTopicListSize(String namespaceName,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Map<String, String> properties) {
+ return getTopicListSize(TopicListing.namespaceTopics(namespaceName, mode), properties);
+ }
+
+ /**
+ * Get the topic list size result holder for the given topic listing and topic listing properties.
+ * @param listing the topic listing
+ * @param properties the topic listing properties (PIP-452), null or empty when not provided
+ * @return the topic list size result holder
+ */
+ public ResultHolder getTopicListSize(TopicListing listing, Map<String, String> properties) {
Review Comment:
The properties come from the client, so every distinct property map gets its own `ResultHolder`, and its first `getSizeAsync()` returns the 10 KB initial estimate. With `DefaultPulsarResourcesExtended`, which ignores the properties, a request with a different property value each time lists the whole namespace while only 10 KB is reserved; the list is charged at its actual size only in the update after it has been built. Such requests also don't wait for an earlier estimate. Before this change, the learned (namespace, mode) estimate applied to all of them.
Could a properties-keyed holder start from the estimate of the same listing without properties (e.g. `max(INITIAL_TOPIC_LIST_HEAP_SIZE, <estimate without properties>)`) and refine from there? That would keep separate estimates for filtered listings without letting a new key start below the known size.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #26945:
URL: https://github.com/apache/pulsar/pull/26945
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26945:
URL: https://github.com/apache/pulsar/pull/26945#discussion_r4324460405
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -3720,18 +3722,37 @@ private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName
return null;
});
}, t -> {
- log.warn()
- .exceptionMessage(t)
- .log("Failed to acquire initial heap memory permits for GetTopicsOfNamespace");
listSizeHolder.resetIfInitializing();
- writeAndFlush(Commands.newError(requestId, ServerError.TooManyRequests,
- "Failed due to heap memory limit exceeded"));
+ if (!isTopicListRequestCancelled(t,
+ "Failed to acquire initial heap memory permits for GetTopicsOfNamespace")) {
+ writeAndFlush(Commands.newError(requestId, ServerError.TooManyRequests,
Review Comment:
Good point, done in 4084c3e79be and 5ddbde0b10e. A topic list that has been measured to be larger than the limit now gets `NotAllowedError` with "Failed since the topic list is larger than the topic list heap memory limit" (or "direct" for the response), so the client doesn't retry it. `TooManyRequests` stays for permit timeouts and full queues. This covers the three rejections in `internalHandleGetTopicsOfNamespace`, and the watch request paths in `TopicListService`, so that a pattern consumer's watcher creation fails the same way and falls back to polling.
One exception: a listing with new properties starts from the estimate of the listing without properties, but its own topic list can be smaller (with a `PulsarResourcesExtended` plugin that filters by the properties). Until its own list has been measured, a rejection of that inherited estimate still gets `TooManyRequests` (`ResultHolder.isMeasured()`).
Tests: `testGetTopicsOfNamespaceLargerThanHeapMemoryLimitFailsFastAfterMeasuring` asserts `NotAllowedError`. `testGetTopicsOfNamespacePermitTimeoutIsRetriable` and `testGetTopicsOfNamespaceWithNewPropertiesAndOversizedEstimateIsRetriable` cover the `TooManyRequests` cases. `testDuplicateWatchRequestGetsNonRetriableErrorWhenTopicListIsLargerThanMemoryLimit` covers a waiting watch request; it got the watcher-closed error before, since closing the watcher failed the waiting callbacks first.
##########
pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java:
##########
@@ -412,27 +416,27 @@ private void internalPerformGetTopicsOfNamespace(long clientRequestId, String na
return handleWritingGetTopicsResponse(clientRequestId, r,
isPermitRequestCancelled);
}, t2 -> {
- log.warn()
- .attr("clientAddress", clientAddress)
- .attr("GetTopicsOfNamespace", t2.getMessage())
- .log("Failed to acquire actual heap memory permits for " + "");
- writeAndFlush(
- Commands.newError(clientRequestId, ServerError.TooManyRequests,
- "Failed due to heap memory limit exceeded"));
-
+ if (!isTopicListRequestCancelled(t2,
+ "Failed to acquire actual heap memory permits for "
+ + "GetTopicsOfNamespace")) {
+ writeAndFlush(
+ Commands.newError(clientRequestId,
+ ServerError.TooManyRequests,
+ "Failed due to heap memory limit exceeded"));
+ }
return CompletableFuture.completedFuture(null);
});
}
});
}, t -> {
- log.warn()
- .attr("clientAddress", clientAddress)
- .attr("GetTopicsOfNamespace", t.getMessage())
- .log("Failed to acquire initial heap memory permits for");
+ // the command wasn't sent to the broker, release it
+ ReferenceCountUtil.safeRelease(command);
listSizeHolder.resetIfInitializing();
- writeAndFlush(Commands.newError(clientRequestId, ServerError.TooManyRequests,
- "Failed due to heap memory limit exceeded"));
-
+ if (!isTopicListRequestCancelled(t,
+ "Failed to acquire initial heap memory permits for GetTopicsOfNamespace")) {
+ writeAndFlush(Commands.newError(clientRequestId, ServerError.TooManyRequests,
Review Comment:
Done in 4084c3e79be and 5ddbde0b10e: the proxy's own rejections use `NotAllowedError` for a measured topic list larger than the limit, in the same way as the broker. A `NotAllowedError` from the broker for `GetTopicsOfNamespace` is passed through as `NotAllowedError` instead of `ServiceNotReady`. I limited the pass-through to the topic list response (`getTopicListServerError`), so that the error mapping of lookups, partition metadata and schema requests doesn't change in this PR.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26945:
URL: https://github.com/apache/pulsar/pull/26945#discussion_r4297413573
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/topiclistlimit/TopicListSizeResultCache.java:
##########
@@ -116,7 +228,34 @@ public void resetIfInitializing() {
*/
public ResultHolder getTopicListSize(String namespaceName,
CommandGetTopicsOfNamespace.Mode mode) {
+ return getTopicListSize(namespaceName, mode, null);
+ }
+
+ /**
+ * Get the topic list size result holder for the given namespace, mode and topic listing properties.
+ * The properties are passed to a customized topic listing implementation (PIP-452) which can return
+ * a different topic list for different properties. Therefore, the estimate is kept separately for each
+ * distinct set of properties.
+ * @param namespaceName the namespace name in the format of "tenant/namespace"
+ * @param mode the mode of the topic list request (PERSISTENT, NON_PERSISTENT, ALL)
+ * @param properties the topic listing properties, null or empty when not provided
+ * @return the topic list size result holder
+ */
+ public ResultHolder getTopicListSize(String namespaceName,
+ CommandGetTopicsOfNamespace.Mode mode,
+ Map<String, String> properties) {
+ return getTopicListSize(TopicListing.namespaceTopics(namespaceName, mode), properties);
+ }
+
+ /**
+ * Get the topic list size result holder for the given topic listing and topic listing properties.
+ * @param listing the topic listing
+ * @param properties the topic listing properties (PIP-452), null or empty when not provided
+ * @return the topic list size result holder
+ */
+ public ResultHolder getTopicListSize(TopicListing listing, Map<String, String> properties) {
Review Comment:
Done in 27b3538fefd: a listing with a new set of properties now starts from `max(initial estimate, known estimate of the same listing without properties)`, so new property values can't start below the known size. Properties-only traffic for a namespace that has never been listed without properties still starts at the initial estimate, which is the same exposure as the first listing of any namespace. Covered by `TopicListSizeResultCacheTest.testNewPropertiesStartFromEstimateWithoutProperties`.
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/topiclistlimit/TopicListSizeResultCache.java:
##########
@@ -54,23 +130,59 @@ record CacheKey(String namespaceName, CommandGetTopicsOfNamespace.Mode mode) {
public static class ResultHolder {
private final AtomicReference<CompletableFuture<Long>> topicListSizeFuture =
new AtomicReference<>(null);
+ private final long sizeLimit;
+ private final long oversizedRemeasureIntervalNanos;
+ // the earliest time when a request for an oversized listing may measure the topic list again
+ private final AtomicLong nextOversizedRemeasureNanos;
+
+ public ResultHolder() {
+ this(Long.MAX_VALUE, 0);
+ }
+
+ ResultHolder(long sizeLimit, long oversizedRemeasureIntervalNanos) {
+ this.sizeLimit = sizeLimit;
+ this.oversizedRemeasureIntervalNanos = oversizedRemeasureIntervalNanos;
+ this.nextOversizedRemeasureNanos = new AtomicLong(System.nanoTime());
+ }
/**
* Get the topic list size estimate. The first request will return the initial estimate
* and update the estimate based on the returned size of the topic list. Other concurrent requests
* will wait for the first request to complete and use the estimate of the first request.
* Subsequent requests will use estimate which gets updated based on the returned size of the topic list
* of each request.
+ * When the estimate is larger than the size limit, it's returned as-is so that the request fails without
+ * waiting for permits, except for one request per interval, which gets the size limit so that it can
+ * measure the topic list again.
* @return a future that will return the topic list size estimate
*/
public CompletableFuture<Long> getSizeAsync() {
+ CompletableFuture<Long> estimate;
if (topicListSizeFuture.compareAndSet(null, new CompletableFuture<>())) {
// let the first request proceed with the initial estimate
- return CompletableFuture.completedFuture(INITIAL_TOPIC_LIST_HEAP_SIZE);
+ estimate = CompletableFuture.completedFuture(Math.min(INITIAL_TOPIC_LIST_HEAP_SIZE, sizeLimit));
} else {
// all other requests wait for the first one to complete
- return topicListSizeFuture.get();
+ estimate = topicListSizeFuture.get();
}
+ return sizeLimit == Long.MAX_VALUE ? estimate : estimate.thenApply(this::limitOversizedEstimate);
+ }
+
+ private long limitOversizedEstimate(long estimate) {
+ if (estimate <= sizeLimit) {
+ return estimate;
+ }
+ long now = System.nanoTime();
+ long next = nextOversizedRemeasureNanos.get();
+ if (now - next >= 0
+ && nextOversizedRemeasureNanos.compareAndSet(next, now + oversizedRemeasureIntervalNanos)) {
+ // a request that can be granted, so that the topic list gets measured again
+ return sizeLimit;
Review Comment:
Agreed that requesting the whole limit while other listings are in flight is a problem, but I'd rather not let the oversized estimate lapse to the initial estimate: that would list a namespace that is already known to be larger than the limit with only 10 KB reserved, once per interval and per listing key, which brings back the unbounded allocation the limiter is there to prevent.
In 27b3538fefd the re-measure keeps the full reservation, but it no longer makes other listings wait behind it:
- it's only granted while the heap limiter is idle (no acquired permits and no queued requests), so the request doesn't queue behind in-flight listings, and the other requests fail fast while it runs, instead of queuing behind it
- only one listing is re-measured at a time on the broker (a gate that the re-measuring listing releases when it records its size, with a timeout in case the request fails before that)
- so that a broker that is never idle can still recover, the re-measure is granted regardless of the limiter state 10 minutes after the oversized measurement
The idle check and the permit request aren't atomic, so another listing can still acquire permits in between; that window is short. Closing it would need a non-queuing acquire in `AsyncSemaphoreImpl`, which I'd rather do after #26929 is in since it rewrites that class.
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/topiclistlimit/TopicListSizeResultCache.java:
##########
@@ -54,23 +130,59 @@ record CacheKey(String namespaceName, CommandGetTopicsOfNamespace.Mode mode) {
public static class ResultHolder {
private final AtomicReference<CompletableFuture<Long>> topicListSizeFuture =
new AtomicReference<>(null);
+ private final long sizeLimit;
+ private final long oversizedRemeasureIntervalNanos;
+ // the earliest time when a request for an oversized listing may measure the topic list again
+ private final AtomicLong nextOversizedRemeasureNanos;
+
+ public ResultHolder() {
+ this(Long.MAX_VALUE, 0);
+ }
+
+ ResultHolder(long sizeLimit, long oversizedRemeasureIntervalNanos) {
+ this.sizeLimit = sizeLimit;
+ this.oversizedRemeasureIntervalNanos = oversizedRemeasureIntervalNanos;
+ this.nextOversizedRemeasureNanos = new AtomicLong(System.nanoTime());
Review Comment:
Done in 27b3538fefd: the deadline is now set when `updateSize()` records a size larger than the limit, so the first re-measure happens one interval after the oversized measurement. `testOversizedEstimateFailsFastUntilIntervalHasPassed` covers it.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -186,76 +180,35 @@ protected CompletableFuture<List<String>> internalGetListOfTopics(AsyncResponse
CommandGetTopicsOfNamespace.Mode mode) {
// Use maxTopicListInFlightLimiter to limit inflight get topic listing responses
// to avoid OOME caused by a lot of clients using HTTP service lookups to list topics
- AsyncDualMemoryLimiterImpl maxTopicListInFlightLimiter =
- pulsar().getBrokerService().getMaxTopicListInFlightLimiter();
- TopicListSizeResultCache.ResultHolder listSizeHolder = pulsar().getBrokerService().getTopicListSizeResultCache()
- .getTopicListSize(namespaceName.toString(), mode);
- // setup the permit cancellation function
- AtomicBoolean permitRequestCancelled = new AtomicBoolean(false);
- BooleanSupplier isPermitRequestCancelled = permitRequestCancelled::get;
- // add callback that releases permits when the response completes
- AtomicReference<AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit> initialPermitsRef =
- new AtomicReference<>();
- AtomicReference<AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit> permitsRef = new AtomicReference<>();
- response.register(new CompletionCallback() {
- @Override
- public void onComplete(Throwable throwable) {
- if (throwable != null) {
- // for failed request
- // handle resetting the TopicListSizeResultCache.ResultHolder
- listSizeHolder.resetIfInitializing();
- // cancel any pending permit request
- permitRequestCancelled.set(true);
- }
- AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit initialPermit = initialPermitsRef.get();
- if (initialPermit != null) {
- maxTopicListInFlightLimiter.release(initialPermit);
- }
- AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit permits = permitsRef.get();
- if (permits != null) {
- maxTopicListInFlightLimiter.release(permits);
- }
- }
- });
- return listSizeHolder.getSizeAsync().thenCompose(initialSize -> maxTopicListInFlightLimiter.acquire(initialSize,
- AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled).exceptionally(t -> {
- throw new CompletionException(
- new RestException(Status.TOO_MANY_REQUESTS, "Failed due to heap memory limit exceeded"));
- }).thenCompose(initialPermits -> {
- initialPermitsRef.set(initialPermits);
- // perform the actual get list of topics operation
- return doInternalGetListOfTopics(policies, mode).thenCompose(topicList -> {
- long actualSize = TopicListMemoryLimiter.estimateTopicListSize(topicList);
- listSizeHolder.updateSize(actualSize);
- return maxTopicListInFlightLimiter.update(initialPermits, actualSize, isPermitRequestCancelled)
- .exceptionally(t -> {
- throw new CompletionException(new RestException(Status.TOO_MANY_REQUESTS,
- "Failed due to heap memory limit exceeded"));
- }).thenApply(permits -> {
- permitsRef.set(permits);
- initialPermitsRef.set(null);
- return topicList;
- });
- });
- }));
- }
-
- private CompletableFuture<List<String>> doInternalGetListOfTopics(Policies policies,
- CommandGetTopicsOfNamespace.Mode mode) {
switch (mode) {
case ALL:
- return pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName)
- .thenCombine(internalGetNonPersistentTopics(policies),
- (persistentTopics, nonPersistentTopics) ->
- ListUtils.union(persistentTopics, nonPersistentTopics));
+ // the persistent topics are listed while holding permits for them, the non-persistent topics are
+ // collected afterwards without holding additional permits while waiting for the bundle listings
+ return listPersistentTopicsWithMemoryLimit(response)
Review Comment:
Good catch. In 27b3538fefd the `ALL` mode collects the non-persistent topics first with no permits held, and then acquires permits once for the whole topic list (persistent and non-persistent), with the estimate of the `ALL` listing. It never holds permits for one part while waiting for permits for the other part or for the bundle listings, so the hold-and-wait is gone and the `ALL` listing can also be re-measured when it's oversized. I'd rather not leave the bundle listing unlimited, since it's also a public endpoint. `testAllModeDoesNotHoldPermitsWhileWaitingForBundleListings` now uses a persistent topic list of 60% of the limit and a bundle listing estimate of 60% of the limit.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java:
##########
@@ -702,14 +817,16 @@ private void performOperationWithPermitAcquiringRetries(long watcherId, String o
AtomicReference<Supplier<CompletableFuture<Void>>> operationRef = new AtomicReference<>();
// create the error handler for the operation
Function<Throwable, CompletableFuture<Void>> permitAcquireErrorHandler =
- createPermitAcquireErrorHandler(watcherId, operationName, () -> operationRef.get().get());
+ createPermitAcquireErrorHandler(watcherId, operationName, failOnCancellation,
+ () -> operationRef.get().get());
// create the async operation using the factory function. Pass the error handler to the factory function.
Supplier<CompletableFuture<Void>> asyncOperation = asyncOperationFactory.apply(permitAcquireErrorHandler);
// set the operation to run into the operation reference
operationRef.set(() -> {
if (!connection.isActive() || !watchers.containsKey(watcherId)) {
// do nothing if the connection has already been closed or the watcher has been removed
- return CompletableFuture.completedFuture(null);
+ return failOnCancellation ? CompletableFuture.failedFuture(newWatcherClosedException(watcherId))
Review Comment:
Fixed in 27b3538fefd: `performOperationWithPermitAcquiringRetries` returns the future of the whole operation and the completion is handled on it, so the failed first attempt gets an error response. This also fixed the completion callback running twice after a retry (once for the first attempt's future, which completes after the retry, and once for the retried attempt). While writing the test I found that `updateTopics()` ran the waiting requests' callbacks while iterating `updateCallbacks` under the watcher lock, and a callback that ends up closing the watcher modified the list (`ConcurrentModificationException`). The callbacks now run outside the lock with a snapshot of the matching topics taken under the lock, and a new watcher's response waits for an update that started after it was created, so a response never contains the topics while they're being refreshed. Covered by `testWatchRequestGetsErrorWhenWatcherIsRemovedBeforeFirstSendAttempt` and `testWaitingWatchRequestRespondsW
ithSnapshotOfItsUpdate`.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java:
##########
@@ -135,7 +135,7 @@ public void getList(
@Parameter(description = "properties for customized topic listing plugin, format: k1=v1,k2=v2")
@QueryParam("properties") String propertiesStr) {
validateNamespaceName(tenant, namespace);
- internalGetListAsync(Optional.ofNullable(bundle), parseProperties(propertiesStr))
+ internalGetListAsync(asyncResponse, Optional.ofNullable(bundle), parseProperties(propertiesStr))
Review Comment:
Done in 27b3538fefd: the topic listing endpoints (this one, `getPartitionedTopicList`, the non-persistent, bundle and scalable listings and `Namespaces.getTopics`) no longer log 429 and the cancellation 503 at ERROR level.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #26945:
URL: https://github.com/apache/pulsar/pull/26945#discussion_r4296926761
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -186,76 +180,35 @@ protected CompletableFuture<List<String>> internalGetListOfTopics(AsyncResponse
CommandGetTopicsOfNamespace.Mode mode) {
// Use maxTopicListInFlightLimiter to limit inflight get topic listing responses
// to avoid OOME caused by a lot of clients using HTTP service lookups to list topics
- AsyncDualMemoryLimiterImpl maxTopicListInFlightLimiter =
- pulsar().getBrokerService().getMaxTopicListInFlightLimiter();
- TopicListSizeResultCache.ResultHolder listSizeHolder = pulsar().getBrokerService().getTopicListSizeResultCache()
- .getTopicListSize(namespaceName.toString(), mode);
- // setup the permit cancellation function
- AtomicBoolean permitRequestCancelled = new AtomicBoolean(false);
- BooleanSupplier isPermitRequestCancelled = permitRequestCancelled::get;
- // add callback that releases permits when the response completes
- AtomicReference<AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit> initialPermitsRef =
- new AtomicReference<>();
- AtomicReference<AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit> permitsRef = new AtomicReference<>();
- response.register(new CompletionCallback() {
- @Override
- public void onComplete(Throwable throwable) {
- if (throwable != null) {
- // for failed request
- // handle resetting the TopicListSizeResultCache.ResultHolder
- listSizeHolder.resetIfInitializing();
- // cancel any pending permit request
- permitRequestCancelled.set(true);
- }
- AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit initialPermit = initialPermitsRef.get();
- if (initialPermit != null) {
- maxTopicListInFlightLimiter.release(initialPermit);
- }
- AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit permits = permitsRef.get();
- if (permits != null) {
- maxTopicListInFlightLimiter.release(permits);
- }
- }
- });
- return listSizeHolder.getSizeAsync().thenCompose(initialSize -> maxTopicListInFlightLimiter.acquire(initialSize,
- AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled).exceptionally(t -> {
- throw new CompletionException(
- new RestException(Status.TOO_MANY_REQUESTS, "Failed due to heap memory limit exceeded"));
- }).thenCompose(initialPermits -> {
- initialPermitsRef.set(initialPermits);
- // perform the actual get list of topics operation
- return doInternalGetListOfTopics(policies, mode).thenCompose(topicList -> {
- long actualSize = TopicListMemoryLimiter.estimateTopicListSize(topicList);
- listSizeHolder.updateSize(actualSize);
- return maxTopicListInFlightLimiter.update(initialPermits, actualSize, isPermitRequestCancelled)
- .exceptionally(t -> {
- throw new CompletionException(new RestException(Status.TOO_MANY_REQUESTS,
- "Failed due to heap memory limit exceeded"));
- }).thenApply(permits -> {
- permitsRef.set(permits);
- initialPermitsRef.set(null);
- return topicList;
- });
- });
- }));
- }
-
- private CompletableFuture<List<String>> doInternalGetListOfTopics(Policies policies,
- CommandGetTopicsOfNamespace.Mode mode) {
switch (mode) {
case ALL:
- return pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName)
- .thenCombine(internalGetNonPersistentTopics(policies),
- (persistentTopics, nonPersistentTopics) ->
- ListUtils.union(persistentTopics, nonPersistentTopics));
+ // the persistent topics are listed while holding permits for them, the non-persistent topics are
+ // collected afterwards without holding additional permits while waiting for the bundle listings
+ return listPersistentTopicsWithMemoryLimit(response)
Review Comment:
In the `ALL` mode, the permits of the persistent topic listing are held until the response completes, so they're still held while `listNonPersistentTopicsWithMemoryLimit` waits for the bundle listings. With this PR, those (`NonPersistentTopics.getListFromBundle`, called through `getListInBundleAsync`) acquire permits from the same limiter, on this broker when it owns the bundle. When the limiter is saturated, a bundle listing request waits behind a head request that doesn't fit, while the outer requests that hold permits can't complete until their bundle listings time out. That's the hold-and-wait that `listTopicsFromOtherListings` avoids for `NON_PERSISTENT`. `testNonPersistentListingsDontWaitForTheirOwnPermits` passes because the permits are updated to the actual persistent list, which is tiny there, before the bundle listings run. `HttpLookupService` uses `mode=ALL` for pattern consumers with `RegexSubscriptionMode.AllTopics`.
Could the `ALL` mode collect the non-persistent topics first (holding no permits) and list the persistent topics after that? Alternatively, the bundle listing, which reads the in-memory topics map, could stay unlimited.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26945:
URL: https://github.com/apache/pulsar/pull/26945#discussion_r4298343210
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java:
##########
@@ -418,47 +501,104 @@ public void handleWatchTopicList(NamespaceName namespaceName, long watcherId, lo
.attr("requestId", requestId)
.log("Received WatchTopicList");
sendTopicListSuccessWithPermitAcquiringRetries(watcherId, requestId, topicList, hash,
- watcher::sendingCompleted, watcher::close);
+ watcher::sendingCompleted, failure -> {
+ if (!isCancellation(failure)) {
+ // the watcher can't be used without the success response
+ deleteTopicListWatcher(watcherId);
+ }
+ // the request is waiting for a response unless the connection was closed
+ if (connection.isActive()) {
+ ServerError error =
+ failure instanceof AsyncSemaphore.PermitAcquireLimitExceededException
+ ? ServerError.TooManyRequests
+ : BrokerServiceException.getClientErrorCode(
+ new BrokerServiceException.ServerMetadataException(
+ failure));
+ connection.getCommandSender().sendErrorResponse(requestId, error,
+ failure.getMessage());
+ }
+ });
lookupSemaphore.release();
})
.exceptionally(ex -> {
- log.warn()
- .attr("connection", connection)
- .attr("namespace", namespaceName)
- .attr("requestId", requestId)
- .exceptionMessage(ex)
- .log("Error WatchTopicList");
- connection.getCommandSender().sendErrorResponse(requestId,
- BrokerServiceException.getClientErrorCode(
- new BrokerServiceException.ServerMetadataException(ex)), ex.getMessage());
- watchers.remove(watcherId, finalWatcherFuture);
+ if (connection.isActive()) {
+ log.warn()
+ .attr("connection", connection)
+ .attr("namespace", namespaceName)
+ .attr("requestId", requestId)
+ .exceptionMessage(ex)
+ .log("Error WatchTopicList");
+ connection.getCommandSender().sendErrorResponse(requestId,
+ BrokerServiceException.getClientErrorCode(
+ new BrokerServiceException.ServerMetadataException(ex)), ex.getMessage());
+ } else {
+ // the connection was closed, there's no client to respond to
+ log.debug()
+ .attr("connection", connection)
+ .attr("namespace", namespaceName)
+ .attr("requestId", requestId)
+ .exceptionMessage(ex)
+ .log("WatchTopicList was cancelled since the connection was closed");
+ }
+ if (createdWatcherFuture != null) {
+ watchers.remove(watcherId, createdWatcherFuture);
Review Comment:
Good catch, thanks. Fixed in 2e1d0c1a2c6 as you suggested: when the failed request removes the watcher that it created, it also deregisters the listener and closes the watcher. `testFailedWatchRequestClosesTheWatcherItCreated` starts an update (a `ConnectionLost` session event) while the watcher is being created, fails it, and checks the error response, the lookup permit, the map entry and the deregistration.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicListService.java:
##########
@@ -208,29 +278,32 @@ synchronized boolean prepareUpdateTopics(Runnable afterUpdateCompletionCallback)
}
}
- synchronized void updateTopics(List<String> topics) {
- if (closed) {
- return;
- }
- matchingTopics.clear();
- TopicList.filterTopicsToStream(topics, topicsPattern).forEach(matchingTopics::add);
- updatingTopics = false;
- if (disconnected) {
- handleNewAndDeletedTopicsWhileDisconnected();
- matchingTopicsBeforeDisconnected = null;
- disconnected = false;
- }
- for (Runnable callback : updateCallbacks) {
- try {
- callback.run();
- } catch (Exception e) {
- log.warn()
- .attr("callback", callback)
- .exception(e)
- .log("Error executing topic list update callback");
+ void updateTopics(List<String> topics) {
+ List<BiConsumer<Collection<String>, Throwable>> callbacks;
+ Collection<String> snapshot;
+ synchronized (this) {
+ if (closed) {
+ return;
+ }
+ matchingTopics.clear();
+ TopicList.filterTopicsToStream(topics, topicsPattern).forEach(matchingTopics::add);
+ updatingTopics = false;
+ if (disconnected) {
+ handleNewAndDeletedTopicsWhileDisconnected();
+ matchingTopicsBeforeDisconnected = null;
+ disconnected = false;
}
+ callbacks = new ArrayList<>(updateCallbacks);
+ updateCallbacks.clear();
+ // the waiting requests respond with the result of this update, also if the matching topics change
+ // after the lock has been released, for example when another update starts
+ snapshot = callbacks.isEmpty() ? null : new ArrayList<>(matchingTopics);
+ }
+ // the callbacks run without holding the lock, since a callback can close the watcher, for example when
+ // sending the response of a waiting request fails
+ for (BiConsumer<Collection<String>, Throwable> callback : callbacks) {
+ runUpdateCallback(callback, snapshot, null);
}
- updateCallbacks.clear();
sendingCompleted();
Review Comment:
Right, thanks. Fixed in 2e1d0c1a2c6: `sendingCompleted()` returns early while `updatingTopics` is true, and the update calls it again when it completes. `testTopicChangesAreQueuedWhileUpdateIsInProgress` covers a send that completes during an update.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -186,76 +180,40 @@ protected CompletableFuture<List<String>> internalGetListOfTopics(AsyncResponse
CommandGetTopicsOfNamespace.Mode mode) {
// Use maxTopicListInFlightLimiter to limit inflight get topic listing responses
// to avoid OOME caused by a lot of clients using HTTP service lookups to list topics
- AsyncDualMemoryLimiterImpl maxTopicListInFlightLimiter =
- pulsar().getBrokerService().getMaxTopicListInFlightLimiter();
- TopicListSizeResultCache.ResultHolder listSizeHolder = pulsar().getBrokerService().getTopicListSizeResultCache()
- .getTopicListSize(namespaceName.toString(), mode);
- // setup the permit cancellation function
- AtomicBoolean permitRequestCancelled = new AtomicBoolean(false);
- BooleanSupplier isPermitRequestCancelled = permitRequestCancelled::get;
- // add callback that releases permits when the response completes
- AtomicReference<AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit> initialPermitsRef =
- new AtomicReference<>();
- AtomicReference<AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit> permitsRef = new AtomicReference<>();
- response.register(new CompletionCallback() {
- @Override
- public void onComplete(Throwable throwable) {
- if (throwable != null) {
- // for failed request
- // handle resetting the TopicListSizeResultCache.ResultHolder
- listSizeHolder.resetIfInitializing();
- // cancel any pending permit request
- permitRequestCancelled.set(true);
- }
- AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit initialPermit = initialPermitsRef.get();
- if (initialPermit != null) {
- maxTopicListInFlightLimiter.release(initialPermit);
- }
- AsyncDualMemoryLimiter.AsyncDualMemoryLimiterPermit permits = permitsRef.get();
- if (permits != null) {
- maxTopicListInFlightLimiter.release(permits);
- }
- }
- });
- return listSizeHolder.getSizeAsync().thenCompose(initialSize -> maxTopicListInFlightLimiter.acquire(initialSize,
- AsyncDualMemoryLimiter.LimitType.HEAP_MEMORY, isPermitRequestCancelled).exceptionally(t -> {
- throw new CompletionException(
- new RestException(Status.TOO_MANY_REQUESTS, "Failed due to heap memory limit exceeded"));
- }).thenCompose(initialPermits -> {
- initialPermitsRef.set(initialPermits);
- // perform the actual get list of topics operation
- return doInternalGetListOfTopics(policies, mode).thenCompose(topicList -> {
- long actualSize = TopicListMemoryLimiter.estimateTopicListSize(topicList);
- listSizeHolder.updateSize(actualSize);
- return maxTopicListInFlightLimiter.update(initialPermits, actualSize, isPermitRequestCancelled)
- .exceptionally(t -> {
- throw new CompletionException(new RestException(Status.TOO_MANY_REQUESTS,
- "Failed due to heap memory limit exceeded"));
- }).thenApply(permits -> {
- permitsRef.set(permits);
- initialPermitsRef.set(null);
- return topicList;
- });
- });
- }));
- }
-
- private CompletableFuture<List<String>> doInternalGetListOfTopics(Policies policies,
- CommandGetTopicsOfNamespace.Mode mode) {
switch (mode) {
case ALL:
- return pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName)
- .thenCombine(internalGetNonPersistentTopics(policies),
- (persistentTopics, nonPersistentTopics) ->
- ListUtils.union(persistentTopics, nonPersistentTopics));
+ // The non-persistent topics are collected first from the bundle listings, which acquire permits
+ // from the same limiter, possibly on this broker, so no permits are held while waiting for them.
+ // The listing then acquires permits once, for the whole topic list, with the estimate of the ALL
+ // listing, so that it doesn't hold permits for one part while waiting for permits for the other.
+ return internalGetNonPersistentTopics(policies)
Review Comment:
Done in 2e1d0c1a2c6: the `ALL` listing now gets its size estimate before collecting, and an estimate larger than the limit fails right away without collecting (a re-measuring request still collects). The permits are still acquired only after collecting. This is in `TopicListResponseMemoryLimiter.listTopicsAfterCollecting`, covered by `testListTopicsAfterCollectingSkipsCollectingWhenEstimateIsLargerThanLimit`.
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/topiclistlimit/TopicListSizeResultCache.java:
##########
@@ -54,23 +177,70 @@ record CacheKey(String namespaceName, CommandGetTopicsOfNamespace.Mode mode) {
public static class ResultHolder {
private final AtomicReference<CompletableFuture<Long>> topicListSizeFuture =
new AtomicReference<>(null);
+ private final OversizedRemeasurePolicy policy;
+ private final long sizeLimit;
+ private final long initialEstimate;
+ // the earliest time when a request for an oversized listing may measure the topic list again
+ private volatile long nextRemeasureNanos;
+ // the earliest time when the topic list is measured again even if the limiter is busy
+ private volatile long nextForcedRemeasureNanos;
+
+ public ResultHolder() {
+ this(OversizedRemeasurePolicy.UNLIMITED, 0);
+ }
+
+ private ResultHolder(OversizedRemeasurePolicy policy, long knownSizeEstimate) {
+ this.policy = policy;
+ this.sizeLimit = policy.sizeLimit();
+ this.initialEstimate = Math.max(Math.min(INITIAL_TOPIC_LIST_HEAP_SIZE, sizeLimit), knownSizeEstimate);
+ long now = System.nanoTime();
+ // a listing that starts from an oversized known estimate hasn't been measured yet
+ this.nextRemeasureNanos = now;
Review Comment:
Agreed, done in 2e1d0c1a2c6: forced re-measures (the ones that don't wait for an idle limiter) are now limited for the whole broker, at most one per forced interval (10 minutes), in addition to the per-listing schedule and the gate that allows one re-measure at a time. Re-measures while the limiter is idle aren't limited by it, since they don't make other listings wait. `testForcedRemeasuresAreLimitedForTheWholeCache` covers it with two properties-keyed listings.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix topic listing memory limit enforcement [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #26945:
URL: https://github.com/apache/pulsar/pull/26945#discussion_r4321554635
##########
pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java:
##########
@@ -412,27 +416,27 @@ private void internalPerformGetTopicsOfNamespace(long clientRequestId, String na
return handleWritingGetTopicsResponse(clientRequestId, r,
isPermitRequestCancelled);
}, t2 -> {
- log.warn()
- .attr("clientAddress", clientAddress)
- .attr("GetTopicsOfNamespace", t2.getMessage())
- .log("Failed to acquire actual heap memory permits for " + "");
- writeAndFlush(
- Commands.newError(clientRequestId, ServerError.TooManyRequests,
- "Failed due to heap memory limit exceeded"));
-
+ if (!isTopicListRequestCancelled(t2,
+ "Failed to acquire actual heap memory permits for "
+ + "GetTopicsOfNamespace")) {
+ writeAndFlush(
+ Commands.newError(clientRequestId,
+ ServerError.TooManyRequests,
+ "Failed due to heap memory limit exceeded"));
+ }
return CompletableFuture.completedFuture(null);
});
}
});
}, t -> {
- log.warn()
- .attr("clientAddress", clientAddress)
- .attr("GetTopicsOfNamespace", t.getMessage())
- .log("Failed to acquire initial heap memory permits for");
+ // the command wasn't sent to the broker, release it
+ ReferenceCountUtil.safeRelease(command);
listSizeHolder.resetIfInitializing();
- writeAndFlush(Commands.newError(clientRequestId, ServerError.TooManyRequests,
- "Failed due to heap memory limit exceeded"));
-
+ if (!isTopicListRequestCancelled(t,
+ "Failed to acquire initial heap memory permits for GetTopicsOfNamespace")) {
+ writeAndFlush(Commands.newError(clientRequestId, ServerError.TooManyRequests,
Review Comment:
Same as my comment on `ServerCnx` (non-blocking): this rejection and the one at the permit update above could use a non-retriable error for `PermitAcquireLimitExceededException`. A rejection from the broker is relayed through `getServerError`, which maps everything except authentication and authorization errors to `ServiceNotReady`, and the client retries that too. So the proxy would also need to pass a non-retriable broker error through, for example `NotAllowedException` as `NotAllowedError`.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -3720,18 +3722,37 @@ private void internalHandleGetTopicsOfNamespace(String namespace, NamespaceName
return null;
});
}, t -> {
- log.warn()
- .exceptionMessage(t)
- .log("Failed to acquire initial heap memory permits for GetTopicsOfNamespace");
listSizeHolder.resetIfInitializing();
- writeAndFlush(Commands.newError(requestId, ServerError.TooManyRequests,
- "Failed due to heap memory limit exceeded"));
+ if (!isTopicListRequestCancelled(t,
+ "Failed to acquire initial heap memory permits for GetTopicsOfNamespace")) {
+ writeAndFlush(Commands.newError(requestId, ServerError.TooManyRequests,
Review Comment:
Non-blocking, fine as a follow-up: since #24974, which this branch now includes, `BinaryProtoLookupService.getTopicsUnderNamespace` retries every error that `PulsarClientException.isRetriableError` accepts until the operation timeout, and `TooManyRequestsException` is retriable. That's right for a permit timeout or a full queue, but not for a topic list that is larger than the limit (`PermitAcquireLimitExceededException`): the list is measured again one minute after the oversized measurement at the earliest, so a retry within the operation timeout almost always fails the same way. With the defaults, the client sends about 10 requests over 30 s and then fails with a generic `TimeoutException`, and the broker logs a WARN line for each request. A pattern consumer for such a namespace hits this when it subscribes, and on every recheck, since its topic list watcher can't be created either.
It's not a regression, since such a request got no response at all before this PR, but this PR defines the response for this case. Could the limit-exceeded case get a non-retriable error with a specific message, for example `ServerError.NotAllowedError` with "the topic list is larger than the topic list heap memory limit" like the REST path's message, while `TooManyRequests` stays for permit timeouts and full queues? The same applies to the other `PermitAcquireLimitExceededException` rejections in this method, such as the one at the permit update above, which rejects the first oversized measurement.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org