Resource Groups (PIP-82): status of tenant-level attachment and dispatch rate enforcement

[Goodkat]

Context

PIP-82: Tenant and Namespace Level Rate Limiting introduced Resource Groups as a way to apply rate limits across all topics of a tenant or namespace, no matter how many topics it contains. The proposal covers:

• attaching a Resource Group to a tenant or to a namespace;
• limits for publish rates (msg/s, bytes/s) and dispatch rates (msg/s, bytes/s);
• sharing usage between brokers through a non-persistent system topic, so the limit is enforced approximately across the cluster.

As far as I can tell from the current code base and documentation, the actual implementation is smaller:

1. Resource Groups can be attached to namespaces only. There is no admin API/CLI for attaching them to a tenant, although PIP-82 says the tenant APIs would mirror the namespace ones.
2. Only publish rate limits are enforced. The dispatchRateInMsgs / dispatchRateInBytes fields exist in the Resource Group configuration (and dispatch usage may be reported), but I could not find a dispatch-side limiter that applies them to consumers.

If I've missed something, for example a later PR or PIP that covers this, please point me to it.

Questions

1. Why were tenant-level attachment and dispatch-rate enforcement not implemented? Were there technical blockers (e.g. how a namespace-wide dispatch limiter would interact with the existing per-topic/per-subscription DispatchRateLimiter, or with the rate limiter refactoring in PIP-322), or was the work simply never finished? Is there an open issue or a roadmap item for it?

2. Geo-replication. PIP-82 explicitly says that Resource Group limits will not be enforced on cross-cluster replication. Is that still the intended behavior? Specifically:

   • Is replicator traffic (the replicator reading from the source cluster and producing to the remote cluster) excluded from the Resource Group publish accounting today? And would it be excluded from dispatch accounting too if dispatch limits were added?
   • On the remote cluster, are messages written by the replicator counted against the publish quota of the namespace's Resource Group? If they are, could a tight publish limit throttle replication and grow the replication backlog?
   • Should Resource Groups get a separate replication quota, or should replication stay governed only by the existing replicatorDispatchRate policies?

3. Dispatch limits at namespace (and tenant) level. We need an aggregate dispatch rate limit per namespace at minimum, and ideally per tenant, across all topics and subscriptions. The existing namespace dispatchRate / subscriptionDispatchRate policies are applied per topic/subscription, not as a namespace-wide total. Would the community accept a contribution that implements dispatch-rate that still the intended behavior? Specifically:

   • Is replicator traffic (the replicator reading from the source cluster and producing to the remote cluster) excluded from the Resource Group publish accounting today? And would it be excluded from dispatch accounting too if dispatch limits were added?
   • On the remote cluster, are messages written by the replicator counted against the publish quota of the namespace's Resource Group? If they are, could a tight publish limit throttle replication and grow the replication backlog?
   • Should Resource Groups get a separate replication quota, or should replication stay governed only by the existing replicatorDispatchRate policies?

Our use case

We run a shared, multi-tenant Pulsar cluster and need to keep specific tenants within publish and dispatch rate quotas.

Today we can't cap a tenant's total throughput, so from time to time one or more tenants put heavy load on the cluster. When this happens:

• the cluster becomes unstable, and in the worst cases it goes down;
• other tenants are affected: their clients fail to establish connections and report performance degradation, even though their own traffic hasn't changed.

The per-topic and per-subscription limits that exist today don't solve this. A tenant can spread its load over many topics, subscriptions and namespaces, so no single limit is reached while the total load still overwhelms the brokers and bookies. What we need is aggregate enforcement: ideally per tenant, as PIP-82 describes, and at least per namespace as a first step, for both publish and dispatch.

Environment: Pulsar version 4.0.14, geo-replication: enabled between 2 clusters.

Thanks!

Sources:

• PIP-82: Tenant and Namespace Level Rate Limiting
• PIP-322: Pulsar Rate Limiting Refactoring
• PIP-3: Message dispatch throttling

[lhotari]

This is a broad question to answer. Resource Groups don't really solve the complete problem of QoS in Pulsar.

Current state

The work to improve QoS in Pulsar has been delayed. The Pulsar 4.0 announcement blog post contains some context about the planned future improvements. In that blog post, the assumption was that admission control could be handled primarily with rate limits, as is done with the existing PIP-82. However, a rate-limit-based approach alone cannot solve the challenges.

The next detail to address is the lack of produce backpressure in Pulsar. I have a WIP PIP document at https://github.com/lhotari/pulsar/blob/lh-produce-backpressure/pip/pip-produce-backpressure-WIP.md, which has a lot of background context and details. It's Claude-generated with human assistance, so due to its length it isn't really readable.

Pulsar (including 5.0.0) also contains some areas that perform poorly today: tiered storage and topic compaction. Those are resource-heavy, and there's potential to reduce the load on the cluster significantly once they have been optimized. Topic compaction also isn't currently fully covered by a backpressure solution, and therefore it can cause broker crashes.

Future state

Adapting to the actual capacity, for example by inspecting queue sizes and service time, and implementing adaptive limits is how the available capacity can be shared. Dynamically adjusted rate limiters can be used as part of this.

The first goal would be to have "adaptive throttling", so that the Pulsar broker wouldn't crash under load. This requires "backpressure", which is about flow control.

It would be possible to improve PIP-82's rate limiting so that the cluster-wide tenant-level and dispatch limits would also be enforced. However, since that anyway requires solving backpressure and preventing broker overload to avoid crashes, backpressure should be fixed first. Fixing backpressure also makes it possible to use the available infrastructure resources most efficiently, without needing to provision for peak workloads, since there would be ways to "smooth the peaks" (as the batch job example below shows).

Once proper admission control has been implemented, it should be possible to set weights for tenants and for different workloads. For batch-job-like consumers and producers, low latency isn't a requirement, and throughput isn't a hard requirement either, as long as the flow control can adjust to the available capacity. The produce backpressure WIP document contains details about why this isn't currently the case. For real-time use cases, there should be a way to prioritize the workload and also to start shedding traffic by sampling or dropping data when producing cannot keep up. Some real-time sources cannot stop the data from flowing, and the only option is to drop some data if the producer cannot keep up. There's currently no easy way to implement this with Pulsar (it's possible, but it requires additional solutions on the client side).

Pulsar 5.0 has significant performance improvements, and more are coming. Coding agents can drive the Pulsar Performance Testing Framework with the AGENTS.md instructions and create automated experiments to make improvements. This is an extremely powerful tool for continuing to improve Pulsar's performance.

Mitigations available today

Broker memory limits and cache. The PIP-442 document explains the current memory-based backpressure controls. In your case, you should first ensure that you have them in place. For example, managedLedgerMaxReadsInFlightSizeInMB is disabled by default, and maxMessagePublishBufferSizeInMB bounds the memory used for buffering published messages. It's also strongly recommended to tune the memory limits in the brokers and to add a sufficient amount of cache (managedLedgerCacheSizeMB, which defaults to 20% of direct memory, together with managedLedgerMaxReadsInFlightSizeInMB, and enough direct memory for the broker JVM to cover both).

Producers and clients. Since producers are usually the main source of problems, one way to currently address the problem is to set explicit maxPendingMessages / maxPendingMessagesAcrossPartitions, or to disable the client memory limit (memoryLimit set to 0) so that the default max pending messages limits are applied (1000 per producer and 50000 across partitions, since #26342 which is included in 4.0.14). The send timeout (sendTimeout) should be disabled (set to 0) for batch jobs. In addition, it's strongly recommended to keep clients at the latest stable and supported version.

Upgrade to Pulsar 5.0. It's recommended to go with Pulsar 5.0 (5.0.1 will be released very soon). Before Pulsar 5.0, the cache can consume significantly more memory than the configured setting. Pulsar 5.0 isn't only about scalable topics, and it's safe to upgrade from Pulsar 4.x to 5.0. There are more details in the Pulsar 5.0 release highlights and in the Pulsar 5.0 announcement blog post. Users who don't want to start using the new features can opt out (for example with scalableTopicsEnabled=false, together with the other rollback settings described in the upgrade guide), so that they can ensure complete compatibility with 4.x and also keep a downgrade path in case there's a need to roll back a deployment.

Capacity. Until the problem has been resolved, the best approach to deal with it is to overprovision the cluster. Pulsar already supports "broker isolation", where you can reserve specific brokers to serve only certain tenants. The problem with this is that it's an overprovisioning solution where you reserve extra capacity. A potentially better approach is to add more capacity to the whole cluster so that it can handle the peak workloads. Upgrading to Pulsar 5.0 is also a way to add capacity without adding more infrastructure resources, thanks to its performance improvements.

[lhotari]

Today we can't cap a tenant's total throughput, so from time to time one or more tenants put heavy load on the cluster. When this happens:

• the cluster becomes unstable, and in the worst cases it goes down;
• other tenants are affected: their clients fail to establish connections and report performance degradation, even though their own traffic hasn't changed.

@Goodkat Would you like to contribute a performance scenario to Pulsar's Performance testing framework that reproduces your issue and simulates your workload in the IoT domain that has been chosen as explained in the testing strategy. I'd like to avoid synthetic workloads in the scenarios so adapting a workload pattern to the IoT domain with a good background story is the current approach there. The scale of your workload doesn't need to be exposed. Most workloads can be down scaled to demonstrate the performance issues which occur in an actual much higher scale in production. You could use statistics from your own environment like batch message (entry) size variance, batch sizing etc. to find a way to simulate a similar type of workload. Downsizing the message sizes wouldn't make sense so that would be useful to keep at similar sizing as it's in your use cases.