You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/02/25 09:30:56 UTC
(pulsar) branch master updated: [fix][broker] PIP-322 Fix issue with rate limiters where rates can exceed limits initially and consumption pauses until token balance is positive (#24012)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new e547beaa6da [fix][broker] PIP-322 Fix issue with rate limiters where rates can exceed limits initially and consumption pauses until token balance is positive (#24012)
e547beaa6da is described below
commit e547beaa6dad72359bb5b8d30ce9fa80a6989f71
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Feb 25 11:30:49 2025 +0200
[fix][broker] PIP-322 Fix issue with rate limiters where rates can exceed limits initially and consumption pauses until token balance is positive (#24012)
Co-authored-by: Matteo Merli <mm...@apache.org>
---
conf/broker.conf | 7 +-
conf/standalone.conf | 7 +-
microbench/README.md | 9 +
.../broker/qos/AsyncTokenBucketBenchmark.java | 14 +-
...rk.java => DefaultMonotonicClockBenchmark.java} | 52 +---
.../apache/pulsar/broker/ServiceConfiguration.java | 11 +
.../org/apache/pulsar/broker/PulsarService.java | 21 +-
.../apache/pulsar/broker/qos/AsyncTokenBucket.java | 232 ++++++----------
.../pulsar/broker/qos/AsyncTokenBucketBuilder.java | 43 +--
.../pulsar/broker/qos/DefaultMonotonicClock.java | 35 +++
.../broker/qos/DefaultMonotonicSnapshotClock.java | 287 --------------------
.../broker/qos/DynamicRateAsyncTokenBucket.java | 11 +-
.../qos/DynamicRateAsyncTokenBucketBuilder.java | 6 +-
.../broker/qos/FinalRateAsyncTokenBucket.java | 20 +-
.../qos/FinalRateAsyncTokenBucketBuilder.java | 13 +-
...tonicSnapshotClock.java => MonotonicClock.java} | 24 +-
.../pulsar/broker/resourcegroup/ResourceGroup.java | 4 +-
.../resourcegroup/ResourceGroupPublishLimiter.java | 6 +-
.../broker/service/AbstractBaseDispatcher.java | 66 ++++-
.../pulsar/broker/service/AbstractTopic.java | 2 +-
.../pulsar/broker/service/BrokerService.java | 24 +-
.../broker/service/PublishRateLimiterImpl.java | 12 +-
.../NonPersistentDispatcherMultipleConsumers.java | 2 +
...onPersistentDispatcherSingleActiveConsumer.java | 4 +
.../service/persistent/DispatchRateLimiter.java | 136 ++--------
...> DispatchRateLimiterAsyncTokenBucketImpl.java} | 164 ++----------
.../persistent/DispatchRateLimiterClassicImpl.java | 215 +++++++++++++++
.../persistent/DispatchRateLimiterFactory.java | 28 ++
...DispatchRateLimiterFactoryAsyncTokenBucket.java | 45 ++++
.../DispatchRateLimiterFactoryClassic.java | 43 +++
.../PersistentDispatcherMultipleConsumers.java | 58 +---
...rsistentDispatcherMultipleConsumersClassic.java | 58 +---
.../PersistentDispatcherSingleActiveConsumer.java | 60 +----
.../service/persistent/PersistentReplicator.java | 4 +-
.../broker/service/persistent/PersistentTopic.java | 4 +-
.../service/persistent/SubscribeRateLimiter.java | 4 +-
.../pulsar/broker/qos/AsyncTokenBucketTest.java | 50 +---
.../qos/DefaultMonotonicSnapshotClockTest.java | 185 -------------
.../RGUsageMTAggrWaitForAllMsgsTest.java | 3 -
.../broker/service/AbstractBaseDispatcherTest.java | 5 +
.../pulsar/broker/service/AbstractTopicTest.java | 2 +-
.../DispatchRateLimiterOverconsumingTest.java | 284 ++++++++++++++++++++
.../PublishRateLimiterOverconsumingTest.java | 266 +++++++++++++++++++
.../broker/service/PublishRateLimiterTest.java | 2 +-
.../broker/service/ReplicatorRateLimiterTest.java | 3 -
.../service/TopicPublishRateThrottleTest.java | 4 +-
.../service/persistent/MessageDuplicationTest.java | 2 +-
.../api/AbstractMessageDispatchThrottlingTest.java | 3 -
.../client/impl/MessagePublishThrottlingTest.java | 3 -
.../pulsar/common/util/RateLimitFunction.java | 26 ++
.../org/apache/pulsar/common/util/RateLimiter.java | 291 +++++++++++++++++++++
.../apache/pulsar/common/util/RateLimiterTest.java | 245 +++++++++++++++++
52 files changed, 1846 insertions(+), 1259 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index f68306ec7b4..782b3755754 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -247,7 +247,7 @@ messageExpiryCheckIntervalInMinutes=5
# How long to delay rewinding cursor and dispatching messages when active consumer is changed
activeConsumerFailoverDelayTimeMillis=1000
-# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type.
+# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type.
# For non-partitioned topics, consistent hashing is used by default.
activeConsumerFailoverConsistentHashing=false
@@ -423,6 +423,11 @@ subscribeThrottlingRatePerConsumer=0
# Rate period for {subscribeThrottlingRatePerConsumer}. Default is 30s.
subscribeRatePeriodPerConsumerInSecond=30
+# The class name of the factory that creates DispatchRateLimiter implementations. Current options are
+# org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryAsyncTokenBucket (default, PIP-322 implementation)
+# and org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryClassic (legacy implementation)
+dispatchRateLimiterFactoryClassName=org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryAsyncTokenBucket
+
# Default messages per second dispatch throttling-limit for whole broker. Using a value of 0, is disabling default
# message dispatch-throttling
dispatchThrottlingRateInMsg=0
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 2036556da43..ee2d9732bdf 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -152,7 +152,7 @@ maxMessageSizeCheckIntervalInSeconds=60
# How long to delay rewinding cursor and dispatching messages when active consumer is changed
activeConsumerFailoverDelayTimeMillis=1000
-# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type.
+# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type.
# For non-partitioned topics, consistent hashing is used by default.
activeConsumerFailoverConsistentHashing=false
@@ -276,6 +276,11 @@ brokerPublisherThrottlingMaxMessageRate=0
# (Disable byte rate limit with value 0)
brokerPublisherThrottlingMaxByteRate=0
+# The class name of the factory that creates DispatchRateLimiter implementations. Current options are
+# org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryAsyncTokenBucket (default, PIP-322 implementation)
+# and org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryClassic (legacy implementation)
+dispatchRateLimiterFactoryClassName=org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryAsyncTokenBucket
+
# Default messages per second dispatch throttling-limit for every topic. Using a value of 0, is disabling default
# message dispatch-throttling
dispatchThrottlingRatePerTopicInMsg=0
diff --git a/microbench/README.md b/microbench/README.md
index f50c3036ff4..5c69c3bba81 100644
--- a/microbench/README.md
+++ b/microbench/README.md
@@ -67,3 +67,12 @@ Checking what benchmarks match the pattern:
java -jar microbench/target/microbenchmarks.jar ".*BenchmarkName.*" -lp
```
+Profiling benchmarks with [async-profiler](https://github.com/async-profiler/async-profiler):
+
+```shell
+# example of profiling with async-profiler
+# download async-profiler from https://github.com/async-profiler/async-profiler/releases
+LIBASYNCPROFILER_PATH=$HOME/async-profiler/lib/libasyncProfiler.dylib
+java -jar microbench/target/microbenchmarks.jar -prof async:libPath=$LIBASYNCPROFILER_PATH\;output=flamegraph\;dir=profile-results ".*BenchmarkName.*"
+```
+
diff --git a/microbench/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBenchmark.java b/microbench/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBenchmark.java
index 1b210258f13..d3fc62dce0f 100644
--- a/microbench/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBenchmark.java
+++ b/microbench/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBenchmark.java
@@ -30,19 +30,22 @@ import org.openjdk.jmh.annotations.OutputTimeUnit;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
-import org.openjdk.jmh.annotations.TearDown;
import org.openjdk.jmh.annotations.Threads;
import org.openjdk.jmh.annotations.Warmup;
import org.openjdk.jmh.infra.Blackhole;
+/**
+ * On MacOS, the performance of System.nanoTime() is not great. Running benchmarks on Linux is recommended due
+ * to the bottleneck of System.nanoTime() implementation on MacOS.
+ */
@Fork(3)
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
@State(Scope.Thread)
public class AsyncTokenBucketBenchmark {
private AsyncTokenBucket asyncTokenBucket;
- private DefaultMonotonicSnapshotClock monotonicSnapshotClock =
- new DefaultMonotonicSnapshotClock(TimeUnit.MILLISECONDS.toNanos(8), System::nanoTime);
+ private DefaultMonotonicClock monotonicSnapshotClock =
+ new DefaultMonotonicClock();
@Setup(Level.Iteration)
public void setup() {
@@ -51,11 +54,6 @@ public class AsyncTokenBucketBenchmark {
.initialTokens(2 * ratePerSecond).capacity(2 * ratePerSecond).build();
}
- @TearDown(Level.Iteration)
- public void teardown() {
- monotonicSnapshotClock.close();
- }
-
@Threads(1)
@Benchmark
@Measurement(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
diff --git a/microbench/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClockBenchmark.java b/microbench/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicClockBenchmark.java
similarity index 55%
rename from microbench/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClockBenchmark.java
rename to microbench/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicClockBenchmark.java
index d9054b8fe4b..1e9fc06ae21 100644
--- a/microbench/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClockBenchmark.java
+++ b/microbench/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicClockBenchmark.java
@@ -23,36 +23,33 @@ import java.util.concurrent.TimeUnit;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
-import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.State;
-import org.openjdk.jmh.annotations.TearDown;
import org.openjdk.jmh.annotations.Threads;
import org.openjdk.jmh.annotations.Warmup;
import org.openjdk.jmh.infra.Blackhole;
+/**
+ * On MacOS, the performance of System.nanoTime() is not great. Running benchmarks on Linux is recommended due
+ * to the bottleneck of System.nanoTime() implementation on MacOS.
+ */
@Fork(3)
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
@State(Scope.Thread)
-public class DefaultMonotonicSnapshotClockBenchmark {
- private DefaultMonotonicSnapshotClock monotonicSnapshotClock =
- new DefaultMonotonicSnapshotClock(TimeUnit.MILLISECONDS.toNanos(1), System::nanoTime);
-
- @TearDown(Level.Iteration)
- public void teardown() {
- monotonicSnapshotClock.close();
- }
+public class DefaultMonotonicClockBenchmark {
+ private DefaultMonotonicClock monotonicSnapshotClock =
+ new DefaultMonotonicClock();
@Threads(1)
@Benchmark
@Measurement(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
@Warmup(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
public void getTickNanos001Threads(Blackhole blackhole) {
- consumeTokenAndGetTokens(blackhole, false);
+ blackhole.consume(monotonicSnapshotClock.getTickNanos());
}
@Threads(10)
@@ -60,7 +57,7 @@ public class DefaultMonotonicSnapshotClockBenchmark {
@Measurement(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
@Warmup(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
public void getTickNanos010Threads(Blackhole blackhole) {
- consumeTokenAndGetTokens(blackhole, false);
+ blackhole.consume(monotonicSnapshotClock.getTickNanos());
}
@Threads(100)
@@ -68,35 +65,6 @@ public class DefaultMonotonicSnapshotClockBenchmark {
@Measurement(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
@Warmup(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
public void getTickNanos100Threads(Blackhole blackhole) {
- consumeTokenAndGetTokens(blackhole, false);
- }
-
- @Threads(1)
- @Benchmark
- @Measurement(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
- @Warmup(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
- public void getTickNanosRequestSnapshot001Threads(Blackhole blackhole) {
- consumeTokenAndGetTokens(blackhole, true);
- }
-
- @Threads(10)
- @Benchmark
- @Measurement(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
- @Warmup(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
- public void getTickNanosRequestSnapshot010Threads(Blackhole blackhole) {
- consumeTokenAndGetTokens(blackhole, true);
- }
-
- @Threads(100)
- @Benchmark
- @Measurement(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
- @Warmup(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
- public void getTickNanosRequestSnapshot100Threads(Blackhole blackhole) {
- consumeTokenAndGetTokens(blackhole, true);
- }
-
- private void consumeTokenAndGetTokens(Blackhole blackhole, boolean requestSnapshot) {
- // blackhole is used to ensure that the compiler doesn't do dead code elimination
- blackhole.consume(monotonicSnapshotClock.getTickNanos(requestSnapshot));
+ blackhole.consume(monotonicSnapshotClock.getTickNanos());
}
}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 37ef1cfaf62..96973fe0052 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -1042,6 +1042,7 @@ public class ServiceConfiguration implements PulsarConfiguration {
doc = "Enable precise rate limit for topic publish"
)
private boolean preciseTopicPublishRateLimiterEnable = false;
+
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
@@ -1064,6 +1065,16 @@ public class ServiceConfiguration implements PulsarConfiguration {
+ "when broker publish rate limiting enabled. (Disable byte rate limit with value 0)"
)
private long brokerPublisherThrottlingMaxByteRate = 0;
+
+ @FieldContext(category = CATEGORY_SERVER, doc =
+ "The class name of the factory that creates DispatchRateLimiter implementations. Current options are "
+ + "org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryAsyncTokenBucket "
+ + "(default, PIP-322 implementation) "
+ + "org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryClassic (legacy "
+ + "implementation)")
+ private String dispatchRateLimiterFactoryClassName =
+ "org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryAsyncTokenBucket";
+
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index a8331d86fc4..a09ed5142b8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -96,8 +96,8 @@ import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl
import org.apache.pulsar.broker.lookup.v1.TopicLookup;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.protocol.ProtocolHandlers;
-import org.apache.pulsar.broker.qos.DefaultMonotonicSnapshotClock;
-import org.apache.pulsar.broker.qos.MonotonicSnapshotClock;
+import org.apache.pulsar.broker.qos.DefaultMonotonicClock;
+import org.apache.pulsar.broker.qos.MonotonicClock;
import org.apache.pulsar.broker.resourcegroup.ResourceGroupService;
import org.apache.pulsar.broker.resourcegroup.ResourceUsageTopicTransportManager;
import org.apache.pulsar.broker.resourcegroup.ResourceUsageTransportManager;
@@ -300,7 +300,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
private TransactionPendingAckStoreProvider transactionPendingAckStoreProvider;
private final ExecutorProvider transactionExecutorProvider;
- private final DefaultMonotonicSnapshotClock monotonicSnapshotClock;
+ private final MonotonicClock monotonicClock;
private String brokerId;
private final CompletableFuture<Void> readyForIncomingRequestsFuture = new CompletableFuture<>();
private final List<Runnable> pendingTasksBeforeReadyForIncomingRequests = new ArrayList<>();
@@ -395,8 +395,11 @@ public class PulsarService implements AutoCloseable, ShutdownService {
// here in the constructor we don't have the offloader scheduler yet
this.offloaderStats = LedgerOffloaderStats.create(false, false, null, 0);
- this.monotonicSnapshotClock = new DefaultMonotonicSnapshotClock(TimeUnit.MILLISECONDS.toNanos(
- DEFAULT_MONOTONIC_CLOCK_GRANULARITY_MILLIS), System::nanoTime);
+ this.monotonicClock = createMonotonicClock(config);
+ }
+
+ protected MonotonicClock createMonotonicClock(ServiceConfiguration config) {
+ return new DefaultMonotonicClock();
}
public MetadataStore createConfigurationMetadataStore(PulsarMetadataEventSynchronizer synchronizer,
@@ -701,7 +704,9 @@ public class PulsarService implements AutoCloseable, ShutdownService {
brokerClientSharedScheduledExecutorProvider.shutdownNow();
brokerClientSharedLookupExecutorProvider.shutdownNow();
brokerClientSharedTimer.stop();
- monotonicSnapshotClock.close();
+ if (monotonicClock instanceof AutoCloseable c) {
+ c.close();
+ }
if (openTelemetryTransactionPendingAckStoreStats != null) {
openTelemetryTransactionPendingAckStoreStats.close();
@@ -1982,8 +1987,8 @@ public class PulsarService implements AutoCloseable, ShutdownService {
return brokerService.getListenPortTls();
}
- public MonotonicSnapshotClock getMonotonicSnapshotClock() {
- return monotonicSnapshotClock;
+ public MonotonicClock getMonotonicClock() {
+ return monotonicClock;
}
public static WorkerConfig initializeWorkerConfigFromBrokerConfig(ServiceConfiguration brokerConfig,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucket.java
index 8c43fa0a816..e4feb24453a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucket.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucket.java
@@ -27,7 +27,10 @@ import java.util.concurrent.atomic.LongAdder;
* An asynchronous token bucket algorithm implementation that is optimized for performance with highly concurrent
* use. CAS (compare-and-swap) operations are used and multiple levels of CAS fields are used to minimize contention
* when using CAS fields. The {@link LongAdder} class is used in the hot path to hold the sum of consumed tokens.
- * It is eventually consistent, meaning that the tokens are not updated on every call to the "consumeTokens" method.
+ * The internal implementation is eventually consistent, meaning that the tokens field value is not updated
+ * on every call to the "consumeTokens" method. Externally, the token balance is provided as a snapshot of the
+ * current balance, which is guaranteed to be off by at most the amount of new tokens gained during the
+ * addTokensResolutionNanos.
* <p>Main usage flow:
* 1. Tokens are consumed by invoking the "consumeTokens" or "consumeTokensAndCheckIfContainsTokens" methods.
* 2. The "consumeTokensAndCheckIfContainsTokens" or "containsTokens" methods return false if there are no
@@ -42,43 +45,24 @@ import java.util.concurrent.atomic.LongAdder;
* connection or client from the throttling queue to unthrottle. Before unthrottling, the application should check
* for available tokens. If tokens are still not available, the application should continue with throttling and
* repeat the throttling loop.
- * <p>By default, the AsyncTokenBucket is eventually consistent. This means that the token balance is updated
- * with added tokens and consumed tokens at most once during each "increment", when time advances more than the
- * configured resolution. There are settings for configuring consistency, please see {@link AsyncTokenBucketBuilder}
- * for details.
* <p>This class does not produce side effects outside its own scope. It functions similarly to a stateful function,
* akin to a counter function. In essence, it is a sophisticated counter. It can serve as a foundational component for
* constructing higher-level asynchronous rate limiter implementations, which require side effects for throttling.
- * <p>To achieve optimal performance, pass a {@link DefaultMonotonicSnapshotClock} instance as the clock .
*/
public abstract class AsyncTokenBucket {
- public static final MonotonicSnapshotClock DEFAULT_SNAPSHOT_CLOCK = requestSnapshot -> System.nanoTime();
+ public static final MonotonicClock DEFAULT_SNAPSHOT_CLOCK = new DefaultMonotonicClock();
static final long ONE_SECOND_NANOS = TimeUnit.SECONDS.toNanos(1);
- // 2^24 nanoseconds is 16 milliseconds
- private static final long DEFAULT_RESOLUTION_NANOS = TimeUnit.MILLISECONDS.toNanos(16);
-
- // The default resolution is 16 milliseconds. This means that the consumed tokens are subtracted from the
- // current amount of tokens about every 16 milliseconds. This solution helps prevent a CAS loop what could cause
+ // The default add tokens resolution is 16 milliseconds. This means that new tokens are calculated and added
+ // to the token balance about every 16 milliseconds. This solution helps prevent a CAS loop what could cause
// extra CPU usage when a single CAS field is updated at a high rate from multiple threads.
- static long defaultResolutionNanos = DEFAULT_RESOLUTION_NANOS;
-
- // used in tests to disable the optimization and instead use a consistent view of the tokens
- public static void switchToConsistentTokensView() {
- defaultResolutionNanos = 0;
- }
-
- public static void resetToDefaultEventualConsistentTokensView() {
- defaultResolutionNanos = DEFAULT_RESOLUTION_NANOS;
- }
+ // 2^24 nanoseconds is 16 milliseconds
+ public static final long DEFAULT_ADD_TOKENS_RESOLUTION_NANOS = TimeUnit.MILLISECONDS.toNanos(16);
// atomic field updaters for the volatile fields in this class
private static final AtomicLongFieldUpdater<AsyncTokenBucket> LAST_NANOS_UPDATER =
AtomicLongFieldUpdater.newUpdater(AsyncTokenBucket.class, "lastNanos");
- private static final AtomicLongFieldUpdater<AsyncTokenBucket> LAST_INCREMENT_UPDATER =
- AtomicLongFieldUpdater.newUpdater(AsyncTokenBucket.class, "lastIncrement");
-
private static final AtomicLongFieldUpdater<AsyncTokenBucket> TOKENS_UPDATER =
AtomicLongFieldUpdater.newUpdater(AsyncTokenBucket.class, "tokens");
@@ -97,25 +81,21 @@ public abstract class AsyncTokenBucket {
* By default, a monotonic clock (System.nanoTime()) is used.
*/
private volatile long lastNanos;
- /**
- * This field represents the last time the tokens were updated, in increments.
- */
- private volatile long lastIncrement;
/**
* As time progresses, tokens are added to the bucket. When the rate is low, significant rounding errors could
* accumulate over time if the remainder nanoseconds are not accounted for in the calculations. This field is used
* to carry forward the leftover nanoseconds in the update calculation.
*/
private volatile long remainderNanos;
-
/**
- * The resolution in nanoseconds. This is the amount of time that must pass before the tokens are updated.
+ * The add tokens resolution in nanoseconds. This is the amount of time that must pass before new tokens are
+ * updated.
*/
- protected final long resolutionNanos;
+ protected final long addTokensResolutionNanos;
/**
* This field is used to obtain the current monotonic clock time in nanoseconds.
*/
- private final MonotonicSnapshotClock clockSource;
+ private final MonotonicClock clockSource;
/**
* This field is used to hold the sum of consumed tokens that are pending to be subtracted from the total amount of
* tokens. This solution is to prevent CAS loop contention problem. pendingConsumedTokens used JVM's LongAdder
@@ -123,28 +103,10 @@ public abstract class AsyncTokenBucket {
*/
private final LongAdder pendingConsumedTokens = new LongAdder();
- /**
- * By default, AsyncTokenBucket is eventually consistent. This means that the consumed tokens are subtracted from
- * the total amount of tokens at most once during each "increment", when time advances more than the configured
- * resolution. This setting determines if the consumed tokens are subtracted from tokens balance consistently.
- * For high performance, it is recommended to keep this setting as false.
- */
- private final boolean consistentConsumedTokens;
- /**
- * By default, AsyncTokenBucket is eventually consistent. This means that the added tokens are calculated based
- * on elapsed time at most once during each "increment", when time advances more than the configured
- * resolution. This setting determines if the added tokens are calculated and added to tokens balance consistently.
- * For high performance, it is recommended to keep this setting as false.
- */
- private final boolean consistentAddedTokens;
-
- protected AsyncTokenBucket(MonotonicSnapshotClock clockSource, long resolutionNanos,
- boolean consistentConsumedTokens, boolean consistentAddedTokens) {
+ protected AsyncTokenBucket(MonotonicClock clockSource, long addTokensResolutionNanos) {
this.clockSource = clockSource;
- this.resolutionNanos = resolutionNanos;
+ this.addTokensResolutionNanos = addTokensResolutionNanos;
this.lastNanos = Long.MIN_VALUE;
- this.consistentConsumedTokens = consistentConsumedTokens;
- this.consistentAddedTokens = consistentAddedTokens;
}
public static FinalRateAsyncTokenBucketBuilder builder() {
@@ -160,34 +122,28 @@ public abstract class AsyncTokenBucket {
protected abstract long getTargetAmountOfTokensAfterThrottling();
/**
- * Consumes tokens and possibly updates the tokens balance. New tokens are calculated and added to the current
- * tokens balance each time the update takes place. The update takes place once in every interval of the configured
- * resolutionNanos or when the forceConsistentTokens parameter is true.
- * When the tokens balance isn't updated, the consumed tokens are added to the pendingConsumedTokens LongAdder
- * counter which gets flushed the next time the tokens are updated. This makes the tokens balance
- * eventually consistent. The reason for this design choice is to optimize performance by preventing CAS loop
- * contention which could cause excessive CPU consumption.
+ * Consumes tokens and possibly updates the token balance. New tokens are calculated if the last new token
+ * calculation occurred more than addTokensResolutionNanos nanoseconds ago. When new tokens are added, the
+ * token balance held in the `tokens` field is updated.
+ * If the token balance isn't updated, the consumed tokens are added to the pendingConsumedTokens LongAdder
+ * counter, which gets flushed the next time the tokens are updated. This design choice optimizes performance
+ * by preventing CAS loop contention, which could cause excessive CPU consumption.
+ * The returned balance is guaranteed to be off by at most the amount of new tokens gained during
+ * addTokensResolutionNanos. In the case of not updating the token balance, the current balance is calculated
+ * by subtracting the pendingConsumedTokens sum from the current balance.
*
- * @param consumeTokens number of tokens to consume, can be 0 to update the tokens balance
- * @param forceConsistentTokens if true, the token balance is updated consistently
- * @return the current number of tokens in the bucket or Long.MIN_VALUE when the number of tokens is unknown due
- * to eventual consistency
+ * @param consumeTokens number of tokens to consume; can be 0 to update the token balance
+ * @return the current token balance, guaranteed to be off by at most the amount of new tokens gained during
+ * addTokensResolutionNanos
*/
- private long consumeTokensAndMaybeUpdateTokensBalance(long consumeTokens, boolean forceConsistentTokens) {
+ private long consumeTokensAndMaybeUpdateTokensBalance(long consumeTokens) {
if (consumeTokens < 0) {
throw new IllegalArgumentException("consumeTokens must be >= 0");
}
- boolean requestConsistentTickNanosSnapshot =
- consistentAddedTokens || consistentConsumedTokens || forceConsistentTokens || resolutionNanos == 0;
- long currentNanos = clockSource.getTickNanos(requestConsistentTickNanosSnapshot);
- long newTokens = 0;
- // check if the tokens should be updated immediately
- if (shouldAddTokensImmediately(currentNanos, forceConsistentTokens)) {
- // calculate the number of new tokens since the last update
- newTokens = calculateNewTokensSinceLastUpdate(currentNanos, forceConsistentTokens);
- }
- // update tokens if there are new tokens or if resolutionNanos is set to 0 which is currently used for testing
- if (newTokens > 0 || resolutionNanos == 0 || consistentConsumedTokens || forceConsistentTokens) {
+ long currentNanos = clockSource.getTickNanos();
+ long newTokens = calculateNewTokensSinceLastUpdate(currentNanos);
+ // update token balance if there are new tokens
+ if (newTokens > 0) {
// flush the pendingConsumedTokens by calling "sumThenReset"
long currentPendingConsumedTokens = pendingConsumedTokens.sumThenReset();
// calculate the token delta by subtracting the consumed tokens from the new tokens
@@ -203,38 +159,19 @@ public abstract class AsyncTokenBucket {
return tokens;
}
} else {
- // eventual consistent fast path, tokens are not updated immediately
+ // tokens are not updated immediately to prevent CAS loop contention
// add the consumed tokens to the pendingConsumedTokens LongAdder counter
if (consumeTokens > 0) {
pendingConsumedTokens.add(consumeTokens);
}
- // return Long.MIN_VALUE if the current value of tokens is unknown due to the eventual consistency
- return Long.MIN_VALUE;
+ // return token balance without updating the balance
+ // this might be off by up to the amount of new tokens gained during addTokensResolutionNanos
+ return tokens - pendingConsumedTokens.sum();
}
}
- /**
- * Check if the tokens should be updated immediately.
- *
- * The tokens will be updated once every resolutionNanos nanoseconds.
- * This method checks if the configured resolutionNanos has passed since the last update.
- * If the forceConsistentTokens is true, the tokens will be updated immediately.
- *
- * @param currentNanos the current monotonic clock time in nanoseconds
- * @param forceConsistentTokens if true, the tokens are added even if the configured resolution hasn't fully passed
- * @return true if the tokens should be updated immediately, false otherwise
- */
- private boolean shouldAddTokensImmediately(long currentNanos, boolean forceConsistentTokens) {
- long currentIncrement = resolutionNanos != 0 ? currentNanos / resolutionNanos : 0;
- long currentLastIncrement = lastIncrement;
- return currentIncrement == 0
- || (currentIncrement > currentLastIncrement
- && LAST_INCREMENT_UPDATER.compareAndSet(this, currentLastIncrement, currentIncrement))
- || consistentAddedTokens || forceConsistentTokens;
- }
-
/**
* Calculate the number of new tokens since the last update.
* This will carry forward the remainder nanos so that a possible rounding error is eliminated.
@@ -242,12 +179,16 @@ public abstract class AsyncTokenBucket {
* @param currentNanos the current monotonic clock time in nanoseconds
* @return the number of new tokens to add since the last update
*/
- private long calculateNewTokensSinceLastUpdate(long currentNanos, boolean forceConsistentTokens) {
+ private long calculateNewTokensSinceLastUpdate(long currentNanos) {
long previousLastNanos = lastNanos;
long newLastNanos;
- // update lastNanos only if at least resolutionNanos/2 nanoseconds has passed since the last update
- // unless consistency is needed
- long minimumIncrementNanos = forceConsistentTokens || consistentAddedTokens ? 0L : resolutionNanos / 2;
+ long minimumIncrementNanos;
+ // update lastNanos only if there would be at least one token added
+ if (getNanosForOneToken() > addTokensResolutionNanos) {
+ minimumIncrementNanos = getNanosForOneToken() - remainderNanos - 1;
+ } else {
+ minimumIncrementNanos = addTokensResolutionNanos;
+ }
if (currentNanos > previousLastNanos + minimumIncrementNanos) {
newLastNanos = currentNanos;
} else {
@@ -276,97 +217,82 @@ public abstract class AsyncTokenBucket {
}
/**
- * Eventually consume tokens from the bucket.
- * The number of tokens is eventually consistent with the configured granularity of resolutionNanos.
+ * Consume tokens from the bucket.
*
* @param consumeTokens the number of tokens to consume
*/
public void consumeTokens(long consumeTokens) {
- consumeTokensAndMaybeUpdateTokensBalance(consumeTokens, false);
+ if (consumeTokens < 0) {
+ throw new IllegalArgumentException("consumeTokens must be >= 0");
+ }
+ if (consumeTokens > 0) {
+ pendingConsumedTokens.add(consumeTokens);
+ }
}
/**
- * Eventually consume tokens from the bucket and check if tokens remain available.
- * The number of tokens is eventually consistent with the configured granularity of resolutionNanos.
- * Therefore, the returned result is not definite.
+ * Consume tokens from the bucket and check if tokens remain available.
*
* @param consumeTokens the number of tokens to consume
- * @return true if there is tokens remains, false if tokens are all consumed. The answer isn't definite since the
- * comparison is made with eventually consistent token value.
+ * @return true if there is tokens remains, false if tokens are all consumed.
*/
public boolean consumeTokensAndCheckIfContainsTokens(long consumeTokens) {
- long currentTokens = consumeTokensAndMaybeUpdateTokensBalance(consumeTokens, false);
- if (currentTokens > 0) {
- // tokens remain in the bucket
- return true;
- } else if (currentTokens == Long.MIN_VALUE) {
- // when currentTokens is Long.MIN_VALUE, the current tokens balance is unknown since consumed tokens
- // was added to the pendingConsumedTokens LongAdder counter. In this case, assume that tokens balance
- // hasn't been updated yet and calculate a best guess of the current value by substracting the consumed
- // tokens from the current tokens balance
- return tokens - consumeTokens > 0;
- } else {
- // no tokens remain in the bucket
- return false;
- }
+ return consumeTokensAndMaybeUpdateTokensBalance(consumeTokens) > 0;
}
/**
- * Returns the current token balance. When forceConsistentTokens is true, the tokens balance is updated before
- * returning. If forceConsistentTokens is false, the tokens balance could be updated if the last updated happened
- * more than resolutionNanos nanoseconds ago.
+ * Returns the current token balance.
*
* @return the current token balance
*/
private long tokens() {
- long currentTokens = consumeTokensAndMaybeUpdateTokensBalance(0, false);
- if (currentTokens != Long.MIN_VALUE) {
- // when currentTokens isn't Long.MIN_VALUE, the current tokens balance is known
- return currentTokens;
- } else {
- // return the current tokens balance, ignore the possible pendingConsumedTokens LongAdder counter
- return tokens;
- }
+ return consumeTokensAndMaybeUpdateTokensBalance(0);
}
/**
* Calculate the required throttling duration in nanoseconds to fill up the bucket with the minimum amount of
- * tokens.
- * This method shouldn't be called from the hot path since it calculates a consistent value for the tokens which
- * isn't necessary on the hotpath.
+ * tokens. Will return 0 if there are available tokens in the bucket.
*/
public long calculateThrottlingDuration() {
- long currentTokens = consumeTokensAndMaybeUpdateTokensBalance(0, true);
- if (currentTokens == Long.MIN_VALUE) {
- throw new IllegalArgumentException(
- "Unexpected result from updateAndConsumeTokens with forceConsistentTokens set to true");
- }
- if (currentTokens > 0) {
+ return calculateThrottlingDuration(Math.max(1, getTargetAmountOfTokensAfterThrottling()));
+ }
+
+ /**
+ * Calculate the required throttling duration in nanoseconds to fill up the bucket with the required amount
+ * of tokens. Will return 0 if the required amount of tokens is already in the bucket.
+ */
+ public long calculateThrottlingDuration(long requiredTokens) {
+ long currentTokens = consumeTokensAndMaybeUpdateTokensBalance(0);
+ if (currentTokens >= requiredTokens) {
return 0L;
}
- // currentTokens is negative, so subtracting a negative value results in adding the absolute value (-(-x) -> +x)
- long needTokens = getTargetAmountOfTokensAfterThrottling() - currentTokens;
+ // when currentTokens is negative, subtracting a negative value results in
+ // adding the absolute value (-(-x) -> +x)
+ long needTokens = requiredTokens - currentTokens;
return (needTokens * getRatePeriodNanos()) / getRate();
}
+ /**
+ * Returns the configured capacity of the bucket.
+ */
public abstract long getCapacity();
/**
- * Returns the current number of tokens in the bucket.
- * The token balance is updated if the configured resolutionNanos has passed since the last update unless
- * consistentConsumedTokens is true.
+ * Returns the current token balance of the bucket.
*/
public final long getTokens() {
return tokens();
}
+ /**
+ * Returns the configured rate of the bucket.
+ */
public abstract long getRate();
+ protected abstract long getNanosForOneToken();
+
/**
* Checks if the bucket contains tokens.
- * The token balance is updated before the comparison if the configured resolutionNanos has passed since the last
- * update. It's possible that the returned result is not definite since the token balance is eventually consistent
- * if consistentConsumedTokens is false.
*
* @return true if the bucket contains tokens, false otherwise
*/
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBuilder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBuilder.java
index 1c05f1a213e..9c0f5f5605c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBuilder.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBuilder.java
@@ -21,10 +21,8 @@ package org.apache.pulsar.broker.qos;
// CHECKSTYLE.OFF: ClassTypeParameterName
public abstract class AsyncTokenBucketBuilder<SELF extends AsyncTokenBucketBuilder<SELF>> {
- protected MonotonicSnapshotClock clock = AsyncTokenBucket.DEFAULT_SNAPSHOT_CLOCK;
- protected long resolutionNanos = AsyncTokenBucket.defaultResolutionNanos;
- protected boolean consistentConsumedTokens;
- protected boolean consistentAddedTokens;
+ protected MonotonicClock clock = AsyncTokenBucket.DEFAULT_SNAPSHOT_CLOCK;
+ protected long addTokensResolutionNanos = AsyncTokenBucket.DEFAULT_ADD_TOKENS_RESOLUTION_NANOS;
protected AsyncTokenBucketBuilder() {
}
@@ -34,44 +32,21 @@ public abstract class AsyncTokenBucketBuilder<SELF extends AsyncTokenBucketBuild
}
/**
- * Set the clock source for the token bucket. It's recommended to use the {@link DefaultMonotonicSnapshotClock}
+ * Set the clock source for the token bucket. It's recommended to use the {@link DefaultMonotonicClock}
* for most use cases.
*/
- public SELF clock(MonotonicSnapshotClock clock) {
+ public SELF clock(MonotonicClock clock) {
this.clock = clock;
return self();
}
/**
- * By default, AsyncTokenBucket is eventually consistent. This means that the token balance is updated, when time
- * advances more than the configured resolution. This setting determines the duration of the increment.
- * Setting this value to 0 will make the token balance fully consistent. There's a performance trade-off
- * when setting this value to 0.
+ * This setting determines the minimum time interval that must pass since the last operation
+ * before new tokens are added to the token balance.
+ * Setting this value to 0 ensures the token balance is always up-to-date, but it may slightly impact performance.
*/
- public SELF resolutionNanos(long resolutionNanos) {
- this.resolutionNanos = resolutionNanos;
- return self();
- }
-
- /**
- * By default, AsyncTokenBucket is eventually consistent. This means that the consumed tokens are subtracted from
- * the total amount of tokens at most once during each "increment", when time advances more than the configured
- * resolution. This setting determines if the consumed tokens are subtracted from tokens balance consistently.
- * For high performance, it is recommended to keep this setting as false.
- */
- public SELF consistentConsumedTokens(boolean consistentConsumedTokens) {
- this.consistentConsumedTokens = consistentConsumedTokens;
- return self();
- }
-
- /**
- * By default, AsyncTokenBucket is eventually consistent. This means that the added tokens are calculated based
- * on elapsed time at most once during each "increment", when time advances more than the configured
- * resolution. This setting determines if the added tokens are calculated and added to tokens balance consistently.
- * For high performance, it is recommended to keep this setting as false.
- */
- public SELF consistentAddedTokens(boolean consistentAddedTokens) {
- this.consistentAddedTokens = consistentAddedTokens;
+ public SELF addTokensResolutionNanos(long addTokensResolutionNanos) {
+ this.addTokensResolutionNanos = addTokensResolutionNanos;
return self();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicClock.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicClock.java
new file mode 100644
index 00000000000..38b3d175651
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicClock.java
@@ -0,0 +1,35 @@
+/*
+ * 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.qos;
+
+/**
+ * Default implementation of {@link MonotonicClock} for use with {@link AsyncTokenBucket}.
+ *
+ * <p>
+ * This delegates directly to System.nanoTime() and relies on the underlying system clock for monotonic timekeeping.
+ * </p>
+ */
+public class DefaultMonotonicClock implements MonotonicClock {
+ /** {@inheritDoc} */
+ @Override
+ public long getTickNanos() {
+ return System.nanoTime();
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java
deleted file mode 100644
index 23b9359c804..00000000000
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java
+++ /dev/null
@@ -1,287 +0,0 @@
-/*
- * 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.qos;
-
-import java.util.Objects;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.LongConsumer;
-import java.util.function.LongSupplier;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Default implementation of {@link MonotonicSnapshotClock} optimized for use with {@link AsyncTokenBucket}.
- *
- * <p>
- * This class provides a monotonic snapshot value that consistently increases, ensuring reliable behavior
- * even in environments where the underlying clock source may not be strictly monotonic across all CPUs,
- * such as certain virtualized platforms.
- * </p>
- *
- * <p>
- * Upon instantiation, a daemon thread is launched to periodically update the snapshot value at a configured
- * interval. It is essential to invoke the {@link #close()} method to gracefully terminate this thread when it is
- * no longer needed.
- * </p>
- *
- * <p>
- * The {@link AsyncTokenBucket} utilizes this clock to obtain tick values. It does not require a consistent value on
- * every retrieval. However, when a consistent snapshot is necessary, the {@link #getTickNanos(boolean)} method
- * is called with the {@code requestSnapshot} parameter set to {@code true}.
- * </p>
- *
- * <p>
- * By employing a single thread to update the monotonic clock value, this implementation ensures that the snapshot
- * value remains strictly increasing. This approach mitigates potential inconsistencies that may arise from clock
- * source discrepancies across different CPUs.
- * </p>
- */
-public class DefaultMonotonicSnapshotClock implements MonotonicSnapshotClock, AutoCloseable {
- private static final Logger LOG = LoggerFactory.getLogger(DefaultMonotonicSnapshotClock.class);
- private final TickUpdaterThread tickUpdaterThread;
- private volatile long snapshotTickNanos;
-
- public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier clockSource) {
- if (snapshotIntervalNanos < TimeUnit.MILLISECONDS.toNanos(1)) {
- throw new IllegalArgumentException("snapshotIntervalNanos must be at least 1 millisecond");
- }
- tickUpdaterThread = new TickUpdaterThread(snapshotIntervalNanos,
- Objects.requireNonNull(clockSource, "clockSource must not be null"), this::setSnapshotTickNanos);
- tickUpdaterThread.start();
- }
-
- private void setSnapshotTickNanos(long snapshotTickNanos) {
- this.snapshotTickNanos = snapshotTickNanos;
- }
-
- /** {@inheritDoc} */
- @Override
- public long getTickNanos(boolean requestSnapshot) {
- if (requestSnapshot) {
- tickUpdaterThread.requestUpdateAndWait();
- }
- return snapshotTickNanos;
- }
-
- @Override
- public void close() {
- tickUpdaterThread.interrupt();
- }
-
- /**
- * A thread that updates snapshotTickNanos value periodically with a configured interval.
- * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
- * called.
- * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
- * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
- * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
- * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
- */
- private static class TickUpdaterThread extends Thread {
- private final Object tickUpdateDelayMonitor = new Object();
- private final Object tickUpdatedMonitor = new Object();
- private final MonotonicLeapDetectingTickUpdater tickUpdater;
- private volatile boolean running;
- private boolean tickUpdateDelayMonitorNotified;
- private AtomicLong requestCount = new AtomicLong();
- private final long sleepMillis;
- private final int sleepNanos;
-
- TickUpdaterThread(long snapshotIntervalNanos, LongSupplier clockSource, LongConsumer setSnapshotTickNanos) {
- super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
- // set as daemon thread so that it doesn't prevent the JVM from exiting
- setDaemon(true);
- // set the highest priority
- setPriority(MAX_PRIORITY);
- this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
- this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
- tickUpdater = new MonotonicLeapDetectingTickUpdater(clockSource, setSnapshotTickNanos,
- snapshotIntervalNanos);
- }
-
- @Override
- public void run() {
- try {
- running = true;
- long updatedForRequestCount = -1;
- while (!isInterrupted()) {
- try {
- // track if the thread has waited for the whole duration of the snapshot interval
- // before updating the tick value
- boolean waitedSnapshotInterval = false;
- // sleep for the configured interval on a monitor that can be notified to stop the sleep
- // and update the tick value immediately. This is used in requestUpdate method.
- synchronized (tickUpdateDelayMonitor) {
- tickUpdateDelayMonitorNotified = false;
- // only wait if no explicit request has been made since the last update
- if (requestCount.get() == updatedForRequestCount) {
- // if no request has been made, sleep for the configured interval
- tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
- waitedSnapshotInterval = !tickUpdateDelayMonitorNotified;
- }
- }
- updatedForRequestCount = requestCount.get();
- // update the tick value using the tick updater which will tolerate leaps backward
- tickUpdater.update(waitedSnapshotInterval);
- notifyAllTickUpdated();
- } catch (InterruptedException e) {
- interrupt();
- break;
- }
- }
- } catch (Throwable t) {
- // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
- // this is very unlikely to happen, but it's better to log it in any case
- LOG.error("Unexpected fatal error that stopped the clock.", t);
- } finally {
- LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
- running = false;
- notifyAllTickUpdated();
- }
- }
-
- private void notifyAllTickUpdated() {
- synchronized (tickUpdatedMonitor) {
- // notify all threads that are waiting for the tick value to be updated
- tickUpdatedMonitor.notifyAll();
- }
- }
-
- public void requestUpdateAndWait() {
- if (!running) {
- synchronized (tickUpdater) {
- // thread has stopped running, fallback to update the value directly without optimizations
- tickUpdater.update(false);
- }
- return;
- }
- // increment the request count that ensures that the thread will update the tick value after this request
- // was made also when there's a race condition between the request and the update
- // this solution doesn't prevent all races, and it's not guaranteed that the tick value is always updated
- // it will prevent the request having to wait for the delayed update cycle. This is sufficient for the
- // use case.
- requestCount.incrementAndGet();
- synchronized (tickUpdatedMonitor) {
- // notify the thread to stop waiting and update the tick value
- synchronized (tickUpdateDelayMonitor) {
- tickUpdateDelayMonitorNotified = true;
- tickUpdateDelayMonitor.notify();
- }
- // wait until the tick value has been updated
- try {
- tickUpdatedMonitor.wait();
- } catch (InterruptedException e) {
- currentThread().interrupt();
- }
- }
- }
-
- @Override
- public synchronized void start() {
- // wait until the thread is started and the tick value has been updated
- synchronized (tickUpdatedMonitor) {
- super.start();
- try {
- tickUpdatedMonitor.wait();
- } catch (InterruptedException e) {
- currentThread().interrupt();
- }
- }
- }
- }
-
- /**
- * Handles updating the tick value in a monotonic way so that the value is always increasing,
- * regardless of leaps backward in the clock source value.
- */
- static class MonotonicLeapDetectingTickUpdater {
- private final LongSupplier clockSource;
- private final long snapshotInternalNanos;
- private final long maxDeltaNanosForLeapDetection;
- private final LongConsumer tickUpdatedCallback;
- private long referenceClockSourceValue = Long.MIN_VALUE;
- private long baseSnapshotTickNanos;
- private long previousSnapshotTickNanos;
-
- MonotonicLeapDetectingTickUpdater(LongSupplier clockSource, LongConsumer tickUpdatedCallback,
- long snapshotInternalNanos) {
- this.clockSource = clockSource;
- this.snapshotInternalNanos = snapshotInternalNanos;
- this.maxDeltaNanosForLeapDetection = 2 * snapshotInternalNanos;
- this.tickUpdatedCallback = tickUpdatedCallback;
- }
-
- /**
- * Updates the snapshot tick value. The tickUpdatedCallback is called if the value has changed.
- * The value is updated in a monotonic way so that the value is always increasing, regardless of leaps backward
- * in the clock source value.
- * Leap detection is done by comparing the new value with the previous value and the maximum delta value.
- *
- * @param waitedSnapshotInterval if true, the method has waited for the snapshot interval since the previous
- * call.
- */
- public void update(boolean waitedSnapshotInterval) {
- // get the current clock source value
- long clockValue = clockSource.getAsLong();
-
- // Initialization on first call
- if (referenceClockSourceValue == Long.MIN_VALUE) {
- referenceClockSourceValue = clockValue;
- baseSnapshotTickNanos = clockValue;
- previousSnapshotTickNanos = clockValue;
- // update the tick value using the callback
- tickUpdatedCallback.accept(clockValue);
- return;
- }
-
- // calculate the duration since the reference clock source value
- // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
- // monotonic across all CPUs and leaps backward
- long durationSinceReference = clockValue - referenceClockSourceValue;
- // calculate the new snapshot tick value as a duration since the reference clock source value
- // and add it to the base snapshot tick value
- long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
-
- // reset the reference clock source value if the clock source value leaps backward
- // more than the maximum delta value
- if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDeltaNanosForLeapDetection) {
- // when the clock source value leaps backward, reset the reference value to the new value
- // for future duration calculations
- referenceClockSourceValue = clockValue;
- // if the updater thread has waited for the snapshot interval since the previous call,
- // increment the base snapshot tick value by the snapshot interval value
- long incrementWhenLeapDetected = waitedSnapshotInterval ? snapshotInternalNanos : 0;
- // set the base snapshot tick value to the new value
- baseSnapshotTickNanos = previousSnapshotTickNanos + incrementWhenLeapDetected;
- // set the new snapshot tick value to the base value
- newSnapshotTickNanos = baseSnapshotTickNanos;
- }
-
- // update snapshotTickNanos value if the new value is greater than the previous value
- if (newSnapshotTickNanos > previousSnapshotTickNanos) {
- // store the previous value
- previousSnapshotTickNanos = newSnapshotTickNanos;
- // update the tick value using the callback
- tickUpdatedCallback.accept(newSnapshotTickNanos);
- }
- }
- }
-}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DynamicRateAsyncTokenBucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DynamicRateAsyncTokenBucket.java
index f2eae8aed8d..e50d3be7b72 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DynamicRateAsyncTokenBucket.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DynamicRateAsyncTokenBucket.java
@@ -33,11 +33,10 @@ public class DynamicRateAsyncTokenBucket extends AsyncTokenBucket {
private final double targetFillFactorAfterThrottling;
protected DynamicRateAsyncTokenBucket(double capacityFactor, LongSupplier rateFunction,
- MonotonicSnapshotClock clockSource, LongSupplier ratePeriodNanosFunction,
- long resolutionNanos, boolean consistentConsumedTokens,
- boolean consistentAddedTokens, double initialTokensFactor,
+ MonotonicClock clockSource, long addTokensResolutionNanos,
+ LongSupplier ratePeriodNanosFunction, double initialTokensFactor,
double targetFillFactorAfterThrottling) {
- super(clockSource, resolutionNanos, consistentConsumedTokens, consistentAddedTokens);
+ super(clockSource, addTokensResolutionNanos);
this.capacityFactor = capacityFactor;
this.rateFunction = rateFunction;
this.ratePeriodNanosFunction = ratePeriodNanosFunction;
@@ -66,4 +65,8 @@ public class DynamicRateAsyncTokenBucket extends AsyncTokenBucket {
return rateFunction.getAsLong();
}
+ @Override
+ protected long getNanosForOneToken() {
+ return getRatePeriodNanos() / getRate();
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DynamicRateAsyncTokenBucketBuilder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DynamicRateAsyncTokenBucketBuilder.java
index 8aebecddf90..35362cd83c8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DynamicRateAsyncTokenBucketBuilder.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DynamicRateAsyncTokenBucketBuilder.java
@@ -63,8 +63,8 @@ public class DynamicRateAsyncTokenBucketBuilder
@Override
public AsyncTokenBucket build() {
- return new DynamicRateAsyncTokenBucket(this.capacityFactor, this.rateFunction,
- this.clock, this.ratePeriodNanosFunction, this.resolutionNanos, this.consistentConsumedTokens,
- this.consistentAddedTokens, this.initialFillFactor, targetFillFactorAfterThrottling);
+ return new DynamicRateAsyncTokenBucket(this.capacityFactor, this.rateFunction, this.clock,
+ this.addTokensResolutionNanos, this.ratePeriodNanosFunction, this.initialFillFactor,
+ targetFillFactorAfterThrottling);
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/FinalRateAsyncTokenBucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/FinalRateAsyncTokenBucket.java
index d83290b723f..43969d783db 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/FinalRateAsyncTokenBucket.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/FinalRateAsyncTokenBucket.java
@@ -24,20 +24,24 @@ package org.apache.pulsar.broker.qos;
* The rate and capacity of the token bucket are constant and do not change over time.
*/
class FinalRateAsyncTokenBucket extends AsyncTokenBucket {
+ private static final double DEFAULT_TARGET_FILL_FACTOR_AFTER_THROTTLING = 0.01d;
private final long capacity;
private final long rate;
private final long ratePeriodNanos;
+ private final long nanosForOneToken;
private final long targetAmountOfTokensAfterThrottling;
- protected FinalRateAsyncTokenBucket(long capacity, long rate, MonotonicSnapshotClock clockSource,
- long ratePeriodNanos, long resolutionNanos, boolean consistentConsumedTokens,
- boolean consistentAddedTokens, long initialTokens) {
- super(clockSource, resolutionNanos, consistentConsumedTokens, consistentAddedTokens);
+ protected FinalRateAsyncTokenBucket(long capacity, long rate, MonotonicClock clockSource,
+ long addTokensResolutionNanos, long ratePeriodNanos, long initialTokens,
+ long targetAmountOfTokensAfterThrottling) {
+ super(clockSource, addTokensResolutionNanos);
this.capacity = capacity;
this.rate = rate;
this.ratePeriodNanos = ratePeriodNanos != -1 ? ratePeriodNanos : ONE_SECOND_NANOS;
- // The target amount of tokens is the amount of tokens made available in the resolution duration
- this.targetAmountOfTokensAfterThrottling = Math.max(this.resolutionNanos * rate / ratePeriodNanos, 1);
+ this.nanosForOneToken = ratePeriodNanos / rate;
+ this.targetAmountOfTokensAfterThrottling = targetAmountOfTokensAfterThrottling >= 0
+ ? targetAmountOfTokensAfterThrottling
+ : (long) (capacity * DEFAULT_TARGET_FILL_FACTOR_AFTER_THROTTLING);
this.tokens = initialTokens;
getTokens();
}
@@ -62,4 +66,8 @@ class FinalRateAsyncTokenBucket extends AsyncTokenBucket {
return rate;
}
+ @Override
+ protected long getNanosForOneToken() {
+ return nanosForOneToken;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/FinalRateAsyncTokenBucketBuilder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/FinalRateAsyncTokenBucketBuilder.java
index a292000eaa8..0bedf5db88d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/FinalRateAsyncTokenBucketBuilder.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/FinalRateAsyncTokenBucketBuilder.java
@@ -26,6 +26,7 @@ public class FinalRateAsyncTokenBucketBuilder
extends AsyncTokenBucketBuilder<FinalRateAsyncTokenBucketBuilder> {
protected Long capacity;
protected Long initialTokens;
+ protected Long targetAmountOfTokensAfterThrottling;
protected Long rate;
protected long ratePeriodNanos = AsyncTokenBucket.ONE_SECOND_NANOS;
@@ -52,11 +53,17 @@ public class FinalRateAsyncTokenBucketBuilder
return this;
}
+ public FinalRateAsyncTokenBucketBuilder targetAmountOfTokensAfterThrottling(
+ long targetAmountOfTokensAfterThrottling) {
+ this.targetAmountOfTokensAfterThrottling = targetAmountOfTokensAfterThrottling;
+ return this;
+ }
+
public AsyncTokenBucket build() {
return new FinalRateAsyncTokenBucket(this.capacity != null ? this.capacity : this.rate, this.rate,
- this.clock,
- this.ratePeriodNanos, this.resolutionNanos, this.consistentConsumedTokens, this.consistentAddedTokens,
- this.initialTokens != null ? this.initialTokens : this.rate
+ this.clock, this.addTokensResolutionNanos,
+ this.ratePeriodNanos, this.initialTokens != null ? this.initialTokens : this.rate,
+ this.targetAmountOfTokensAfterThrottling != null ? this.targetAmountOfTokensAfterThrottling : -1
);
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/MonotonicSnapshotClock.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/MonotonicClock.java
similarity index 51%
rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/MonotonicSnapshotClock.java
rename to pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/MonotonicClock.java
index 8f61bd5125b..6878aa7f148 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/MonotonicSnapshotClock.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/MonotonicClock.java
@@ -27,32 +27,12 @@ package org.apache.pulsar.broker.qos;
* The monotonic clock is not related to the wall-clock time and is not affected by changes to the system time.
* The tick value is only significant when compared to other values obtained from the same clock source
* and should not be used for other purposes.
- *
- * This interface assumes that the implementation can be implemented in a granular way. This means that the value is
- * advanced in steps of a configurable resolution that snapshots the underlying high precision monotonic clock source
- * value.
- * This design allows for optimizations that can improve performance on platforms where obtaining the value of a
- * platform monotonic clock is relatively expensive.
*/
-public interface MonotonicSnapshotClock {
+public interface MonotonicClock {
/**
* Retrieves the latest snapshot of the tick value of the monotonic clock in nanoseconds.
*
- * When requestSnapshot is set to true, the method will snapshot the underlying high-precision monotonic clock
- * source so that the latest snapshot value is as accurate as possible. This may be a relatively expensive
- * compared to a non-snapshot request.
- *
- * When requestSnapshot is set to false, the method will return the latest snapshot value which is updated by
- * either a call that requested a snapshot or by an update thread that is configured to update the snapshot value
- * periodically.
- *
- * This method returns a value that is guaranteed to be monotonic, meaning it will always increase or remain the
- * same, never decrease. The returned value is only significant when compared to other values obtained from the same
- * clock source and should not be used for other purposes.
- *
- * @param requestSnapshot If set to true, the method will request a new snapshot from the underlying more
- * high-precision monotonic clock.
* @return The current tick value of the monotonic clock in nanoseconds.
*/
- long getTickNanos(boolean requestSnapshot);
+ long getTickNanos();
}
\ No newline at end of file
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroup.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroup.java
index 541a645f18b..55cda0ca9dd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroup.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroup.java
@@ -83,7 +83,7 @@ public class ResourceGroup {
this.setResourceGroupConfigParameters(rgConfig);
this.setDefaultResourceUsageTransportHandlers();
this.resourceGroupPublishLimiter = new ResourceGroupPublishLimiter(rgConfig, rgs.getPulsar()
- .getMonotonicSnapshotClock());
+ .getMonotonicClock());
log.info("attaching publish rate limiter {} to {} get {}", this.resourceGroupPublishLimiter, name,
this.getResourceGroupPublishLimiter());
}
@@ -99,7 +99,7 @@ public class ResourceGroup {
this.setResourceGroupMonitoringClassFields();
this.setResourceGroupConfigParameters(rgConfig);
this.resourceGroupPublishLimiter = new ResourceGroupPublishLimiter(rgConfig, rgs.getPulsar()
- .getMonotonicSnapshotClock());
+ .getMonotonicClock());
this.ruPublisher = rgPublisher;
this.ruConsumer = rgConsumer;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java
index a733db555a3..fc4514db81f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java
@@ -18,7 +18,7 @@
*/
package org.apache.pulsar.broker.resourcegroup;
-import org.apache.pulsar.broker.qos.MonotonicSnapshotClock;
+import org.apache.pulsar.broker.qos.MonotonicClock;
import org.apache.pulsar.broker.resourcegroup.ResourceGroup.BytesAndMessagesCount;
import org.apache.pulsar.broker.service.PublishRateLimiterImpl;
import org.apache.pulsar.common.policies.data.Policies;
@@ -29,8 +29,8 @@ public class ResourceGroupPublishLimiter extends PublishRateLimiterImpl {
private volatile long publishMaxMessageRate;
private volatile long publishMaxByteRate;
- public ResourceGroupPublishLimiter(ResourceGroup resourceGroup, MonotonicSnapshotClock monotonicSnapshotClock) {
- super(monotonicSnapshotClock);
+ public ResourceGroupPublishLimiter(ResourceGroup resourceGroup, MonotonicClock monotonicClock) {
+ super(monotonicClock);
update(resourceGroup);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
index fb5c457fcc8..015a340ef21 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.broker.service;
import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.andAckSet;
import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.isAckSetEmpty;
+import static org.apache.pulsar.broker.service.persistent.PersistentTopic.MESSAGE_RATE_BACKOFF_MS;
import io.netty.buffer.ByteBuf;
import io.prometheus.client.Gauge;
import java.util.ArrayList;
@@ -35,6 +36,7 @@ import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.bookkeeper.mledger.impl.AckSetStateUtil;
import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.intercept.BrokerInterceptor;
@@ -368,25 +370,63 @@ public abstract class AbstractBaseDispatcher extends EntryFilterSupport implemen
protected abstract void reScheduleRead();
- protected Pair<Integer, Long> updateMessagesToRead(DispatchRateLimiter dispatchRateLimiter,
- int messagesToRead, long bytesToRead) {
- // update messagesToRead according to available dispatch rate limit.
- return computeReadLimits(messagesToRead,
- (int) dispatchRateLimiter.getAvailableDispatchRateLimitOnMsg(),
- bytesToRead, dispatchRateLimiter.getAvailableDispatchRateLimitOnByte());
+ public abstract String getName();
+
+ protected boolean hasAnyDispatchRateLimiter() {
+ return subscription.getTopic().getBrokerDispatchRateLimiter().isPresent()
+ || subscription.getTopic().getDispatchRateLimiter().isPresent()
+ || getRateLimiter().isPresent();
}
- protected static Pair<Integer, Long> computeReadLimits(int messagesToRead, int availablePermitsOnMsg,
- long bytesToRead, long availablePermitsOnByte) {
- if (availablePermitsOnMsg >= 0) {
- messagesToRead = Math.min(messagesToRead, availablePermitsOnMsg);
+ protected Pair<Integer, Long> applyRateLimitsToMessagesAndBytesToRead(int messagesToRead, long bytesToRead) {
+ MutablePair<Integer, Long> readLimits = new MutablePair<>(messagesToRead, bytesToRead);
+ Topic topic = subscription.getTopic();
+
+ boolean success = true;
+
+ if (topic.getBrokerDispatchRateLimiter().isPresent()) {
+ success = applyDispatchRateLimitsToReadLimits(topic.getBrokerDispatchRateLimiter().get(), readLimits,
+ DispatchRateLimiter.Type.BROKER);
}
- if (availablePermitsOnByte >= 0) {
- bytesToRead = Math.min(bytesToRead, availablePermitsOnByte);
+ if (success && topic.getDispatchRateLimiter().isPresent()) {
+ success = applyDispatchRateLimitsToReadLimits(topic.getDispatchRateLimiter().get(), readLimits,
+ DispatchRateLimiter.Type.TOPIC);
+ }
+
+ if (success && getRateLimiter().isPresent()) {
+ success = applyDispatchRateLimitsToReadLimits(getRateLimiter().get(), readLimits,
+ DispatchRateLimiter.Type.SUBSCRIPTION);
}
- return Pair.of(messagesToRead, bytesToRead);
+ return readLimits;
+ }
+
+ private boolean applyDispatchRateLimitsToReadLimits(DispatchRateLimiter rateLimiter,
+ MutablePair<Integer, Long> readLimits,
+ DispatchRateLimiter.Type limiterType) {
+ // update messagesToRead according to available dispatch rate limit.
+ int availablePermitsOnMsg = (int) rateLimiter.getAvailableDispatchRateLimitOnMsg();
+ if (availablePermitsOnMsg >= 0) {
+ readLimits.setLeft(Math.min(readLimits.getLeft(), availablePermitsOnMsg));
+ }
+ long availablePermitsOnByte = rateLimiter.getAvailableDispatchRateLimitOnByte();
+ if (availablePermitsOnByte >= 0) {
+ readLimits.setRight(Math.min(readLimits.getRight(), availablePermitsOnByte));
+ }
+ if (readLimits.getLeft() == 0 || readLimits.getRight() == 0) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] message-read exceeded {} message-rate {}/{}, schedule after {}ms", getName(),
+ limiterType.name().toLowerCase(),
+ rateLimiter.getDispatchRateOnMsg(), rateLimiter.getDispatchRateOnByte(),
+ MESSAGE_RATE_BACKOFF_MS);
+ }
+ reScheduleRead();
+ readLimits.setLeft(-1);
+ readLimits.setRight(-1L);
+ return false;
+ }
+ return true;
}
protected byte[] peekStickyKey(ByteBuf metadataAndPayload) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 9a115e6d1ca..f6397f21962 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -194,7 +194,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
this.lastActive = System.nanoTime();
this.preciseTopicPublishRateLimitingEnable = config.isPreciseTopicPublishRateLimiterEnable();
- topicPublishRateLimiter = new PublishRateLimiterImpl(brokerService.getPulsar().getMonotonicSnapshotClock());
+ topicPublishRateLimiter = new PublishRateLimiterImpl(brokerService.getPulsar().getMonotonicClock());
updateActiveRateLimiters();
additionalSystemCursorNames = brokerService.pulsar().getConfiguration().getAdditionalSystemCursorNames();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 29bb81afa3c..3812d03f72b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -132,6 +132,8 @@ import org.apache.pulsar.broker.service.nonpersistent.NonPersistentSystemTopic;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
+import org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactory;
+import org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryClassic;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.persistent.SystemTopic;
import org.apache.pulsar.broker.service.plugin.EntryFilterProvider;
@@ -283,6 +285,7 @@ public class BrokerService implements Closeable {
private final ScheduledExecutorService consumedLedgersMonitor;
private ScheduledExecutorService deduplicationSnapshotMonitor;
protected final PublishRateLimiter brokerPublishRateLimiter;
+ private final DispatchRateLimiterFactory dispatchRateLimiterFactory;
protected volatile DispatchRateLimiter brokerDispatchRateLimiter = null;
private DistributedIdGenerator producerNameGenerator;
@@ -336,9 +339,10 @@ public class BrokerService implements Closeable {
this.pulsar = pulsar;
this.clock = pulsar.getClock();
this.dynamicConfigurationMap = prepareDynamicConfigurationMap();
- this.brokerPublishRateLimiter = new PublishRateLimiterImpl(pulsar.getMonotonicSnapshotClock());
+ this.brokerPublishRateLimiter = new PublishRateLimiterImpl(pulsar.getMonotonicClock());
this.preciseTopicPublishRateLimitingEnable =
pulsar.getConfiguration().isPreciseTopicPublishRateLimiterEnable();
+ this.dispatchRateLimiterFactory = createDispatchRateLimiterFactory(pulsar.getConfig());
this.managedLedgerStorage = pulsar.getManagedLedgerStorage();
this.keepAliveIntervalSeconds = pulsar.getConfiguration().getKeepAliveIntervalSeconds();
this.pendingTopicLoadingQueue = Queues.newConcurrentLinkedQueue();
@@ -474,6 +478,22 @@ public class BrokerService implements Closeable {
this.bundlesQuotas = new BundlesQuotas(pulsar);
}
+ protected DispatchRateLimiterFactory createDispatchRateLimiterFactory(ServiceConfiguration config)
+ throws ClassNotFoundException, InstantiationException, IllegalAccessException {
+ String dispatchRateLimiterFactoryClassName = config.getDispatchRateLimiterFactoryClassName();
+ if (isNotBlank(dispatchRateLimiterFactoryClassName)) {
+ try {
+ return (DispatchRateLimiterFactory) Class.forName(dispatchRateLimiterFactoryClassName).newInstance();
+ } catch (Exception e) {
+ log.warn("Failed to initialize dispatch rate limiter factory class {}",
+ dispatchRateLimiterFactoryClassName, e);
+ throw e;
+ }
+ } else {
+ return new DispatchRateLimiterFactoryClassic();
+ }
+ }
+
private int getPendingLookupRequest() {
return pulsar.getConfig().getMaxConcurrentLookupRequest() - lookupRequestSemaphore.get().availablePermits();
}
@@ -2919,7 +2939,7 @@ public class BrokerService implements Closeable {
private void updateBrokerDispatchThrottlingMaxRate() {
if (brokerDispatchRateLimiter == null) {
- brokerDispatchRateLimiter = new DispatchRateLimiter(this);
+ brokerDispatchRateLimiter = dispatchRateLimiterFactory.createBrokerDispatchRateLimiter(this);
} else {
brokerDispatchRateLimiter.updateDispatchRate();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java
index 90c8de5f97a..0015f2675a2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java
@@ -26,7 +26,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.qos.AsyncTokenBucket;
-import org.apache.pulsar.broker.qos.MonotonicSnapshotClock;
+import org.apache.pulsar.broker.qos.MonotonicClock;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.PublishRate;
import org.jctools.queues.MessagePassingQueue;
@@ -36,15 +36,15 @@ import org.jctools.queues.MpscUnboundedArrayQueue;
public class PublishRateLimiterImpl implements PublishRateLimiter {
private volatile AsyncTokenBucket tokenBucketOnMessage;
private volatile AsyncTokenBucket tokenBucketOnByte;
- private final MonotonicSnapshotClock monotonicSnapshotClock;
+ private final MonotonicClock monotonicClock;
private final MessagePassingQueue<Producer> unthrottlingQueue = new MpscUnboundedArrayQueue<>(1024);
private final AtomicInteger throttledProducersCount = new AtomicInteger(0);
private final AtomicBoolean processingQueuedProducers = new AtomicBoolean(false);
- public PublishRateLimiterImpl(MonotonicSnapshotClock monotonicSnapshotClock) {
- this.monotonicSnapshotClock = monotonicSnapshotClock;
+ public PublishRateLimiterImpl(MonotonicClock monotonicClock) {
+ this.monotonicClock = monotonicClock;
}
/**
@@ -172,13 +172,13 @@ public class PublishRateLimiterImpl implements PublishRateLimiter {
protected void updateTokenBuckets(long publishThrottlingRateInMsg, long publishThrottlingRateInByte) {
if (publishThrottlingRateInMsg > 0) {
tokenBucketOnMessage =
- AsyncTokenBucket.builder().rate(publishThrottlingRateInMsg).clock(monotonicSnapshotClock).build();
+ AsyncTokenBucket.builder().rate(publishThrottlingRateInMsg).clock(monotonicClock).build();
} else {
tokenBucketOnMessage = null;
}
if (publishThrottlingRateInByte > 0) {
tokenBucketOnByte =
- AsyncTokenBucket.builder().rate(publishThrottlingRateInByte).clock(monotonicSnapshotClock).build();
+ AsyncTokenBucket.builder().rate(publishThrottlingRateInByte).clock(monotonicClock).build();
} else {
tokenBucketOnByte = null;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java
index 399a524a197..5941093e71c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java
@@ -23,6 +23,7 @@ import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import lombok.Getter;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers;
@@ -51,6 +52,7 @@ public class NonPersistentDispatcherMultipleConsumers extends AbstractDispatcher
protected final Subscription subscription;
private CompletableFuture<Void> closeFuture = null;
+ @Getter
protected final String name;
protected final Rate msgDrop;
protected static final AtomicIntegerFieldUpdater<NonPersistentDispatcherMultipleConsumers>
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
index ec9b7ac40ce..ec83b4590b7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.service.nonpersistent;
import java.util.List;
+import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer;
@@ -38,6 +39,8 @@ public final class NonPersistentDispatcherSingleActiveConsumer extends AbstractD
private final NonPersistentTopic topic;
private final Rate msgDrop;
+ @Getter
+ protected final String name;
private final Subscription subscription;
private final RedeliveryTracker redeliveryTracker;
@@ -47,6 +50,7 @@ public final class NonPersistentDispatcherSingleActiveConsumer extends AbstractD
topic.getBrokerService().pulsar().getConfiguration(), null);
this.topic = topic;
this.subscription = subscription;
+ this.name = topic.getName() + " / " + subscription.getName();
this.msgDrop = new Rate();
this.redeliveryTracker = RedeliveryTrackerDisabled.REDELIVERY_TRACKER_DISABLED;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
index f43b134eb12..b78af75a516 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
@@ -20,10 +20,7 @@ package org.apache.pulsar.broker.service.persistent;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
import org.apache.pulsar.broker.ServiceConfiguration;
-import org.apache.pulsar.broker.qos.AsyncTokenBucket;
-import org.apache.pulsar.broker.qos.AsyncTokenBucketBuilder;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
@@ -32,7 +29,7 @@ import org.apache.pulsar.common.policies.data.Policies;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class DispatchRateLimiter {
+public abstract class DispatchRateLimiter {
public enum Type {
TOPIC,
SUBSCRIPTION,
@@ -40,20 +37,14 @@ public class DispatchRateLimiter {
BROKER
}
- private final PersistentTopic topic;
- private final String topicName;
- private final String subscriptionName;
- private final Type type;
+ protected final PersistentTopic topic;
+ protected final String topicName;
+ protected final String subscriptionName;
+ protected final Type type;
- private final BrokerService brokerService;
- private volatile AsyncTokenBucket dispatchRateLimiterOnMessage;
- private volatile AsyncTokenBucket dispatchRateLimiterOnByte;
+ protected final BrokerService brokerService;
- public DispatchRateLimiter(PersistentTopic topic, Type type) {
- this(topic, null, type);
- }
-
- public DispatchRateLimiter(PersistentTopic topic, String subscriptionName, Type type) {
+ protected DispatchRateLimiter(PersistentTopic topic, String subscriptionName, Type type) {
this.topic = topic;
this.topicName = topic.getName();
this.subscriptionName = subscriptionName;
@@ -62,7 +53,7 @@ public class DispatchRateLimiter {
updateDispatchRate();
}
- public DispatchRateLimiter(BrokerService brokerService) {
+ protected DispatchRateLimiter(BrokerService brokerService) {
this.topic = null;
this.topicName = null;
this.subscriptionName = null;
@@ -76,21 +67,14 @@ public class DispatchRateLimiter {
*
* @return
*/
- public long getAvailableDispatchRateLimitOnMsg() {
- AsyncTokenBucket localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
- return localDispatchRateLimiterOnMessage == null ? -1 :
- Math.max(localDispatchRateLimiterOnMessage.getTokens(), 0);
- }
+ public abstract long getAvailableDispatchRateLimitOnMsg();
/**
* returns available byte-permit if msg-dispatch-throttling is enabled else it returns -1.
*
* @return
*/
- public long getAvailableDispatchRateLimitOnByte() {
- AsyncTokenBucket localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
- return localDispatchRateLimiterOnByte == null ? -1 : Math.max(localDispatchRateLimiterOnByte.getTokens(), 0);
- }
+ public abstract long getAvailableDispatchRateLimitOnByte();
/**
* It acquires msg and bytes permits from rate-limiter and returns if acquired permits succeed.
@@ -98,25 +82,14 @@ public class DispatchRateLimiter {
* @param numberOfMessages
* @param byteSize
*/
- public void consumeDispatchQuota(long numberOfMessages, long byteSize) {
- AsyncTokenBucket localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
- if (numberOfMessages > 0 && localDispatchRateLimiterOnMessage != null) {
- localDispatchRateLimiterOnMessage.consumeTokens(numberOfMessages);
- }
- AsyncTokenBucket localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
- if (byteSize > 0 && localDispatchRateLimiterOnByte != null) {
- localDispatchRateLimiterOnByte.consumeTokens(byteSize);
- }
- }
+ public abstract void consumeDispatchQuota(long numberOfMessages, long byteSize);
/**
* Checks if dispatch-rate limiting is enabled.
*
* @return
*/
- public boolean isDispatchRateLimitingEnabled() {
- return dispatchRateLimiterOnMessage != null || dispatchRateLimiterOnByte != null;
- }
+ public abstract boolean isDispatchRateLimitingEnabled();
/**
* createDispatchRate according to broker service config.
@@ -164,7 +137,7 @@ public class DispatchRateLimiter {
* Topic-level has the highest priority, then namespace-level, and finally use dispatch-throttling-rate in
* broker-level
*/
- public void updateDispatchRate() {
+ public final void updateDispatchRate() {
DispatchRate dispatchRate;
switch (type) {
case TOPIC:
@@ -215,103 +188,28 @@ public class DispatchRateLimiter {
*
* @param dispatchRate
*/
- public synchronized void updateDispatchRate(DispatchRate dispatchRate) {
- // synchronized to prevent race condition from concurrent zk-watch
- log.info("setting message-dispatch-rate {}", dispatchRate);
-
- long msgRate = dispatchRate.getDispatchThrottlingRateInMsg();
- long byteRate = dispatchRate.getDispatchThrottlingRateInByte();
- long ratePeriodNanos = TimeUnit.SECONDS.toNanos(Math.max(dispatchRate.getRatePeriodInSecond(), 1));
-
- // update msg-rateLimiter
- if (msgRate > 0) {
- if (dispatchRate.isRelativeToPublishRate()) {
- this.dispatchRateLimiterOnMessage =
- configureAsyncTokenBucket(AsyncTokenBucket.builderForDynamicRate())
- .rateFunction(() -> getRelativeDispatchRateInMsg(dispatchRate))
- .ratePeriodNanosFunction(() -> ratePeriodNanos)
- .build();
- } else {
- this.dispatchRateLimiterOnMessage =
- configureAsyncTokenBucket(AsyncTokenBucket.builder())
- .rate(msgRate).ratePeriodNanos(ratePeriodNanos)
- .build();
- }
- } else {
- this.dispatchRateLimiterOnMessage = null;
- }
-
- // update byte-rateLimiter
- if (byteRate > 0) {
- if (dispatchRate.isRelativeToPublishRate()) {
- this.dispatchRateLimiterOnByte =
- configureAsyncTokenBucket(AsyncTokenBucket.builderForDynamicRate())
- .rateFunction(() -> getRelativeDispatchRateInByte(dispatchRate))
- .ratePeriodNanosFunction(() -> ratePeriodNanos)
- .build();
- } else {
- this.dispatchRateLimiterOnByte =
- configureAsyncTokenBucket(AsyncTokenBucket.builder())
- .rate(byteRate).ratePeriodNanos(ratePeriodNanos)
- .build();
- }
- } else {
- this.dispatchRateLimiterOnByte = null;
- }
- }
-
- private <T extends AsyncTokenBucketBuilder<T>> T configureAsyncTokenBucket(T builder) {
- builder.clock(brokerService.getPulsar().getMonotonicSnapshotClock());
- return builder;
- }
-
- private long getRelativeDispatchRateInMsg(DispatchRate dispatchRate) {
- return (topic != null && dispatchRate != null)
- ? (long) topic.getLastUpdatedAvgPublishRateInMsg() + dispatchRate.getDispatchThrottlingRateInMsg()
- : 0;
- }
-
- private long getRelativeDispatchRateInByte(DispatchRate dispatchRate) {
- return (topic != null && dispatchRate != null)
- ? (long) topic.getLastUpdatedAvgPublishRateInByte() + dispatchRate.getDispatchThrottlingRateInByte()
- : 0;
- }
+ public abstract void updateDispatchRate(DispatchRate dispatchRate);
/**
* Get configured msg dispatch-throttling rate. Returns -1 if not configured
*
* @return
*/
- public long getDispatchRateOnMsg() {
- AsyncTokenBucket localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
- return localDispatchRateLimiterOnMessage != null ? localDispatchRateLimiterOnMessage.getRate() : -1;
- }
+ public abstract long getDispatchRateOnMsg();
/**
* Get configured byte dispatch-throttling rate. Returns -1 if not configured
*
* @return
*/
- public long getDispatchRateOnByte() {
- AsyncTokenBucket localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
- return localDispatchRateLimiterOnByte != null ? localDispatchRateLimiterOnByte.getRate() : -1;
- }
+ public abstract long getDispatchRateOnByte();
+ public abstract void close();
public static boolean isDispatchRateEnabled(DispatchRate dispatchRate) {
return dispatchRate != null && (dispatchRate.getDispatchThrottlingRateInMsg() > 0
|| dispatchRate.getDispatchThrottlingRateInByte() > 0);
}
- public void close() {
- // close rate-limiter
- if (dispatchRateLimiterOnMessage != null) {
- dispatchRateLimiterOnMessage = null;
- }
- if (dispatchRateLimiterOnByte != null) {
- dispatchRateLimiterOnByte = null;
- }
- }
-
private static final Logger log = LoggerFactory.getLogger(DispatchRateLimiter.class);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterAsyncTokenBucketImpl.java
similarity index 55%
copy from pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
copy to pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterAsyncTokenBucketImpl.java
index f43b134eb12..9c8aca0e936 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterAsyncTokenBucketImpl.java
@@ -18,57 +18,28 @@
*/
package org.apache.pulsar.broker.service.persistent;
-import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
-import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.apache.pulsar.broker.qos.AsyncTokenBucketBuilder;
import org.apache.pulsar.broker.service.BrokerService;
-import org.apache.pulsar.common.naming.NamespaceName;
-import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.DispatchRate;
-import org.apache.pulsar.common.policies.data.Policies;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class DispatchRateLimiter {
- public enum Type {
- TOPIC,
- SUBSCRIPTION,
- REPLICATOR,
- BROKER
- }
-
- private final PersistentTopic topic;
- private final String topicName;
- private final String subscriptionName;
- private final Type type;
-
- private final BrokerService brokerService;
+public class DispatchRateLimiterAsyncTokenBucketImpl extends DispatchRateLimiter {
private volatile AsyncTokenBucket dispatchRateLimiterOnMessage;
private volatile AsyncTokenBucket dispatchRateLimiterOnByte;
- public DispatchRateLimiter(PersistentTopic topic, Type type) {
+ public DispatchRateLimiterAsyncTokenBucketImpl(PersistentTopic topic, Type type) {
this(topic, null, type);
}
- public DispatchRateLimiter(PersistentTopic topic, String subscriptionName, Type type) {
- this.topic = topic;
- this.topicName = topic.getName();
- this.subscriptionName = subscriptionName;
- this.brokerService = topic.getBrokerService();
- this.type = type;
- updateDispatchRate();
+ public DispatchRateLimiterAsyncTokenBucketImpl(PersistentTopic topic, String subscriptionName, Type type) {
+ super(topic, subscriptionName, type);
}
- public DispatchRateLimiter(BrokerService brokerService) {
- this.topic = null;
- this.topicName = null;
- this.subscriptionName = null;
- this.brokerService = brokerService;
- this.type = Type.BROKER;
- updateDispatchRate();
+ public DispatchRateLimiterAsyncTokenBucketImpl(BrokerService brokerService) {
+ super(brokerService);
}
/**
@@ -76,6 +47,7 @@ public class DispatchRateLimiter {
*
* @return
*/
+ @Override
public long getAvailableDispatchRateLimitOnMsg() {
AsyncTokenBucket localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
return localDispatchRateLimiterOnMessage == null ? -1 :
@@ -87,6 +59,7 @@ public class DispatchRateLimiter {
*
* @return
*/
+ @Override
public long getAvailableDispatchRateLimitOnByte() {
AsyncTokenBucket localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
return localDispatchRateLimiterOnByte == null ? -1 : Math.max(localDispatchRateLimiterOnByte.getTokens(), 0);
@@ -98,6 +71,7 @@ public class DispatchRateLimiter {
* @param numberOfMessages
* @param byteSize
*/
+ @Override
public void consumeDispatchQuota(long numberOfMessages, long byteSize) {
AsyncTokenBucket localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
if (numberOfMessages > 0 && localDispatchRateLimiterOnMessage != null) {
@@ -114,107 +88,18 @@ public class DispatchRateLimiter {
*
* @return
*/
+ @Override
public boolean isDispatchRateLimitingEnabled() {
return dispatchRateLimiterOnMessage != null || dispatchRateLimiterOnByte != null;
}
- /**
- * createDispatchRate according to broker service config.
- *
- * @return
- */
- private DispatchRate createDispatchRate() {
- int dispatchThrottlingRateInMsg;
- long dispatchThrottlingRateInByte;
- ServiceConfiguration config = brokerService.pulsar().getConfiguration();
-
- switch (type) {
- case TOPIC:
- dispatchThrottlingRateInMsg = config.getDispatchThrottlingRatePerTopicInMsg();
- dispatchThrottlingRateInByte = config.getDispatchThrottlingRatePerTopicInByte();
- break;
- case SUBSCRIPTION:
- dispatchThrottlingRateInMsg = config.getDispatchThrottlingRatePerSubscriptionInMsg();
- dispatchThrottlingRateInByte = config.getDispatchThrottlingRatePerSubscriptionInByte();
- break;
- case REPLICATOR:
- dispatchThrottlingRateInMsg = config.getDispatchThrottlingRatePerReplicatorInMsg();
- dispatchThrottlingRateInByte = config.getDispatchThrottlingRatePerReplicatorInByte();
- break;
- case BROKER:
- dispatchThrottlingRateInMsg = config.getDispatchThrottlingRateInMsg();
- dispatchThrottlingRateInByte = config.getDispatchThrottlingRateInByte();
- break;
- default:
- dispatchThrottlingRateInMsg = -1;
- dispatchThrottlingRateInByte = -1;
- }
-
-
- return DispatchRate.builder()
- .dispatchThrottlingRateInMsg(dispatchThrottlingRateInMsg)
- .dispatchThrottlingRateInByte(dispatchThrottlingRateInByte)
- .ratePeriodInSecond(1)
- .relativeToPublishRate(type != Type.BROKER && config.isDispatchThrottlingRateRelativeToPublishRate())
- .build();
- }
-
- /**
- * Update dispatch-throttling-rate.
- * Topic-level has the highest priority, then namespace-level, and finally use dispatch-throttling-rate in
- * broker-level
- */
- public void updateDispatchRate() {
- DispatchRate dispatchRate;
- switch (type) {
- case TOPIC:
- dispatchRate = topic.getDispatchRate();
- break;
- case SUBSCRIPTION:
- dispatchRate = topic.getSubscriptionDispatchRate(subscriptionName);
- break;
- case REPLICATOR:
- dispatchRate = topic.getReplicatorDispatchRate();
- break;
- case BROKER:
- dispatchRate = createDispatchRate();
- break;
- default:
- log.warn("ignore configured dispatch rate for type {}", type);
- return;
- }
- if (type == Type.BROKER) {
- log.info("configured broker message-dispatch rate {}", dispatchRate);
- } else {
- log.info("[{}] configured {} message-dispatch rate at broker {} subscriptionName [{}]",
- this.topicName, type, subscriptionName == null ? "null" : subscriptionName, dispatchRate);
- }
- updateDispatchRate(dispatchRate);
- }
-
- public static CompletableFuture<Optional<Policies>> getPoliciesAsync(BrokerService brokerService,
- String topicName) {
- final NamespaceName namespace = TopicName.get(topicName).getNamespaceObject();
- return brokerService.pulsar().getPulsarResources().getNamespaceResources().getPoliciesAsync(namespace);
- }
-
- /**
- * @deprecated Avoid using the deprecated method
- * #{@link org.apache.pulsar.broker.resources.NamespaceResources#getPoliciesIfCached(NamespaceName)} and blocking
- * call. we can use #{@link DispatchRateLimiter#getPoliciesAsync(BrokerService, String)} to instead of it.
- */
- @Deprecated
- public static Optional<Policies> getPolicies(BrokerService brokerService, String topicName) {
- final NamespaceName namespace = TopicName.get(topicName).getNamespaceObject();
- return brokerService.pulsar().getPulsarResources().getNamespaceResources().getPoliciesIfCached(namespace);
- }
-
/**
* Update dispatch rate by updating msg and byte rate-limiter. If dispatch-rate is configured < 0 then it closes
* the rate-limiter and disables appropriate rate-limiter.
*
* @param dispatchRate
*/
+ @Override
public synchronized void updateDispatchRate(DispatchRate dispatchRate) {
// synchronized to prevent race condition from concurrent zk-watch
log.info("setting message-dispatch-rate {}", dispatchRate);
@@ -227,13 +112,13 @@ public class DispatchRateLimiter {
if (msgRate > 0) {
if (dispatchRate.isRelativeToPublishRate()) {
this.dispatchRateLimiterOnMessage =
- configureAsyncTokenBucket(AsyncTokenBucket.builderForDynamicRate())
+ configureAsyncTokenBucket(AsyncTokenBucket.builderForDynamicRate(), ratePeriodNanos)
.rateFunction(() -> getRelativeDispatchRateInMsg(dispatchRate))
.ratePeriodNanosFunction(() -> ratePeriodNanos)
.build();
} else {
this.dispatchRateLimiterOnMessage =
- configureAsyncTokenBucket(AsyncTokenBucket.builder())
+ configureAsyncTokenBucket(AsyncTokenBucket.builder(), ratePeriodNanos)
.rate(msgRate).ratePeriodNanos(ratePeriodNanos)
.build();
}
@@ -245,13 +130,13 @@ public class DispatchRateLimiter {
if (byteRate > 0) {
if (dispatchRate.isRelativeToPublishRate()) {
this.dispatchRateLimiterOnByte =
- configureAsyncTokenBucket(AsyncTokenBucket.builderForDynamicRate())
+ configureAsyncTokenBucket(AsyncTokenBucket.builderForDynamicRate(), ratePeriodNanos)
.rateFunction(() -> getRelativeDispatchRateInByte(dispatchRate))
.ratePeriodNanosFunction(() -> ratePeriodNanos)
.build();
} else {
this.dispatchRateLimiterOnByte =
- configureAsyncTokenBucket(AsyncTokenBucket.builder())
+ configureAsyncTokenBucket(AsyncTokenBucket.builder(), ratePeriodNanos)
.rate(byteRate).ratePeriodNanos(ratePeriodNanos)
.build();
}
@@ -260,8 +145,13 @@ public class DispatchRateLimiter {
}
}
- private <T extends AsyncTokenBucketBuilder<T>> T configureAsyncTokenBucket(T builder) {
- builder.clock(brokerService.getPulsar().getMonotonicSnapshotClock());
+ private <T extends AsyncTokenBucketBuilder<T>> T configureAsyncTokenBucket(T builder,
+ long addTokensResolutionNanos) {
+ builder.clock(brokerService.getPulsar().getMonotonicClock());
+ // configures tokens to be added once in every addTokensResolutionNanos
+ // this makes AsyncTokenBucket behave in the similar way as the "classic" dispatch rate limiter implementation
+ // which uses a scheduled task to add tokens to the rate limiter once in every ratePeriod
+ builder.addTokensResolutionNanos(addTokensResolutionNanos);
return builder;
}
@@ -282,6 +172,7 @@ public class DispatchRateLimiter {
*
* @return
*/
+ @Override
public long getDispatchRateOnMsg() {
AsyncTokenBucket localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
return localDispatchRateLimiterOnMessage != null ? localDispatchRateLimiterOnMessage.getRate() : -1;
@@ -292,17 +183,14 @@ public class DispatchRateLimiter {
*
* @return
*/
+ @Override
public long getDispatchRateOnByte() {
AsyncTokenBucket localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
return localDispatchRateLimiterOnByte != null ? localDispatchRateLimiterOnByte.getRate() : -1;
}
- public static boolean isDispatchRateEnabled(DispatchRate dispatchRate) {
- return dispatchRate != null && (dispatchRate.getDispatchThrottlingRateInMsg() > 0
- || dispatchRate.getDispatchThrottlingRateInByte() > 0);
- }
-
+ @Override
public void close() {
// close rate-limiter
if (dispatchRateLimiterOnMessage != null) {
@@ -313,5 +201,5 @@ public class DispatchRateLimiter {
}
}
- private static final Logger log = LoggerFactory.getLogger(DispatchRateLimiter.class);
+ private static final Logger log = LoggerFactory.getLogger(DispatchRateLimiterAsyncTokenBucketImpl.class);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterClassicImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterClassicImpl.java
new file mode 100644
index 00000000000..2f96d385d59
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterClassicImpl.java
@@ -0,0 +1,215 @@
+/*
+ * 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.service.persistent;
+
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.common.policies.data.DispatchRate;
+import org.apache.pulsar.common.util.RateLimiter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DispatchRateLimiterClassicImpl extends DispatchRateLimiter {
+ private volatile RateLimiter dispatchRateLimiterOnMessage;
+ private volatile RateLimiter dispatchRateLimiterOnByte;
+
+ public DispatchRateLimiterClassicImpl(PersistentTopic topic, Type type) {
+ this(topic, null, type);
+ }
+
+ public DispatchRateLimiterClassicImpl(PersistentTopic topic, String subscriptionName, Type type) {
+ super(topic, subscriptionName, type);
+ }
+
+ public DispatchRateLimiterClassicImpl(BrokerService brokerService) {
+ super(brokerService);
+ }
+
+ /**
+ * returns available msg-permit if msg-dispatch-throttling is enabled else it returns -1.
+ *
+ * @return
+ */
+ @Override
+ public long getAvailableDispatchRateLimitOnMsg() {
+ RateLimiter localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
+ return localDispatchRateLimiterOnMessage == null ? -1 :
+ Math.max(localDispatchRateLimiterOnMessage.getAvailablePermits(), 0);
+ }
+
+ /**
+ * returns available byte-permit if msg-dispatch-throttling is enabled else it returns -1.
+ *
+ * @return
+ */
+ @Override
+ public long getAvailableDispatchRateLimitOnByte() {
+ RateLimiter localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
+ return localDispatchRateLimiterOnByte == null ? -1 :
+ Math.max(localDispatchRateLimiterOnByte.getAvailablePermits(), 0);
+ }
+
+ /**
+ * It acquires msg and bytes permits from rate-limiter and returns if acquired permits succeed.
+ *
+ * @param numberOfMessages
+ * @param byteSize
+ */
+ @Override
+ public void consumeDispatchQuota(long numberOfMessages, long byteSize) {
+ RateLimiter localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
+ if (numberOfMessages > 0 && localDispatchRateLimiterOnMessage != null) {
+ localDispatchRateLimiterOnMessage.tryAcquire(numberOfMessages);
+ }
+ RateLimiter localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
+ if (byteSize > 0 && localDispatchRateLimiterOnByte != null) {
+ localDispatchRateLimiterOnByte.tryAcquire(byteSize);
+ }
+ }
+
+ /**
+ * Checks if dispatch-rate limiting is enabled.
+ *
+ * @return
+ */
+ @Override
+ public boolean isDispatchRateLimitingEnabled() {
+ return dispatchRateLimiterOnMessage != null || dispatchRateLimiterOnByte != null;
+ }
+
+ /**
+ * Update dispatch rate by updating msg and byte rate-limiter. If dispatch-rate is configured < 0 then it closes
+ * the rate-limiter and disables appropriate rate-limiter.
+ *
+ * @param dispatchRate
+ */
+ @Override
+ public synchronized void updateDispatchRate(DispatchRate dispatchRate) {
+ // synchronized to prevent race condition from concurrent zk-watch
+ log.info("setting message-dispatch-rate {}", dispatchRate);
+
+ long msgRate = dispatchRate.getDispatchThrottlingRateInMsg();
+ long byteRate = dispatchRate.getDispatchThrottlingRateInByte();
+ long ratePeriod = dispatchRate.getRatePeriodInSecond();
+
+ Supplier<Long> permitUpdaterMsg = dispatchRate.isRelativeToPublishRate()
+ ? () -> getRelativeDispatchRateInMsg(dispatchRate)
+ : null;
+ // update msg-rateLimiter
+ if (msgRate > 0) {
+ if (this.dispatchRateLimiterOnMessage == null) {
+ this.dispatchRateLimiterOnMessage =
+ RateLimiter.builder()
+ .scheduledExecutorService(brokerService.pulsar().getExecutor())
+ .permits(msgRate)
+ .rateTime(ratePeriod)
+ .timeUnit(TimeUnit.SECONDS)
+ .permitUpdater(permitUpdaterMsg)
+ .isDispatchOrPrecisePublishRateLimiter(true)
+ .build();
+ } else {
+ this.dispatchRateLimiterOnMessage.setRate(msgRate, dispatchRate.getRatePeriodInSecond(),
+ TimeUnit.SECONDS, permitUpdaterMsg);
+ }
+ } else {
+ // message-rate should be disable and close
+ if (this.dispatchRateLimiterOnMessage != null) {
+ this.dispatchRateLimiterOnMessage.close();
+ this.dispatchRateLimiterOnMessage = null;
+ }
+ }
+
+ Supplier<Long> permitUpdaterByte = dispatchRate.isRelativeToPublishRate()
+ ? () -> getRelativeDispatchRateInByte(dispatchRate)
+ : null;
+ // update byte-rateLimiter
+ if (byteRate > 0) {
+ if (this.dispatchRateLimiterOnByte == null) {
+ this.dispatchRateLimiterOnByte =
+ RateLimiter.builder()
+ .scheduledExecutorService(brokerService.pulsar().getExecutor())
+ .permits(byteRate)
+ .rateTime(ratePeriod)
+ .timeUnit(TimeUnit.SECONDS)
+ .permitUpdater(permitUpdaterByte)
+ .isDispatchOrPrecisePublishRateLimiter(true)
+ .build();
+ } else {
+ this.dispatchRateLimiterOnByte.setRate(byteRate, dispatchRate.getRatePeriodInSecond(),
+ TimeUnit.SECONDS, permitUpdaterByte);
+ }
+ } else {
+ // message-rate should be disable and close
+ if (this.dispatchRateLimiterOnByte != null) {
+ this.dispatchRateLimiterOnByte.close();
+ this.dispatchRateLimiterOnByte = null;
+ }
+ }
+ }
+
+ private long getRelativeDispatchRateInMsg(DispatchRate dispatchRate) {
+ return (topic != null && dispatchRate != null)
+ ? (long) topic.getLastUpdatedAvgPublishRateInMsg() + dispatchRate.getDispatchThrottlingRateInMsg()
+ : 0;
+ }
+
+ private long getRelativeDispatchRateInByte(DispatchRate dispatchRate) {
+ return (topic != null && dispatchRate != null)
+ ? (long) topic.getLastUpdatedAvgPublishRateInByte() + dispatchRate.getDispatchThrottlingRateInByte()
+ : 0;
+ }
+
+ /**
+ * Get configured msg dispatch-throttling rate. Returns -1 if not configured
+ *
+ * @return
+ */
+ @Override
+ public long getDispatchRateOnMsg() {
+ RateLimiter localDispatchRateLimiterOnMessage = dispatchRateLimiterOnMessage;
+ return localDispatchRateLimiterOnMessage != null ? localDispatchRateLimiterOnMessage.getRate() : -1;
+ }
+
+ /**
+ * Get configured byte dispatch-throttling rate. Returns -1 if not configured
+ *
+ * @return
+ */
+ @Override
+ public long getDispatchRateOnByte() {
+ RateLimiter localDispatchRateLimiterOnByte = dispatchRateLimiterOnByte;
+ return localDispatchRateLimiterOnByte != null ? localDispatchRateLimiterOnByte.getRate() : -1;
+ }
+
+ @Override
+ public void close() {
+ // close rate-limiter
+ if (dispatchRateLimiterOnMessage != null) {
+ dispatchRateLimiterOnMessage.close();
+ dispatchRateLimiterOnMessage = null;
+ }
+ if (dispatchRateLimiterOnByte != null) {
+ dispatchRateLimiterOnByte.close();
+ dispatchRateLimiterOnByte = null;
+ }
+ }
+
+ private static final Logger log = LoggerFactory.getLogger(DispatchRateLimiterClassicImpl.class);
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterFactory.java
new file mode 100644
index 00000000000..35b39baebcc
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterFactory.java
@@ -0,0 +1,28 @@
+/*
+ * 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.service.persistent;
+
+import org.apache.pulsar.broker.service.BrokerService;
+
+public interface DispatchRateLimiterFactory {
+ DispatchRateLimiter createBrokerDispatchRateLimiter(BrokerService brokerService);
+ DispatchRateLimiter createTopicDispatchRateLimiter(PersistentTopic topic);
+ DispatchRateLimiter createSubscriptionDispatchRateLimiter(PersistentTopic topic, String subscriptionName);
+ DispatchRateLimiter createReplicatorDispatchRateLimiter(PersistentTopic topic, String subscriptionName);
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterFactoryAsyncTokenBucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterFactoryAsyncTokenBucket.java
new file mode 100644
index 00000000000..8061add974b
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterFactoryAsyncTokenBucket.java
@@ -0,0 +1,45 @@
+/*
+ * 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.service.persistent;
+
+import org.apache.pulsar.broker.service.BrokerService;
+
+public class DispatchRateLimiterFactoryAsyncTokenBucket implements DispatchRateLimiterFactory {
+ @Override
+ public DispatchRateLimiter createBrokerDispatchRateLimiter(BrokerService brokerService) {
+ return new DispatchRateLimiterAsyncTokenBucketImpl(brokerService);
+ }
+
+ @Override
+ public DispatchRateLimiter createTopicDispatchRateLimiter(PersistentTopic topic) {
+ return new DispatchRateLimiterAsyncTokenBucketImpl(topic, DispatchRateLimiter.Type.TOPIC);
+ }
+
+ @Override
+ public DispatchRateLimiter createSubscriptionDispatchRateLimiter(PersistentTopic topic, String subscriptionName) {
+ return new DispatchRateLimiterAsyncTokenBucketImpl(topic, subscriptionName,
+ DispatchRateLimiter.Type.SUBSCRIPTION);
+ }
+
+ @Override
+ public DispatchRateLimiter createReplicatorDispatchRateLimiter(PersistentTopic topic, String subscriptionName) {
+ return new DispatchRateLimiterAsyncTokenBucketImpl(topic, subscriptionName,
+ DispatchRateLimiter.Type.REPLICATOR);
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterFactoryClassic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterFactoryClassic.java
new file mode 100644
index 00000000000..72a96b722ba
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiterFactoryClassic.java
@@ -0,0 +1,43 @@
+/*
+ * 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.service.persistent;
+
+import org.apache.pulsar.broker.service.BrokerService;
+
+public class DispatchRateLimiterFactoryClassic implements DispatchRateLimiterFactory {
+ @Override
+ public DispatchRateLimiter createBrokerDispatchRateLimiter(BrokerService brokerService) {
+ return new DispatchRateLimiterClassicImpl(brokerService);
+ }
+
+ @Override
+ public DispatchRateLimiter createTopicDispatchRateLimiter(PersistentTopic topic) {
+ return new DispatchRateLimiterClassicImpl(topic, DispatchRateLimiter.Type.TOPIC);
+ }
+
+ @Override
+ public DispatchRateLimiter createSubscriptionDispatchRateLimiter(PersistentTopic topic, String subscriptionName) {
+ return new DispatchRateLimiterClassicImpl(topic, subscriptionName, DispatchRateLimiter.Type.SUBSCRIPTION);
+ }
+
+ @Override
+ public DispatchRateLimiter createReplicatorDispatchRateLimiter(PersistentTopic topic, String subscriptionName) {
+ return new DispatchRateLimiterClassicImpl(topic, subscriptionName, DispatchRateLimiter.Type.REPLICATOR);
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
index 0004f6516dc..6f3fe19f0a1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -68,7 +68,6 @@ import org.apache.pulsar.broker.service.RedeliveryTrackerDisabled;
import org.apache.pulsar.broker.service.SendMessageInfo;
import org.apache.pulsar.broker.service.SharedConsumerAssignor;
import org.apache.pulsar.broker.service.Subscription;
-import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type;
import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.MessageMetadata;
@@ -563,54 +562,14 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
// active-cursor reads message from cache rather from bookkeeper (2) if topic has reached message-rate
// threshold: then schedule the read after MESSAGE_RATE_BACKOFF_MS
if (serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) {
- if (topic.getBrokerDispatchRateLimiter().isPresent()) {
- DispatchRateLimiter brokerRateLimiter = topic.getBrokerDispatchRateLimiter().get();
+ if (hasAnyDispatchRateLimiter()) {
Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(brokerRateLimiter, messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
- if (messagesToRead == 0 || bytesToRead == 0) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] message-read exceeded broker message-rate {}/{}, schedule after a {}", name,
- brokerRateLimiter.getDispatchRateOnMsg(), brokerRateLimiter.getDispatchRateOnByte(),
- MESSAGE_RATE_BACKOFF_MS);
- }
- reScheduleRead();
- return Pair.of(-1, -1L);
- }
- }
-
- if (topic.getDispatchRateLimiter().isPresent()) {
- DispatchRateLimiter topicRateLimiter = topic.getDispatchRateLimiter().get();
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(topicRateLimiter, messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
- if (messagesToRead == 0 || bytesToRead == 0) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] message-read exceeded topic message-rate {}/{}, schedule after a {}", name,
- topicRateLimiter.getDispatchRateOnMsg(), topicRateLimiter.getDispatchRateOnByte(),
- MESSAGE_RATE_BACKOFF_MS);
- }
- reScheduleRead();
- return Pair.of(-1, -1L);
- }
- }
-
- if (dispatchRateLimiter.isPresent()) {
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(dispatchRateLimiter.get(), messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
- if (messagesToRead == 0 || bytesToRead == 0) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] message-read exceeded subscription message-rate {}/{}, schedule after a {}",
- name, dispatchRateLimiter.get().getDispatchRateOnMsg(),
- dispatchRateLimiter.get().getDispatchRateOnByte(),
- MESSAGE_RATE_BACKOFF_MS);
- }
- reScheduleRead();
- return Pair.of(-1, -1L);
+ applyRateLimitsToMessagesAndBytesToRead(messagesToRead, bytesToRead);
+ if (calculateToRead.getLeft() == -1 && calculateToRead.getRight() == -1) {
+ return calculateToRead;
+ } else {
+ messagesToRead = calculateToRead.getLeft();
+ bytesToRead = calculateToRead.getRight();
}
}
}
@@ -1311,7 +1270,8 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
if (!dispatchRateLimiter.isPresent() && DispatchRateLimiter.isDispatchRateEnabled(
topic.getSubscriptionDispatchRate(getSubscriptionName()))) {
this.dispatchRateLimiter =
- Optional.of(new DispatchRateLimiter(topic, getSubscriptionName(), Type.SUBSCRIPTION));
+ Optional.of(topic.getBrokerService().getDispatchRateLimiterFactory()
+ .createSubscriptionDispatchRateLimiter(topic, getSubscriptionName()));
return true;
}
return false;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
index d0f11f2687a..0f496e461b8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
@@ -66,7 +66,6 @@ import org.apache.pulsar.broker.service.SendMessageInfo;
import org.apache.pulsar.broker.service.SharedConsumerAssignor;
import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
import org.apache.pulsar.broker.service.Subscription;
-import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type;
import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.MessageMetadata;
@@ -476,54 +475,14 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
// active-cursor reads message from cache rather from bookkeeper (2) if topic has reached message-rate
// threshold: then schedule the read after MESSAGE_RATE_BACKOFF_MS
if (serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) {
- if (topic.getBrokerDispatchRateLimiter().isPresent()) {
- DispatchRateLimiter brokerRateLimiter = topic.getBrokerDispatchRateLimiter().get();
+ if (hasAnyDispatchRateLimiter()) {
Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(brokerRateLimiter, messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
- if (messagesToRead == 0 || bytesToRead == 0) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] message-read exceeded broker message-rate {}/{}, schedule after a {}", name,
- brokerRateLimiter.getDispatchRateOnMsg(), brokerRateLimiter.getDispatchRateOnByte(),
- MESSAGE_RATE_BACKOFF_MS);
- }
- reScheduleRead();
- return Pair.of(-1, -1L);
- }
- }
-
- if (topic.getDispatchRateLimiter().isPresent()) {
- DispatchRateLimiter topicRateLimiter = topic.getDispatchRateLimiter().get();
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(topicRateLimiter, messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
- if (messagesToRead == 0 || bytesToRead == 0) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] message-read exceeded topic message-rate {}/{}, schedule after a {}", name,
- topicRateLimiter.getDispatchRateOnMsg(), topicRateLimiter.getDispatchRateOnByte(),
- MESSAGE_RATE_BACKOFF_MS);
- }
- reScheduleRead();
- return Pair.of(-1, -1L);
- }
- }
-
- if (dispatchRateLimiter.isPresent()) {
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(dispatchRateLimiter.get(), messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
- if (messagesToRead == 0 || bytesToRead == 0) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] message-read exceeded subscription message-rate {}/{}, schedule after a {}",
- name, dispatchRateLimiter.get().getDispatchRateOnMsg(),
- dispatchRateLimiter.get().getDispatchRateOnByte(),
- MESSAGE_RATE_BACKOFF_MS);
- }
- reScheduleRead();
- return Pair.of(-1, -1L);
+ applyRateLimitsToMessagesAndBytesToRead(messagesToRead, bytesToRead);
+ if (calculateToRead.getLeft() == -1 && calculateToRead.getRight() == -1) {
+ return calculateToRead;
+ } else {
+ messagesToRead = calculateToRead.getLeft();
+ bytesToRead = calculateToRead.getRight();
}
}
}
@@ -1170,7 +1129,8 @@ public class PersistentDispatcherMultipleConsumersClassic extends AbstractPersis
if (!dispatchRateLimiter.isPresent() && DispatchRateLimiter.isDispatchRateEnabled(
topic.getSubscriptionDispatchRate(getSubscriptionName()))) {
this.dispatchRateLimiter =
- Optional.of(new DispatchRateLimiter(topic, getSubscriptionName(), Type.SUBSCRIPTION));
+ Optional.of(topic.getBrokerService().getDispatchRateLimiterFactory()
+ .createSubscriptionDispatchRateLimiter(topic, getSubscriptionName()));
return true;
}
return false;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index 3d910fcb606..99a09d3a5d7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -30,6 +30,7 @@ import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import lombok.Getter;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -48,7 +49,6 @@ import org.apache.pulsar.broker.service.RedeliveryTracker;
import org.apache.pulsar.broker.service.RedeliveryTrackerDisabled;
import org.apache.pulsar.broker.service.SendMessageInfo;
import org.apache.pulsar.broker.service.Subscription;
-import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type;
import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferException;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
@@ -67,6 +67,7 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
protected final PersistentTopic topic;
protected final Executor executor;
+ @Getter
protected final String name;
private Optional<DispatchRateLimiter> dispatchRateLimiter = Optional.empty();
@@ -439,54 +440,14 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
// active-cursor reads message from cache rather from bookkeeper (2) if topic has reached message-rate
// threshold: then schedule the read after MESSAGE_RATE_BACKOFF_MS
if (serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) {
- if (topic.getBrokerDispatchRateLimiter().isPresent()) {
- DispatchRateLimiter brokerRateLimiter = topic.getBrokerDispatchRateLimiter().get();
+ if (hasAnyDispatchRateLimiter()) {
Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(brokerRateLimiter, messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
- if (messagesToRead == 0 || bytesToRead == 0) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] message-read exceeded broker message-rate {}/{}, schedule after a {}", name,
- brokerRateLimiter.getDispatchRateOnMsg(), brokerRateLimiter.getDispatchRateOnByte(),
- MESSAGE_RATE_BACKOFF_MS);
- }
- reScheduleRead();
- return Pair.of(-1, -1L);
- }
- }
-
- if (topic.getDispatchRateLimiter().isPresent()) {
- DispatchRateLimiter topicRateLimiter = topic.getDispatchRateLimiter().get();
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(topicRateLimiter, messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
- if (messagesToRead == 0 || bytesToRead == 0) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] message-read exceeded topic message-rate {}/{}, schedule after a {}", name,
- topicRateLimiter.getDispatchRateOnMsg(), topicRateLimiter.getDispatchRateOnByte(),
- MESSAGE_RATE_BACKOFF_MS);
- }
- reScheduleRead();
- return Pair.of(-1, -1L);
- }
- }
-
- if (dispatchRateLimiter.isPresent()) {
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(dispatchRateLimiter.get(), messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
- if (messagesToRead == 0 || bytesToRead == 0) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] message-read exceeded subscription message-rate {}/{}, schedule after a {}",
- name, dispatchRateLimiter.get().getDispatchRateOnMsg(),
- dispatchRateLimiter.get().getDispatchRateOnByte(),
- MESSAGE_RATE_BACKOFF_MS);
- }
- reScheduleRead();
- return Pair.of(-1, -1L);
+ applyRateLimitsToMessagesAndBytesToRead(messagesToRead, bytesToRead);
+ if (calculateToRead.getLeft() == -1 && calculateToRead.getRight() == -1) {
+ return calculateToRead;
+ } else {
+ messagesToRead = calculateToRead.getLeft();
+ bytesToRead = calculateToRead.getRight();
}
}
}
@@ -601,7 +562,8 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
if (!dispatchRateLimiter.isPresent() && DispatchRateLimiter.isDispatchRateEnabled(
topic.getSubscriptionDispatchRate(getSubscriptionName()))) {
this.dispatchRateLimiter =
- Optional.of(new DispatchRateLimiter(topic, getSubscriptionName(), Type.SUBSCRIPTION));
+ Optional.of(topic.getBrokerService().getDispatchRateLimiterFactory()
+ .createSubscriptionDispatchRateLimiter(topic, getSubscriptionName()));
return true;
}
return false;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
index cc531afc6dd..881b1b804e8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -55,7 +55,6 @@ import org.apache.pulsar.broker.service.AbstractReplicator;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.MessageExpirer;
import org.apache.pulsar.broker.service.Replicator;
-import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;
@@ -696,7 +695,8 @@ public abstract class PersistentReplicator extends AbstractReplicator
if (!dispatchRateLimiter.isPresent()
&& DispatchRateLimiter.isDispatchRateEnabled(topic.getReplicatorDispatchRate())) {
this.dispatchRateLimiter = Optional.of(
- new DispatchRateLimiter(topic, Codec.decode(cursor.getName()), Type.REPLICATOR));
+ topic.getBrokerService().getDispatchRateLimiterFactory()
+ .createReplicatorDispatchRateLimiter(topic, Codec.decode(cursor.getName())));
}
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index ed05e47ed38..bfc95b94e96 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -130,7 +130,6 @@ import org.apache.pulsar.broker.service.SubscriptionOption;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.TopicPoliciesService;
import org.apache.pulsar.broker.service.TransportCnx;
-import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type;
import org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage;
import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
import org.apache.pulsar.broker.service.schema.exceptions.NotExistSchemaException;
@@ -494,7 +493,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
// dispatch rate limiter for topic
if (!dispatchRateLimiter.isPresent()
&& DispatchRateLimiter.isDispatchRateEnabled(topicPolicies.getDispatchRate().get())) {
- this.dispatchRateLimiter = Optional.of(new DispatchRateLimiter(this, Type.TOPIC));
+ this.dispatchRateLimiter = Optional.of(
+ getBrokerService().getDispatchRateLimiterFactory().createTopicDispatchRateLimiter(this));
}
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SubscribeRateLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SubscribeRateLimiter.java
index 0f98ab94142..1872a17c965 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SubscribeRateLimiter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SubscribeRateLimiter.java
@@ -118,9 +118,7 @@ public class SubscribeRateLimiter {
if (ratePerConsumer > 0) {
AsyncTokenBucket tokenBucket =
AsyncTokenBucket.builder()
- .consistentAddedTokens(true)
- .consistentConsumedTokens(true)
- .clock(brokerService.getPulsar().getMonotonicSnapshotClock())
+ .clock(brokerService.getPulsar().getMonotonicClock())
.rate(ratePerConsumer).ratePeriodNanos(ratePeriodNanos).build();
this.subscribeRateLimiter.put(consumerIdentifier, tokenBucket);
} else {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/qos/AsyncTokenBucketTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/qos/AsyncTokenBucketTest.java
index 82793f2748d..55ca9940541 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/qos/AsyncTokenBucketTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/qos/AsyncTokenBucketTest.java
@@ -29,14 +29,14 @@ import org.testng.annotations.Test;
public class AsyncTokenBucketTest {
private AtomicLong manualClockSource;
- private MonotonicSnapshotClock clockSource;
+ private MonotonicClock clockSource;
private AsyncTokenBucket asyncTokenBucket;
@BeforeMethod
public void setup() {
manualClockSource = new AtomicLong(TimeUnit.SECONDS.toNanos(100));
- clockSource = requestSnapshot -> manualClockSource.get();
+ clockSource = () -> manualClockSource.get();
}
@@ -51,7 +51,7 @@ public class AsyncTokenBucketTest {
@Test
void shouldAddTokensWithConfiguredRate() {
asyncTokenBucket =
- AsyncTokenBucket.builder().consistentConsumedTokens(true)
+ AsyncTokenBucket.builder()
.capacity(100).rate(10).initialTokens(0).clock(clockSource).build();
incrementSeconds(5);
assertEquals(asyncTokenBucket.getTokens(), 50);
@@ -94,7 +94,6 @@ public class AsyncTokenBucketTest {
void shouldSupportFractionsAndRetainLeftoverWhenUpdatingTokens() {
asyncTokenBucket =
AsyncTokenBucket.builder().capacity(100)
- .resolutionNanos(TimeUnit.MILLISECONDS.toNanos(1))
.rate(10)
.initialTokens(0)
.clock(clockSource)
@@ -115,7 +114,6 @@ public class AsyncTokenBucketTest {
void shouldSupportFractionsAndRetainLeftoverWhenUpdatingTokens2() {
asyncTokenBucket =
AsyncTokenBucket.builder().capacity(100)
- .resolutionNanos(TimeUnit.MILLISECONDS.toNanos(1))
.rate(1)
.initialTokens(0)
.clock(clockSource)
@@ -138,8 +136,6 @@ public class AsyncTokenBucketTest {
void shouldHandleNegativeBalanceWithEventuallyConsistentTokenUpdates() {
asyncTokenBucket =
AsyncTokenBucket.builder()
- // intentionally pick a coarse resolution
- .resolutionNanos(TimeUnit.SECONDS.toNanos(51))
.capacity(100).rate(10).initialTokens(0).clock(clockSource).build();
// assert that the token balance is 0 initially
assertThat(asyncTokenBucket.getTokens()).isEqualTo(0);
@@ -161,8 +157,6 @@ public class AsyncTokenBucketTest {
void shouldNotExceedTokenBucketSizeWithNegativeTokens() {
asyncTokenBucket =
AsyncTokenBucket.builder()
- // intentionally pick a coarse resolution
- .resolutionNanos(TimeUnit.SECONDS.toNanos(51))
.capacity(100).rate(10).initialTokens(0).clock(clockSource).build();
// assert that the token balance is 0 initially
assertThat(asyncTokenBucket.getTokens()).isEqualTo(0);
@@ -184,48 +178,12 @@ public class AsyncTokenBucketTest {
assertThat(asyncTokenBucket.getTokens()).isEqualTo(90);
}
- @Test
- void shouldAccuratelyCalculateTokensWhenTimeIsLaggingBehindInInconsistentUpdates() {
- clockSource = requestSnapshot -> {
- if (requestSnapshot) {
- return manualClockSource.get();
- } else {
- // let the clock lag behind
- return manualClockSource.get() - TimeUnit.SECONDS.toNanos(52);
- }
- };
- incrementSeconds(1);
- asyncTokenBucket =
- AsyncTokenBucket.builder().resolutionNanos(TimeUnit.SECONDS.toNanos(51))
- .capacity(100).rate(10).initialTokens(100).clock(clockSource).build();
- assertThat(asyncTokenBucket.getTokens()).isEqualTo(100);
-
- // consume tokens without exceeding the rate
- for (int i = 0; i < 10000; i++) {
- asyncTokenBucket.consumeTokens(500);
- incrementSeconds(i == 0 ? 40 : 50);
- }
-
- // let 9 seconds pass
- incrementSeconds(9);
-
- // there should be 90 tokens available
- assertThat(asyncTokenBucket.getTokens()).isEqualTo(90);
- }
-
@Test
void shouldHandleEventualConsistency() {
- AtomicLong offset = new AtomicLong(0);
- long resolutionNanos = TimeUnit.MILLISECONDS.toNanos(1);
- DefaultMonotonicSnapshotClock monotonicSnapshotClock =
- new DefaultMonotonicSnapshotClock(resolutionNanos,
- () -> offset.get() + manualClockSource.get());
long initialTokens = 500L;
asyncTokenBucket =
AsyncTokenBucket.builder()
- .consistentConsumedTokens(true)
- .resolutionNanos(resolutionNanos)
- .capacity(100000).rate(1000).initialTokens(initialTokens).clock(monotonicSnapshotClock).build();
+ .capacity(100000).rate(1000).initialTokens(initialTokens).clock(clockSource).build();
for (int i = 0; i < 100000; i++) {
// increment the clock by 1ms, since rate is 1000 tokens/s, this should make 1 token available
incrementMillis(1);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClockTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClockTest.java
deleted file mode 100644
index 0820b439915..00000000000
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClockTest.java
+++ /dev/null
@@ -1,185 +0,0 @@
-/*
- * 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.qos;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import java.time.Duration;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-import lombok.Cleanup;
-import lombok.extern.slf4j.Slf4j;
-import org.assertj.core.data.Offset;
-import org.testng.annotations.DataProvider;
-import org.testng.annotations.Test;
-
-@Slf4j
-public class DefaultMonotonicSnapshotClockTest {
- @DataProvider
- private static Object[] booleanValues() {
- return new Object[]{ true, false };
- }
-
- @Test(dataProvider = "booleanValues")
- void testClockHandlesTimeLeapsBackwards(boolean requestSnapshot) throws InterruptedException {
- long snapshotIntervalMillis = 5;
- AtomicLong clockValue = new AtomicLong(1);
- @Cleanup
- DefaultMonotonicSnapshotClock clock =
- new DefaultMonotonicSnapshotClock(Duration.ofMillis(snapshotIntervalMillis).toNanos(),
- clockValue::get);
-
-
- long previousTick = -1;
- boolean leapDirection = true;
- for (int i = 0; i < 10000; i++) {
- clockValue.addAndGet(TimeUnit.MILLISECONDS.toNanos(1));
- long tick = clock.getTickNanos(requestSnapshot);
- //log.info("i = {}, tick = {}", i, tick);
- if ((i + 1) % 5 == 0) {
- leapDirection = !leapDirection;
- //log.info("Time leap 5 minutes backwards");
- clockValue.addAndGet(-Duration.ofMinutes(5).toNanos());
- }
- if (previousTick != -1) {
- assertThat(tick)
- .describedAs("i = %d, tick = %d, previousTick = %d", i, tick, previousTick)
- .isGreaterThanOrEqualTo(previousTick)
- .isCloseTo(previousTick,
- // then snapshot is requested, the time difference between the two ticks is accurate
- // otherwise allow time difference at most 4 times the snapshot interval since the
- // clock is updated periodically by a background thread
- Offset.offset(TimeUnit.MILLISECONDS.toNanos(
- requestSnapshot ? 1 : 4 * snapshotIntervalMillis)));
- }
- previousTick = tick;
- }
- }
-
- @Test
- void testRequestUpdate() throws InterruptedException {
- @Cleanup
- DefaultMonotonicSnapshotClock clock =
- new DefaultMonotonicSnapshotClock(Duration.ofSeconds(5).toNanos(), System::nanoTime);
- long tick1 = clock.getTickNanos(false);
- long tick2 = clock.getTickNanos(true);
- assertThat(tick2).isGreaterThan(tick1);
- }
-
- @Test
- void testRequestingSnapshotAfterClosed() throws InterruptedException {
- DefaultMonotonicSnapshotClock clock =
- new DefaultMonotonicSnapshotClock(Duration.ofSeconds(5).toNanos(), System::nanoTime);
- clock.close();
- long tick1 = clock.getTickNanos(true);
- Thread.sleep(10);
- long tick2 = clock.getTickNanos(true);
- assertThat(tick2).isGreaterThan(tick1);
- }
-
- @Test
- void testConstructorValidation() {
- assertThatThrownBy(() -> new DefaultMonotonicSnapshotClock(0, System::nanoTime))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessage("snapshotIntervalNanos must be at least 1 millisecond");
- assertThatThrownBy(() -> new DefaultMonotonicSnapshotClock(-1, System::nanoTime))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessage("snapshotIntervalNanos must be at least 1 millisecond");
- assertThatThrownBy(() -> new DefaultMonotonicSnapshotClock(TimeUnit.MILLISECONDS.toNanos(1), null))
- .isInstanceOf(NullPointerException.class)
- .hasMessage("clockSource must not be null");
- }
-
- @Test
- void testFailureHandlingInClockSource() {
- @Cleanup
- DefaultMonotonicSnapshotClock clock =
- new DefaultMonotonicSnapshotClock(Duration.ofSeconds(5).toNanos(), () -> {
- throw new RuntimeException("Test clock failure");
- });
- // the exception should be propagated
- assertThatThrownBy(() -> clock.getTickNanos(true))
- .isInstanceOf(RuntimeException.class)
- .hasMessage("Test clock failure");
- }
-
- @Test
- void testLeapDetectionIndependently() {
- AtomicLong clockValue = new AtomicLong(0);
- AtomicLong tickValue = new AtomicLong(0);
- long expectedTickValue = 0;
- long snapshotIntervalNanos = TimeUnit.MILLISECONDS.toNanos(1);
- DefaultMonotonicSnapshotClock.MonotonicLeapDetectingTickUpdater updater =
- new DefaultMonotonicSnapshotClock.MonotonicLeapDetectingTickUpdater(clockValue::get, tickValue::set,
- snapshotIntervalNanos);
-
- updater.update(true);
-
- // advance the clock
- clockValue.addAndGet(snapshotIntervalNanos);
- expectedTickValue += snapshotIntervalNanos;
- updater.update(true);
- assertThat(tickValue.get()).isEqualTo(expectedTickValue);
-
- // simulate a leap backwards in time
- clockValue.addAndGet(-10 * snapshotIntervalNanos);
- expectedTickValue += snapshotIntervalNanos;
- updater.update(true);
- assertThat(tickValue.get()).isEqualTo(expectedTickValue);
-
- // advance the clock
- clockValue.addAndGet(snapshotIntervalNanos);
- expectedTickValue += snapshotIntervalNanos;
- updater.update(true);
- assertThat(tickValue.get()).isEqualTo(expectedTickValue);
-
- // simulate a leap backwards in time, without waiting a full snapshot interval
- clockValue.addAndGet(-10 * snapshotIntervalNanos);
- updater.update(false);
- assertThat(tickValue.get()).isEqualTo(expectedTickValue);
-
- // advance the clock
- clockValue.addAndGet(snapshotIntervalNanos);
- expectedTickValue += snapshotIntervalNanos;
- updater.update(true);
- assertThat(tickValue.get()).isEqualTo(expectedTickValue);
-
- // simulate a small leap backwards in time which isn't detected, without waiting a full snapshot interval
- clockValue.addAndGet(-1 * snapshotIntervalNanos);
- updater.update(false);
- assertThat(tickValue.get()).isEqualTo(expectedTickValue);
- // clock doesn't advance for one snapshot interval
- clockValue.addAndGet(snapshotIntervalNanos);
- updater.update(false);
- assertThat(tickValue.get()).isEqualTo(expectedTickValue);
- // now the clock should advance again
- clockValue.addAndGet(snapshotIntervalNanos);
- expectedTickValue += snapshotIntervalNanos;
- updater.update(false);
- assertThat(tickValue.get()).isEqualTo(expectedTickValue);
-
- // simulate a leap forward
- clockValue.addAndGet(10 * snapshotIntervalNanos);
- // no special handling for leap forward
- expectedTickValue += 10 * snapshotIntervalNanos;
- updater.update(true);
- assertThat(tickValue.get()).isEqualTo(expectedTickValue);
- }
-}
\ No newline at end of file
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/RGUsageMTAggrWaitForAllMsgsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/RGUsageMTAggrWaitForAllMsgsTest.java
index 8343680f9bf..c12768049c8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/RGUsageMTAggrWaitForAllMsgsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/RGUsageMTAggrWaitForAllMsgsTest.java
@@ -25,7 +25,6 @@ import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.apache.pulsar.broker.resourcegroup.ResourceGroup.BytesAndMessagesCount;
import org.apache.pulsar.broker.resourcegroup.ResourceGroup.ResourceGroupMonitoringClass;
import org.apache.pulsar.broker.resourcegroup.ResourceGroupService.ResourceGroupUsageStatsType;
@@ -62,7 +61,6 @@ public class RGUsageMTAggrWaitForAllMsgsTest extends ProducerConsumerBase {
@BeforeClass(alwaysRun = true)
@Override
protected void setup() throws Exception {
- AsyncTokenBucket.switchToConsistentTokensView();
super.internalSetup();
this.prepareForOps();
@@ -93,7 +91,6 @@ public class RGUsageMTAggrWaitForAllMsgsTest extends ProducerConsumerBase {
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
- AsyncTokenBucket.resetToDefaultEventualConsistentTokensView();
}
@Test
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractBaseDispatcherTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractBaseDispatcherTest.java
index 6866e097313..23f7278b254 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractBaseDispatcherTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractBaseDispatcherTest.java
@@ -249,6 +249,11 @@ public class AbstractBaseDispatcherTest {
}
+ @Override
+ public String getName() {
+ return "AbstractBaseDispatcherTestHelper for subscription" + subscription.getName();
+ }
+
@Override
public CompletableFuture<Void> addConsumer(Consumer consumer) {
return CompletableFuture.completedFuture(null);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractTopicTest.java
index 337717ed97b..1e15bcf12c0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractTopicTest.java
@@ -48,7 +48,7 @@ public class AbstractTopicTest {
doReturn(pulsarService).when(brokerService).getPulsar();
when(pulsarService.getConfiguration()).thenReturn(serviceConfiguration);
when(brokerService.getBacklogQuotaManager()).thenReturn(backlogQuotaManager);
- doReturn(AsyncTokenBucket.DEFAULT_SNAPSHOT_CLOCK).when(pulsarService).getMonotonicSnapshotClock();
+ doReturn(AsyncTokenBucket.DEFAULT_SNAPSHOT_CLOCK).when(pulsarService).getMonotonicClock();
topic = mock(AbstractTopic.class, withSettings()
.useConstructor("topic", brokerService)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/DispatchRateLimiterOverconsumingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/DispatchRateLimiterOverconsumingTest.java
new file mode 100644
index 00000000000..d45eb24d848
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/DispatchRateLimiterOverconsumingTest.java
@@ -0,0 +1,284 @@
+/*
+ * 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.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.SoftAssertions.assertSoftly;
+import static org.testng.Assert.assertEquals;
+import java.lang.reflect.Method;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
+import java.util.stream.IntStream;
+import lombok.Cleanup;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryAsyncTokenBucket;
+import org.apache.pulsar.broker.service.persistent.DispatchRateLimiterFactoryClassic;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageListener;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.assertj.core.data.Percentage;
+import org.awaitility.Awaitility;
+import org.testng.ITest;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Factory;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class DispatchRateLimiterOverconsumingTest extends BrokerTestBase implements ITest {
+ public enum DispatchRateLimiterImplType {
+ PIP322(DispatchRateLimiterFactoryAsyncTokenBucket.class.getName()),
+ Classic(DispatchRateLimiterFactoryClassic.class.getName());
+
+ private final String factoryClassName;
+
+ DispatchRateLimiterImplType(String implementationClassName) {
+ this.factoryClassName = implementationClassName;
+ }
+
+ public String getFactoryClassName() {
+ return factoryClassName;
+ }
+
+ public static Object[] generateTestInstances(
+ Function<DispatchRateLimiterImplType, Object> testInstanceFactory) {
+ return Arrays.stream(values()).map(testInstanceFactory).toArray();
+ }
+ }
+
+ // Comment out the next line (Factory annotation) to run tests manually in IntelliJ, one-by-one
+ @Factory
+ public static Object[] createTestInstances() {
+ return DispatchRateLimiterImplType.generateTestInstances(DispatchRateLimiterOverconsumingTest::new);
+ }
+
+ public DispatchRateLimiterOverconsumingTest() {
+ // set the default implementation type for manual running in IntelliJ
+ this(DispatchRateLimiterImplType.Classic);
+ }
+
+ public DispatchRateLimiterOverconsumingTest(DispatchRateLimiterImplType implType) {
+ this.implType = implType;
+ }
+
+ private DispatchRateLimiterImplType implType;
+ private String testName;
+
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+ super.baseSetup();
+ }
+
+ @Override
+ public String getTestName() {
+ return testName;
+ }
+
+ @BeforeMethod
+ public void applyTestName(Method method) {
+ testName = method.getName() + " with " + implType.name() + " rate limiter implementation";
+ }
+
+ @Override
+ protected void doInitConf() throws Exception {
+ super.doInitConf();
+ // configure dispatch rate limiter factory class name
+ conf.setDispatchRateLimiterFactoryClassName(implType.getFactoryClassName());
+ // simplify testing by enabling throttling on non-backlog consumers
+ conf.setDispatchThrottlingOnNonBacklogConsumerEnabled(true);
+ // set the dispatch max read batch size to 1 to stress the rate limiting behavior more effectively
+ conf.setDispatcherMaxReadBatchSize(1);
+ // avoid dispatching messages in a separate thread to simplify testing and reduce variance
+ conf.setDispatcherDispatchMessagesInSubscriptionThread(false);
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ this.testName = null;
+ super.internalCleanup();
+ }
+
+ /**
+ * This test verifies the broker dispatch rate limiting behavior with multiple concurrent consumers.
+ * Reproduces issue "with a huge spike in a traffic consume is stuck for a long time" mentioned in
+ * issue https://github.com/apache/pulsar/issues/24001 and prevents future regressions.
+ */
+ @Test
+ public void testOverconsumingTokensWithBrokerDispatchRateLimiter() throws Exception {
+ int rateInMsg = 50;
+ int durationSeconds = 5;
+ int numberOfConsumers = 20;
+ int numberOfMessages = rateInMsg * durationSeconds;
+
+ // configure dispatch throttling rate
+ BrokerService brokerService = pulsar.getBrokerService();
+ admin.brokers().updateDynamicConfiguration("dispatchThrottlingRateInMsg", String.valueOf(rateInMsg));
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(brokerService.getBrokerDispatchRateLimiter()
+ .getAvailableDispatchRateLimitOnMsg(), rateInMsg));
+ assertEquals(brokerService.getBrokerDispatchRateLimiter().getDispatchRateOnByte(), -1L);
+
+ final String topicName = "persistent://" + newTopicName();
+
+ // state for calculating message rate
+ AtomicLong startTimeNanos = new AtomicLong();
+ AtomicLong lastReceivedMessageTimeNanos = new AtomicLong();
+ AtomicInteger totalMessagesReceived = new AtomicInteger();
+ AtomicInteger currentSecondMessagesCount = new AtomicInteger();
+ AtomicInteger lastCalculatedSecond = new AtomicInteger(0);
+ List<Integer> collectedRatesForEachSecond = Collections.synchronizedList(new ArrayList<>());
+
+ // track actual consuming rate of messages per second
+ Runnable rateTracker = () -> {
+ long startTime = startTimeNanos.get();
+ if (startTime == 0) {
+ return;
+ }
+ long durationNanos = System.nanoTime() - startTime;
+ int elapsedFullSeconds = (int) (durationNanos / 1e9);
+ if (elapsedFullSeconds > 0 && lastCalculatedSecond.compareAndSet(elapsedFullSeconds - 1,
+ elapsedFullSeconds)) {
+ int messagesCountForPreviousSecond = currentSecondMessagesCount.getAndSet(0);
+ log.info("Rate for second {}: {} msg/s {}", elapsedFullSeconds, messagesCountForPreviousSecond, TimeUnit.NANOSECONDS.toMillis(durationNanos));
+ collectedRatesForEachSecond.add(messagesCountForPreviousSecond);
+ }
+ };
+ @Cleanup("shutdownNow")
+ ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+ executor.scheduleAtFixedRate(rateTracker, 0, 500, TimeUnit.MILLISECONDS);
+
+ // message listener implementation used for all consumers
+ MessageListener<Integer> messageListener = new MessageListener<>() {
+ @Override
+ public void received(Consumer<Integer> consumer, Message<Integer> msg) {
+ lastReceivedMessageTimeNanos.set(System.nanoTime());
+ currentSecondMessagesCount.incrementAndGet();
+ totalMessagesReceived.incrementAndGet();
+ consumer.acknowledgeAsync(msg);
+ }
+ };
+
+ // create consumers using a shared subscription called "sub"
+ List<Consumer<Integer>> consumerList = IntStream.range(0, numberOfConsumers)
+ .mapToObj(i -> {
+ try {
+ return pulsarClient.newConsumer(Schema.INT32)
+ .topic(topicName)
+ .consumerName("consumer-" + (i + 1))
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName("sub")
+ .receiverQueueSize(10)
+ .messageListener(messageListener)
+ // start paused so that there's a backlog when consumers are resumed
+ .startPaused(true)
+ .subscribe();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }).toList();
+ // handle consumer cleanup when the test completes
+ @Cleanup
+ AutoCloseable consumerCloser = () -> {
+ consumerList.forEach(c -> {
+ try {
+ c.close();
+ } catch (Exception e) {
+ // ignore
+ }
+ });
+ };
+
+ @Cleanup
+ Producer<Integer> producer =
+ pulsarClient.newProducer(Schema.INT32).enableBatching(false).topic(topicName).create();
+ // send messages
+ IntStream.range(0, numberOfMessages).forEach(i -> {
+ try {
+ int messageNumber = i + 1;
+ producer.sendAsync(messageNumber).exceptionally(e -> {
+ log.error("Failed to send message #{}", messageNumber, e);
+ return null;
+ });
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ log.info("Waiting for messages to be sent");
+ // wait until the messages are sent
+ producer.flush();
+
+ // resume the consumers
+ log.info("Resuming consumers");
+ startTimeNanos.set(System.nanoTime());
+ consumerList.forEach(Consumer::resume);
+
+ // wait for results
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(durationSeconds * 2))
+ .pollInterval(Duration.ofMillis(100))
+ .untilAsserted(() -> assertThat(totalMessagesReceived).hasValue(numberOfMessages));
+ List<Integer> collectedRatesSnapshot = new ArrayList<>(collectedRatesForEachSecond);
+ int messagesCountForPreviousSecond = currentSecondMessagesCount.getAndSet(0);
+ if (messagesCountForPreviousSecond > 0) {
+ collectedRatesSnapshot.add(messagesCountForPreviousSecond);
+ }
+ log.info("[{}] Collected rates for each second: {}", implType, collectedRatesSnapshot);
+ long avgMsgRate =
+ totalMessagesReceived.get() / TimeUnit.NANOSECONDS.toSeconds(
+ lastReceivedMessageTimeNanos.get() - startTimeNanos.get());
+ log.info("[{}] Average rate during the test run: {} msg/s", implType, avgMsgRate);
+
+ assertSoftly(softly -> {
+ // check the rate during the test run
+ softly.assertThat(avgMsgRate).describedAs("average rate during the test run")
+ // allow rate in 40% range
+ .isCloseTo(rateInMsg, Percentage.withPercentage(40));
+
+ // check that rates were collected
+ softly.assertThat(collectedRatesSnapshot).describedAs("actual rates for each second").size()
+ .isGreaterThanOrEqualTo(durationSeconds - 1).returnToIterable().satisfies(rates -> {
+ for (int i = 1; i < rates.size() - 1; i++) {
+ int rateAvg = (rates.get(i - 1) + rates.get(i)) / 2;
+ softly.assertThat(rateAvg).describedAs("Average of second %d and %d", i, i + 1)
+ // TODO: relax the rate check by calculating an average of 2 subsequent
+ // seconds and allowing 55% range
+ // This is due to the fact that the first second rate is usually 2x the
+ // configured rate.
+ // This problem can be handled separately.
+ .isCloseTo(rateInMsg, Percentage.withPercentage(55));
+ }
+ });
+ });
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterOverconsumingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterOverconsumingTest.java
new file mode 100644
index 00000000000..aad1dafed07
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterOverconsumingTest.java
@@ -0,0 +1,266 @@
+/*
+ * 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.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.SoftAssertions.assertSoftly;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.IntStream;
+import lombok.Cleanup;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageListener;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.assertj.core.data.Percentage;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class PublishRateLimiterOverconsumingTest extends BrokerTestBase {
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+ super.baseSetup();
+ }
+
+ @Override
+ protected void doInitConf() throws Exception {
+ super.doInitConf();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ /**
+ * This test verifies the broker publish rate limiting behavior with multiple concurrent publishers.
+ * This reproduces the issue https://github.com/apache/pulsar/issues/23920 and prevents future regressions.
+ */
+ @Test
+ public void testOverconsumingTokensWithBrokerPublishRateLimiter() throws Exception {
+ int rateInMsg = 500;
+ int durationSeconds = 5;
+ int numberOfConsumers = 5;
+ int numberOfProducersWithIndependentClients = 5;
+ int numberOfMessagesForEachProducer = (rateInMsg * (durationSeconds + 1)) / 5;
+
+ // configure dispatch throttling rate
+ BrokerService brokerService = pulsar.getBrokerService();
+ admin.brokers().updateDynamicConfiguration("brokerPublisherThrottlingMaxMessageRate", String.valueOf(rateInMsg));
+ Awaitility.await().untilAsserted(() -> {
+ PublishRateLimiterImpl publishRateLimiter =
+ (PublishRateLimiterImpl) brokerService.getBrokerPublishRateLimiter();
+ assertEquals(publishRateLimiter.getTokenBucketOnMessage().getRate(), rateInMsg);
+ assertNull(publishRateLimiter.getTokenBucketOnByte());
+ });
+
+ final String topicName = "persistent://" + newTopicName();
+
+ // state for calculating message rate
+ AtomicLong startTimeNanos = new AtomicLong();
+ AtomicLong lastReceivedMessageTimeNanos = new AtomicLong();
+ AtomicInteger totalMessagesReceived = new AtomicInteger();
+ AtomicInteger currentSecondMessagesCount = new AtomicInteger();
+ AtomicInteger lastCalculatedSecond = new AtomicInteger(0);
+ List<Integer> collectedRatesForEachSecond = Collections.synchronizedList(new ArrayList<>());
+
+ // track actual consuming rate of messages per second
+ Runnable rateTracker = () -> {
+ long startTime = startTimeNanos.get();
+ if (startTime == 0) {
+ startTimeNanos.compareAndSet(0, System.nanoTime());
+ startTime = startTimeNanos.get();
+ }
+ long durationNanos = System.nanoTime() - startTime;
+ int elapsedFullSeconds = (int) (durationNanos / 1e9);
+ if (elapsedFullSeconds > 0 && lastCalculatedSecond.compareAndSet(elapsedFullSeconds - 1,
+ elapsedFullSeconds)) {
+ int messagesCountForPreviousSecond = currentSecondMessagesCount.getAndSet(0);
+ log.info("Rate for second {}: {} msg/s {}", elapsedFullSeconds, messagesCountForPreviousSecond, TimeUnit.NANOSECONDS.toMillis(durationNanos));
+ collectedRatesForEachSecond.add(messagesCountForPreviousSecond);
+ }
+ };
+ @Cleanup("shutdownNow")
+ ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+ ScheduledFuture<?> scheduledFuture = executor.scheduleAtFixedRate(rateTracker, 0, 500, TimeUnit.MILLISECONDS);
+
+ // message listener implementation used for all consumers
+ MessageListener<Integer> messageListener = new MessageListener<>() {
+ @Override
+ public void received(Consumer<Integer> consumer, Message<Integer> msg) {
+ lastReceivedMessageTimeNanos.set(System.nanoTime());
+ currentSecondMessagesCount.incrementAndGet();
+ totalMessagesReceived.incrementAndGet();
+ consumer.acknowledgeAsync(msg);
+ }
+ };
+
+ // create consumers using a shared subscription called "sub"
+ List<Consumer<Integer>> consumerList = IntStream.range(0, numberOfConsumers)
+ .mapToObj(i -> {
+ try {
+ return pulsarClient.newConsumer(Schema.INT32)
+ .topic(topicName)
+ .consumerName("consumer-" + (i + 1))
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName("sub")
+ .messageListener(messageListener)
+ .subscribe();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }).toList();
+ // handle consumer cleanup when the test completes
+ @Cleanup
+ AutoCloseable consumerCloser = () -> {
+ consumerList.forEach(c -> {
+ try {
+ c.close();
+ } catch (Exception e) {
+ // ignore
+ }
+ });
+ };
+
+ // create independent clients for producers so that they don't get blocked by throttling
+ List<PulsarClient> producerClients = IntStream.range(0, numberOfProducersWithIndependentClients)
+ .mapToObj(i -> {
+ try {
+ return PulsarClient.builder()
+ .serviceUrl(pulsar.getBrokerServiceUrl())
+ .ioThreads(1)
+ .statsInterval(0, TimeUnit.SECONDS)
+ .connectionsPerBroker(1)
+ .build();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }).toList();
+ @Cleanup
+ AutoCloseable producerClientsCloser = () -> {
+ producerClients.forEach(c -> {
+ try {
+ c.close();
+ } catch (Exception e) {
+ // ignore
+ }
+ });
+ };
+
+ List<Producer<Integer>> producers = IntStream.range(0, numberOfProducersWithIndependentClients)
+ .mapToObj(i -> {
+ try {
+ return producerClients.get(i)
+ .newProducer(Schema.INT32).enableBatching(true)
+ .producerName("producer-" + (i + 1))
+ .batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS)
+ .batchingMaxMessages(numberOfMessagesForEachProducer / 100)
+ .topic(topicName).create();
+ } catch (PulsarClientException e) {
+ throw new RuntimeException(e);
+ }
+ }).toList();
+
+ // send messages
+ producers.forEach(producer -> {
+ IntStream.range(0, numberOfMessagesForEachProducer).forEach(i -> {
+ try {
+ int messageNumber = i + 1;
+ producer.sendAsync(messageNumber).exceptionally(e -> {
+ log.error("Failed to send message #{}", messageNumber, e);
+ return null;
+ });
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ producer.flushAsync().whenComplete((r, e) -> {
+ if (e != null) {
+ log.error("Failed to flush producer", e);
+ } else {
+ log.info("Producer {} flushed", producer.getProducerName());
+ }
+ });
+ });
+
+ @Cleanup
+ AutoCloseable producersClose = () -> {
+ producers.forEach(p -> {
+ try {
+ p.close();
+ } catch (Exception e) {
+ // ignore
+ }
+ });
+ };
+
+ // wait for results
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(durationSeconds * 2))
+ .pollInterval(Duration.ofMillis(100))
+ .untilAsserted(
+ () -> assertThat(collectedRatesForEachSecond).hasSizeGreaterThanOrEqualTo(durationSeconds));
+ List<Integer> collectedRatesSnapshot = new ArrayList<>(collectedRatesForEachSecond);
+ log.info("Collected rates for each second: {}", collectedRatesSnapshot);
+ long avgMsgRate =
+ totalMessagesReceived.get() / TimeUnit.NANOSECONDS.toSeconds(
+ lastReceivedMessageTimeNanos.get() - startTimeNanos.get());
+ log.info("Average rate during the test run: {} msg/s", avgMsgRate);
+
+ assertSoftly(softly -> {
+ // check the rate during the test run
+ softly.assertThat(avgMsgRate).describedAs("average rate during the test run")
+ // allow rate in 40% range
+ .isCloseTo(rateInMsg, Percentage.withPercentage(40));
+
+ // check that rates were collected
+ // skip the first element as it might contain messages for first 2 seconds
+ softly.assertThat(collectedRatesSnapshot.subList(1, collectedRatesSnapshot.size() - 1))
+ .describedAs("actual rates for each second")
+ .allSatisfy(rates -> {
+ assertThat(rates).describedAs("actual rate for each second")
+ .isCloseTo(rateInMsg, Percentage.withPercentage(50));
+ });
+ });
+ scheduledFuture.cancel(true);
+ producersClose.close();
+ consumerCloser.close();
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java
index 5c149d4e1e7..9c3c6779b02 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java
@@ -56,7 +56,7 @@ public class PublishRateLimiterTest {
policies.publishMaxMessageRate = new HashMap<>();
policies.publishMaxMessageRate.put(CLUSTER_NAME, publishRate);
manualClockSource = new AtomicLong(TimeUnit.SECONDS.toNanos(100));
- publishRateLimiter = new PublishRateLimiterImpl(requestSnapshot -> manualClockSource.get());
+ publishRateLimiter = new PublishRateLimiterImpl(() -> manualClockSource.get());
publishRateLimiter.update(policies, CLUSTER_NAME);
producer = mock(Producer.class);
throttleCount.set(0);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java
index 2e0dd0a90e8..41a692df18b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorRateLimiterTest.java
@@ -36,7 +36,6 @@ import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.common.policies.data.DispatchRate;
-import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.awaitility.Awaitility;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,7 +62,6 @@ public class ReplicatorRateLimiterTest extends ReplicatorTestBase {
@Override
@BeforeClass(timeOut = 300000)
public void setup() throws Exception {
- AsyncTokenBucket.switchToConsistentTokensView();
super.setup();
}
@@ -71,7 +69,6 @@ public class ReplicatorRateLimiterTest extends ReplicatorTestBase {
@AfterClass(alwaysRun = true, timeOut = 300000)
public void cleanup() throws Exception {
super.cleanup();
- AsyncTokenBucket.resetToDefaultEventualConsistentTokensView();
}
enum DispatchRateType {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPublishRateThrottleTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPublishRateThrottleTest.java
index 554bc754c5c..e251e244a83 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPublishRateThrottleTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPublishRateThrottleTest.java
@@ -24,7 +24,6 @@ import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.common.policies.data.PublishRate;
-import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -37,14 +36,13 @@ public class TopicPublishRateThrottleTest extends BrokerTestBase{
@BeforeMethod(alwaysRun = true)
@Override
protected void setup() throws Exception {
- AsyncTokenBucket.switchToConsistentTokensView();
+ // no-op, each test will call baseSetup
}
@AfterMethod(alwaysRun = true)
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
- AsyncTokenBucket.resetToDefaultEventualConsistentTokensView();
}
@Test
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java
index 5b1c78574b4..342e44b962c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java
@@ -268,7 +268,7 @@ public class MessageDuplicationTest extends BrokerTestBase {
doReturn(pulsarService).when(brokerService).pulsar();
doReturn(pulsarService).when(brokerService).getPulsar();
doReturn(new BacklogQuotaManager(pulsarService)).when(brokerService).getBacklogQuotaManager();
- doReturn(AsyncTokenBucket.DEFAULT_SNAPSHOT_CLOCK).when(pulsarService).getMonotonicSnapshotClock();
+ doReturn(AsyncTokenBucket.DEFAULT_SNAPSHOT_CLOCK).when(pulsarService).getMonotonicClock();
PersistentTopic persistentTopic = spyWithClassAndConstructorArgs(PersistentTopic.class, "topic-1", brokerService, managedLedger, messageDeduplication);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AbstractMessageDispatchThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AbstractMessageDispatchThrottlingTest.java
index 31c628b2bc4..666a100378c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AbstractMessageDispatchThrottlingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AbstractMessageDispatchThrottlingTest.java
@@ -24,7 +24,6 @@ import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
-import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.apache.pulsar.broker.service.BrokerService;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
@@ -43,7 +42,6 @@ public abstract class AbstractMessageDispatchThrottlingTest extends ProducerCons
@BeforeClass(alwaysRun = true)
@Override
protected void setup() throws Exception {
- AsyncTokenBucket.switchToConsistentTokensView();
this.conf.setClusterName("test");
internalSetup();
producerBaseSetup();
@@ -53,7 +51,6 @@ public abstract class AbstractMessageDispatchThrottlingTest extends ProducerCons
@Override
protected void cleanup() throws Exception {
internalCleanup();
- AsyncTokenBucket.resetToDefaultEventualConsistentTokensView();
}
@AfterMethod(alwaysRun = true)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessagePublishThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessagePublishThrottlingTest.java
index a848d68f37f..2a2c26ce4b0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessagePublishThrottlingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessagePublishThrottlingTest.java
@@ -34,7 +34,6 @@ import org.apache.pulsar.broker.service.Producer;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.common.policies.data.PublishRate;
-import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.AfterMethod;
@@ -48,7 +47,6 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
@BeforeMethod
@Override
protected void setup() throws Exception {
- AsyncTokenBucket.switchToConsistentTokensView();
this.conf.setClusterName("test");
this.conf.setTopicPublisherThrottlingTickTimeMillis(1);
this.conf.setBrokerPublisherThrottlingTickTimeMillis(1);
@@ -60,7 +58,6 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
- AsyncTokenBucket.resetToDefaultEventualConsistentTokensView();
}
/**
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimitFunction.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimitFunction.java
new file mode 100644
index 00000000000..0a4b62e9500
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimitFunction.java
@@ -0,0 +1,26 @@
+/*
+ * 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.common.util;
+
+/**
+ * Function use when rate limiter renew permit.
+ * */
+public interface RateLimitFunction {
+ void apply();
+}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java
new file mode 100644
index 00000000000..17199664f12
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java
@@ -0,0 +1,291 @@
+/*
+ * 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.common.util;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
+import com.google.common.base.MoreObjects;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import lombok.Builder;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * A Rate Limiter that distributes permits at a configurable rate. Each {@link #acquire()} blocks if necessary until a
+ * permit is available, and then takes it. Each {@link #tryAcquire()} tries to acquire permits from available permits,
+ * it returns true if it succeed else returns false. Rate limiter release configured permits at every configured rate
+ * time, so, on next ticket new fresh permits will be available.
+ *
+ * <p>For example: if RateLimiter is configured to release 10 permits at every 1 second then RateLimiter will allow to
+ * acquire 10 permits at any time with in that 1 second.
+ *
+ * <p>Comparison with other RateLimiter such as {@link com.google.common.util.concurrent.RateLimiter}
+ * <ul>
+ * <li><b>Per second rate-limiting:</b> Per second rate-limiting not satisfied by Guava-RateLimiter</li>
+ * <li><b>Guava RateLimiter:</b> For X permits: it releases X/1000 permits every msec. therefore,
+ * for permits=2/sec => it release 1st permit on first 500msec and 2nd permit on next 500ms. therefore,
+ * if 2 request comes with in 500msec duration then 2nd request fails to acquire permit
+ * though we have configured 2 permits/second.</li>
+ * <li><b>RateLimiter:</b> it releases X permits every second. so, in above usecase:
+ * if 2 requests comes at the same time then both will acquire the permit.</li>
+ * <li><b>Faster: </b>RateLimiter is light-weight and faster than Guava-RateLimiter</li>
+ * </ul>
+ */
+@Slf4j
+public class RateLimiter implements AutoCloseable{
+ private final ScheduledExecutorService executorService;
+ private long rateTime;
+ private TimeUnit timeUnit;
+ private final boolean externalExecutor;
+ private ScheduledFuture<?> renewTask;
+ private volatile long permits;
+ private volatile long acquiredPermits;
+ private boolean isClosed;
+ // permitUpdate helps to update permit-rate at runtime
+ private Supplier<Long> permitUpdater;
+ private RateLimitFunction rateLimitFunction;
+ private boolean isDispatchOrPrecisePublishRateLimiter;
+
+ @Builder
+ RateLimiter(final ScheduledExecutorService scheduledExecutorService, final long permits, final long rateTime,
+ final TimeUnit timeUnit, Supplier<Long> permitUpdater, boolean isDispatchOrPrecisePublishRateLimiter,
+ RateLimitFunction rateLimitFunction) {
+ checkArgument(permits > 0, "rate must be > 0");
+ checkArgument(rateTime > 0, "Renew permit time must be > 0");
+
+ this.rateTime = rateTime;
+ this.timeUnit = timeUnit;
+ this.permits = permits;
+ this.permitUpdater = permitUpdater;
+ this.isDispatchOrPrecisePublishRateLimiter = isDispatchOrPrecisePublishRateLimiter;
+
+ if (scheduledExecutorService != null) {
+ this.executorService = scheduledExecutorService;
+ this.externalExecutor = true;
+ } else {
+ final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1,
+ new DefaultThreadFactory("pulsar-rate-limiter"));
+ executor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
+ executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
+ this.executorService = executor;
+ this.externalExecutor = false;
+ }
+
+ this.rateLimitFunction = rateLimitFunction;
+
+ }
+
+ // default values for Lombok generated builder class
+ public static class RateLimiterBuilder {
+ private long rateTime = 1;
+ private TimeUnit timeUnit = TimeUnit.SECONDS;
+ }
+
+ @Override
+ public synchronized void close() {
+ if (!isClosed) {
+ if (!externalExecutor) {
+ executorService.shutdownNow();
+ }
+ if (renewTask != null) {
+ renewTask.cancel(false);
+ }
+ isClosed = true;
+ // If there is a ratelimit function registered, invoke it to unblock.
+ if (rateLimitFunction != null) {
+ rateLimitFunction.apply();
+ }
+ }
+ }
+
+ public synchronized boolean isClosed() {
+ return isClosed;
+ }
+
+ /**
+ * Acquires the given number of permits from this {@code RateLimiter}, blocking until the request be granted.
+ *
+ * <p>This method is equivalent to {@code acquire(1)}.
+ */
+ public synchronized void acquire() throws InterruptedException {
+ acquire(1);
+ }
+
+ /**
+ * Acquires the given number of permits from this {@code RateLimiter}, blocking until the request be granted.
+ *
+ * @param acquirePermit
+ * the number of permits to acquire
+ */
+ public synchronized void acquire(long acquirePermit) throws InterruptedException {
+ checkArgument(!isClosed(), "Rate limiter is already shutdown");
+ checkArgument(acquirePermit <= this.permits,
+ "acquiring permits must be less or equal than initialized rate =" + this.permits);
+
+ // lazy init and start task only once application start using it
+ if (renewTask == null) {
+ renewTask = createTask();
+ }
+
+ boolean canAcquire = false;
+ do {
+ canAcquire = acquirePermit < 0 || acquiredPermits < this.permits;
+ if (!canAcquire) {
+ wait();
+ } else {
+ acquiredPermits += acquirePermit;
+ }
+ } while (!canAcquire);
+ }
+
+ /**
+ * Acquires permits from this {@link RateLimiter} if it can be acquired immediately without delay.
+ *
+ * <p>This method is equivalent to {@code tryAcquire(1)}.
+ *
+ * @return {@code true} if the permits were acquired, {@code false} otherwise
+ */
+ public synchronized boolean tryAcquire() {
+ return tryAcquire(1);
+ }
+
+ /**
+ * Acquires permits from this {@link RateLimiter} if it can be acquired immediately without delay.
+ *
+ * @param acquirePermit
+ * the number of permits to acquire
+ * @return {@code true} if the permits were acquired, {@code false} otherwise
+ */
+ public synchronized boolean tryAcquire(long acquirePermit) {
+ if (isClosed()) {
+ log.info("The current rate limiter is already shutdown, acquire permits directly.");
+ return true;
+ }
+ // lazy init and start task only once application start using it
+ if (renewTask == null) {
+ renewTask = createTask();
+ }
+
+ boolean canAcquire = acquirePermit < 0 || acquiredPermits < this.permits;
+ if (isDispatchOrPrecisePublishRateLimiter) {
+ // for dispatch rate limiter just add acquirePermit
+ acquiredPermits += acquirePermit;
+
+ // we want to back-pressure from the current state of the rateLimiter therefore we should check if there
+ // are any available premits again
+ canAcquire = acquirePermit < 0 || acquiredPermits < this.permits;
+ } else {
+ // acquired-permits can't be larger than the rate
+ if (acquirePermit + acquiredPermits > this.permits) {
+ return false;
+ }
+
+ if (canAcquire) {
+ acquiredPermits += acquirePermit;
+ }
+ }
+
+ return canAcquire;
+ }
+
+ /**
+ * Return available permits for this {@link RateLimiter}.
+ *
+ * @return returns 0 if permits is not available
+ */
+ public long getAvailablePermits() {
+ return Math.max(0, this.permits - this.acquiredPermits);
+ }
+
+ /**
+ * Resets new rate by configuring new value for permits per configured rate-period.
+ *
+ * @param permits
+ */
+ public synchronized void setRate(long permits) {
+ this.permits = permits;
+ }
+
+ /**
+ * Resets new rate with new permits and rate-time.
+ *
+ * @param permits
+ * @param rateTime
+ * @param timeUnit
+ * @param permitUpdaterByte
+ */
+ public synchronized void setRate(long permits, long rateTime, TimeUnit timeUnit, Supplier<Long> permitUpdaterByte) {
+ if (renewTask != null) {
+ renewTask.cancel(false);
+ }
+ this.permits = permits;
+ this.rateTime = rateTime;
+ this.timeUnit = timeUnit;
+ this.permitUpdater = permitUpdaterByte;
+ this.renewTask = createTask();
+ }
+
+ /**
+ * Returns configured permit rate per pre-configured rate-period.
+ *
+ * @return rate
+ */
+ public synchronized long getRate() {
+ return this.permits;
+ }
+
+ public synchronized long getRateTime() {
+ return this.rateTime;
+ }
+
+ public synchronized TimeUnit getRateTimeUnit() {
+ return this.timeUnit;
+ }
+
+ protected ScheduledFuture<?> createTask() {
+ return executorService.scheduleAtFixedRate(catchingAndLoggingThrowables(this::renew), this.rateTime,
+ this.rateTime, this.timeUnit);
+ }
+
+ synchronized void renew() {
+ acquiredPermits = isDispatchOrPrecisePublishRateLimiter ? Math.max(0, acquiredPermits - permits) : 0;
+ if (permitUpdater != null) {
+ long newPermitRate = permitUpdater.get();
+ if (newPermitRate > 0) {
+ setRate(newPermitRate);
+ }
+ }
+ // release the back-pressure by applying the rateLimitFunction only when there are available permits
+ if (rateLimitFunction != null && this.getAvailablePermits() > 0) {
+ rateLimitFunction.apply();
+ }
+ notifyAll();
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this).add("rateTime", rateTime).add("permits", permits)
+ .add("acquiredPermits", acquiredPermits).toString();
+ }
+
+}
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java
new file mode 100644
index 00000000000..5cbd0245565
--- /dev/null
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java
@@ -0,0 +1,245 @@
+/*
+ * 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.common.util;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import org.testng.annotations.Test;
+
+public class RateLimiterTest {
+
+ @Test
+ public void testInvalidRenewTime() {
+ try {
+ RateLimiter.builder().permits(0).rateTime(100).timeUnit(TimeUnit.SECONDS).build();
+ fail("should have thrown exception: invalid rate, must be > 0");
+ } catch (IllegalArgumentException ie) {
+ // Ok
+ }
+
+ try {
+ RateLimiter.builder().permits(10).rateTime(0).timeUnit(TimeUnit.SECONDS).build();
+ fail("should have thrown exception: invalid rateTime, must be > 0");
+ } catch (IllegalArgumentException ie) {
+ // Ok
+ }
+ }
+
+ @Test
+ public void testClose() throws Exception {
+ RateLimiter rate = RateLimiter.builder().permits(1).rateTime(1000).timeUnit(TimeUnit.MILLISECONDS).build();
+ assertFalse(rate.isClosed());
+ rate.close();
+ assertTrue(rate.isClosed());
+ try {
+ rate.acquire();
+ fail("should have failed, executor is already closed");
+ } catch (IllegalArgumentException e) {
+ // ok
+ }
+ }
+
+ @Test
+ public void testAcquireBlock() throws Exception {
+ final long rateTimeMSec = 1000;
+ RateLimiter rate = RateLimiter.builder().permits(1).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
+ rate.acquire();
+ assertEquals(rate.getAvailablePermits(), 0);
+ long start = System.currentTimeMillis();
+ rate.acquire();
+ long end = System.currentTimeMillis();
+ // no permits are available: need to wait on acquire
+ assertTrue((end - start) > rateTimeMSec / 2);
+ rate.close();
+ }
+
+ @Test
+ public void testAcquire() throws Exception {
+ final long rateTimeMSec = 1000;
+ final int permits = 100;
+ RateLimiter rate = RateLimiter.builder().permits(permits).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < permits; i++) {
+ rate.acquire();
+ }
+ long end = System.currentTimeMillis();
+ assertTrue((end - start) < rateTimeMSec);
+ assertEquals(rate.getAvailablePermits(), 0);
+ rate.close();
+ }
+
+ @Test
+ public void testMultipleAcquire() throws Exception {
+ final long rateTimeMSec = 1000;
+ final int permits = 100;
+ final int acquirePermits = 50;
+ RateLimiter rate = RateLimiter.builder().permits(permits).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < permits / acquirePermits; i++) {
+ rate.acquire(acquirePermits);
+ }
+ long end = System.currentTimeMillis();
+ assertTrue((end - start) < rateTimeMSec);
+ assertEquals(rate.getAvailablePermits(), 0);
+ rate.close();
+ }
+
+ @Test
+ public void testTryAcquireNoPermits() {
+ final long rateTimeMSec = 1000;
+ RateLimiter rate = RateLimiter.builder().permits(1).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
+ assertTrue(rate.tryAcquire());
+ assertFalse(rate.tryAcquire());
+ assertEquals(rate.getAvailablePermits(), 0);
+ rate.close();
+ }
+
+ @Test
+ public void testTryAcquire() {
+ final long rateTimeMSec = 1000;
+ final int permits = 100;
+ RateLimiter rate = RateLimiter.builder().permits(permits).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
+ for (int i = 0; i < permits; i++) {
+ rate.tryAcquire();
+ }
+ assertEquals(rate.getAvailablePermits(), 0);
+ rate.close();
+ }
+
+ @Test
+ public void testTryAcquireMoreThanPermits() {
+ final long rateTimeMSec = 1000;
+ RateLimiter rate = RateLimiter.builder().permits(3).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
+ assertTrue(rate.tryAcquire(2));
+ assertEquals(rate.getAvailablePermits(), 1);
+
+ //try to acquire failed, not decrease availablePermits.
+ assertFalse(rate.tryAcquire(2));
+ assertEquals(rate.getAvailablePermits(), 1);
+
+ assertTrue(rate.tryAcquire(1));
+ assertEquals(rate.getAvailablePermits(), 0);
+
+ rate.close();
+ }
+
+ @Test
+ public void testMultipleTryAcquire() {
+ final long rateTimeMSec = 1000;
+ final int permits = 100;
+ final int acquirePermits = 50;
+ RateLimiter rate = RateLimiter.builder().permits(permits).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
+ for (int i = 0; i < permits / acquirePermits; i++) {
+ rate.tryAcquire(acquirePermits);
+ }
+ assertEquals(rate.getAvailablePermits(), 0);
+ rate.close();
+ }
+
+ @Test
+ public void testResetRate() throws Exception {
+ final long rateTimeMSec = 1000;
+ final int permits = 100;
+ RateLimiter rate = RateLimiter.builder().permits(permits).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
+ rate.tryAcquire(permits);
+ assertEquals(rate.getAvailablePermits(), 0);
+ // check after a rate-time: permits must be renewed
+ Thread.sleep(rateTimeMSec * 2);
+ assertEquals(rate.getAvailablePermits(), permits);
+
+ // change rate-time from 1sec to 5sec
+ rate.setRate(permits, 5 * rateTimeMSec, TimeUnit.MILLISECONDS, null);
+ assertEquals(rate.getAvailablePermits(), 100);
+ assertTrue(rate.tryAcquire(permits));
+ assertEquals(rate.getAvailablePermits(), 0);
+ // check after a rate-time: permits can't be renewed
+ Thread.sleep(rateTimeMSec);
+ assertEquals(rate.getAvailablePermits(), 0);
+
+ rate.close();
+ }
+
+ @Test
+ public void testDispatchRate() throws Exception {
+ final long rateTimeMSec = 1000;
+ final int permits = 100;
+ RateLimiter rate = RateLimiter.builder().permits(permits).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .isDispatchOrPrecisePublishRateLimiter(true)
+ .build();
+ rate.tryAcquire(100);
+ rate.tryAcquire(100);
+ rate.tryAcquire(100);
+ assertEquals(rate.getAvailablePermits(), 0);
+
+ Thread.sleep(rateTimeMSec * 2);
+ // check after two rate-time: acquiredPermits is 100
+ assertEquals(rate.getAvailablePermits(), 0);
+
+ Thread.sleep(rateTimeMSec);
+ // check after three rate-time: acquiredPermits is 0
+ assertTrue(rate.getAvailablePermits() > 0);
+
+ rate.close();
+ }
+
+ @Test
+ public void testRateLimiterWithPermitUpdater() throws Exception {
+ long permits = 10;
+ long rateTime = 1;
+ long newUpdatedRateLimit = 100L;
+ Supplier<Long> permitUpdater = () -> newUpdatedRateLimit;
+ RateLimiter limiter = RateLimiter.builder().permits(permits).rateTime(1).timeUnit(TimeUnit.SECONDS)
+ .permitUpdater(permitUpdater)
+ .build();
+ limiter.acquire();
+ Thread.sleep(rateTime * 3 * 1000);
+ assertEquals(limiter.getAvailablePermits(), newUpdatedRateLimit);
+ }
+
+ @Test
+ public void testRateLimiterWithFunction() {
+ final AtomicInteger atomicInteger = new AtomicInteger(0);
+ long permits = 10;
+ long rateTime = 1;
+ int reNewTime = 3;
+ RateLimitFunction rateLimitFunction = atomicInteger::incrementAndGet;
+ RateLimiter rateLimiter = RateLimiter.builder().permits(permits).rateTime(rateTime).timeUnit(TimeUnit.SECONDS)
+ .rateLimitFunction(rateLimitFunction)
+ .build();
+ for (int i = 0; i < reNewTime; i++) {
+ rateLimiter.renew();
+ }
+ assertEquals(reNewTime, atomicInteger.get());
+ }
+
+}