You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2023/12/15 17:23:40 UTC
(pulsar) branch master updated: [improve][broker] Pulsar Rate Limiting Refactoring changes (PIP-322) (#21681)
This is an automated email from the ASF dual-hosted git repository.
mmerli 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 c4cff0ab3da [improve][broker] Pulsar Rate Limiting Refactoring changes (PIP-322) (#21681)
c4cff0ab3da is described below
commit c4cff0ab3dac169c0a1418ef2f63f61604f6278e
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Dec 15 19:23:23 2023 +0200
[improve][broker] Pulsar Rate Limiting Refactoring changes (PIP-322) (#21681)
---
microbench/README.md | 43 +++
microbench/pom.xml | 133 ++++++++
.../broker/qos/AsyncTokenBucketBenchmark.java | 81 +++++
.../org/apache/pulsar/broker/qos/package-info.java | 10 +-
microbench/src/main/resources/log4j2.xml | 33 ++
pom.xml | 21 ++
.../org/apache/pulsar/broker/PulsarService.java | 12 +
.../apache/pulsar/broker/qos/AsyncTokenBucket.java | 344 +++++++++++++++++++++
.../broker/qos/AsyncTokenBucketBuilder.java} | 33 +-
.../broker/qos/DefaultMonotonicSnapshotClock.java | 89 ++++++
.../broker/qos/DynamicRateAsyncTokenBucket.java | 68 ++++
.../qos/DynamicRateAsyncTokenBucketBuilder.java | 72 +++++
.../broker/qos/FinalRateAsyncTokenBucket.java | 64 ++++
.../qos/FinalRateAsyncTokenBucketBuilder.java | 62 ++++
.../pulsar/broker/qos/MonotonicSnapshotClock.java | 58 ++++
.../org/apache/pulsar/broker/qos/package-info.java | 10 +-
.../pulsar/broker/resourcegroup/ResourceGroup.java | 6 +-
.../resourcegroup/ResourceGroupPublishLimiter.java | 141 +--------
.../broker/resourcegroup/ResourceGroupService.java | 1 -
.../broker/service/AbstractBaseDispatcher.java | 20 +-
.../pulsar/broker/service/AbstractTopic.java | 140 ++-------
.../pulsar/broker/service/BrokerService.java | 178 +----------
.../broker/service/PrecisePublishLimiter.java | 162 ----------
.../org/apache/pulsar/broker/service/Producer.java | 25 +-
.../pulsar/broker/service/PublishRateLimiter.java | 47 +--
.../broker/service/PublishRateLimiterDisable.java | 69 -----
.../broker/service/PublishRateLimiterImpl.java | 195 ++++++++----
.../apache/pulsar/broker/service/ServerCnx.java | 200 +++++-------
.../broker/service/ServerCnxThrottleTracker.java | 146 +++++++++
.../org/apache/pulsar/broker/service/Topic.java | 22 +-
.../apache/pulsar/broker/service/TransportCnx.java | 26 +-
.../service/nonpersistent/NonPersistentTopic.java | 6 -
.../service/persistent/DispatchRateLimiter.java | 96 ++----
.../persistent/GeoPersistentReplicator.java | 2 +-
.../PersistentDispatcherMultipleConsumers.java | 36 +--
.../PersistentDispatcherSingleActiveConsumer.java | 36 +--
.../service/persistent/PersistentReplicator.java | 8 +-
.../broker/service/persistent/PersistentTopic.java | 6 -
.../service/persistent/ShadowReplicator.java | 2 +-
.../service/persistent/SubscribeRateLimiter.java | 76 ++---
.../pulsar/broker/admin/TopicPoliciesTest.java | 28 +-
.../pulsar/broker/qos/AsyncTokenBucketTest.java | 103 ++++++
.../ResourceGroupRateLimiterTest.java | 16 +-
.../broker/service/AbstractBaseDispatcherTest.java | 2 +-
.../pulsar/broker/service/AbstractTopicTest.java | 4 +
.../pulsar/broker/service/BrokerServiceTest.java | 76 -----
.../service/MessagePublishBufferThrottleTest.java | 7 -
.../broker/service/PrecisePublishLimiterTest.java | 60 ----
.../service/PublishRateLimiterDisableTest.java | 11 +-
.../broker/service/PublishRateLimiterTest.java | 163 ++++------
.../broker/service/ReplicatorRateLimiterTest.java | 3 +
...Test.java => TopicPublishRateThrottleTest.java} | 67 +---
.../service/persistent/MessageDuplicationTest.java | 3 +
.../pulsar/broker/stats/PrometheusMetricsTest.java | 33 +-
.../client/api/MessageDispatchThrottlingTest.java | 149 +++------
.../client/impl/MessagePublishThrottlingTest.java | 89 +-----
.../impl/TopicPublishThrottlingInitTest.java | 9 -
.../test/resources/prometheus_metrics_sample.txt | 2 -
.../org/apache/pulsar/common/util/RateLimiter.java | 286 -----------------
.../apache/pulsar/common/util/RateLimiterTest.java | 248 ---------------
.../instance/stats/FunctionStatsManager.java | 17 +-
.../functions/instance/stats/SinkStatsManager.java | 17 +-
.../instance/stats/SourceStatsManager.java | 17 +-
63 files changed, 1954 insertions(+), 2235 deletions(-)
diff --git a/microbench/README.md b/microbench/README.md
new file mode 100644
index 00000000000..780e3a5a1d3
--- /dev/null
+++ b/microbench/README.md
@@ -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.
+
+-->
+
+# Microbenchmarks for Apache Pulsar
+
+This module contains microbenchmarks for Apache Pulsar.
+
+## Running the benchmarks
+
+The benchmarks are written using [JMH](http://openjdk.java.net/projects/code-tools/jmh/). To compile & run the benchmarks, use the following command:
+
+```bash
+# Compile everything for creating the shaded microbenchmarks.jar file
+mvn -Pcore-modules,microbench,-main -T 1C clean package
+
+# run the benchmarks using the standalone shaded jar in any environment
+java -jar microbench/target/microbenchmarks.jar
+```
+
+For fast recompiling of the benchmarks (without compiling Pulsar modules) and creating the shaded jar, you can use the following command:
+
+```bash
+mvn -Pmicrobench -pl microbench clean package
+```
+
diff --git a/microbench/pom.xml b/microbench/pom.xml
new file mode 100644
index 00000000000..a62876e8802
--- /dev/null
+++ b/microbench/pom.xml
@@ -0,0 +1,133 @@
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar</artifactId>
+ <version>3.2.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>microbench</artifactId>
+ <packaging>jar</packaging>
+ <name>Pulsar Microbenchmarks</name>
+
+ <properties>
+ <jmh.version>1.37</jmh.version>
+ </properties>
+
+ <profiles>
+ <profile>
+ <id>microbench</id>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <finalName>microbenchmarks</finalName>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <manifestEntries>
+ <Main-Class>org.openjdk.jmh.Main</Main-Class>
+ <Multi-Release>true</Multi-Release>
+ </manifestEntries>
+ </transformer>
+ </transformers>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.openjdk.jmh</groupId>
+ <artifactId>jmh-core</artifactId>
+ <version>${jmh.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.openjdk.jmh</groupId>
+ <artifactId>jmh-generator-annprocess</artifactId>
+ <version>${jmh.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-broker</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-deploy-plugin</artifactId>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skipTests>true</skipTests>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <annotationProcessorPaths combine.children="append">
+ <path>
+ <groupId>org.openjdk.jmh</groupId>
+ <artifactId>jmh-generator-annprocess</artifactId>
+ <version>${jmh.version}</version>
+ </path>
+ </annotationProcessorPaths>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
\ No newline at end of file
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
new file mode 100644
index 00000000000..4c069e72ea3
--- /dev/null
+++ b/microbench/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBenchmark.java
@@ -0,0 +1,81 @@
+/*
+ * 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.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.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;
+
+@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);
+
+ @Setup(Level.Iteration)
+ public void setup() {
+ long ratePerSecond = 100_000_000;
+ asyncTokenBucket = AsyncTokenBucket.builder().rate(ratePerSecond).clock(monotonicSnapshotClock)
+ .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)
+ @Warmup(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
+ public void consumeTokensBenchmark001Threads() {
+ asyncTokenBucket.consumeTokens(1);
+ }
+
+ @Threads(10)
+ @Benchmark
+ @Measurement(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
+ @Warmup(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
+ public void consumeTokensBenchmark010Threads() {
+ asyncTokenBucket.consumeTokens(1);
+ }
+
+ @Threads(100)
+ @Benchmark
+ @Measurement(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
+ @Warmup(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1)
+ public void consumeTokensBenchmark100Threads() {
+ asyncTokenBucket.consumeTokens(1);
+ }
+}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimitFunction.java b/microbench/src/main/java/org/apache/pulsar/broker/qos/package-info.java
similarity index 84%
copy from pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimitFunction.java
copy to microbench/src/main/java/org/apache/pulsar/broker/qos/package-info.java
index 0a4b62e9500..ccea21a210f 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimitFunction.java
+++ b/microbench/src/main/java/org/apache/pulsar/broker/qos/package-info.java
@@ -16,11 +16,7 @@
* 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();
-}
+ * Benchmarks for Pulsar broker Quality of Service (QoS) related classes.
+ */
+package org.apache.pulsar.broker.qos;
\ No newline at end of file
diff --git a/microbench/src/main/resources/log4j2.xml b/microbench/src/main/resources/log4j2.xml
new file mode 100644
index 00000000000..7ec5ed8169a
--- /dev/null
+++ b/microbench/src/main/resources/log4j2.xml
@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<Configuration status="WARN">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d{ISO8601_OFFSET_DATE_TIME_HHMM} [%t] %-5level %logger{36} - %msg%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Root level="warn">
+ <AppenderRef ref="Console"/>
+ </Root>
+ </Loggers>
+</Configuration>
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index f2352723555..cc06d24289f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2247,6 +2247,8 @@ flexible messaging model and an intuitive client API.</description>
<module>distribution</module>
<module>docker</module>
<module>tests</module>
+
+ <module>microbench</module>
</modules>
</profile>
@@ -2504,6 +2506,25 @@ flexible messaging model and an intuitive client API.</description>
<module>pulsar-sql</module>
</modules>
</profile>
+
+ <profile>
+ <id>microbench</id>
+ <modules>
+ <module>microbench</module>
+ </modules>
+ <properties>
+ <skipSourceReleaseAssembly>true</skipSourceReleaseAssembly>
+ <skipBuildDistribution>true</skipBuildDistribution>
+ <spotbugs.skip>true</spotbugs.skip>
+ <license.skip>true</license.skip>
+ <rat.skip>true</rat.skip>
+ <assembly.skipAssembly>true</assembly.skipAssembly>
+ <narPluginPhase>none</narPluginPhase>
+ <skipDocker>true</skipDocker>
+ <skipTests>true</skipTests>
+ <checkstyle.skip>true</checkstyle.skip>
+ </properties>
+ </profile>
</profiles>
<repositories>
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 02ea3bd7135..a04a4c137cc 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
@@ -91,6 +91,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.resourcegroup.ResourceGroupService;
import org.apache.pulsar.broker.resourcegroup.ResourceUsageTopicTransportManager;
import org.apache.pulsar.broker.resourcegroup.ResourceUsageTransportManager;
@@ -191,6 +193,7 @@ import org.slf4j.LoggerFactory;
public class PulsarService implements AutoCloseable, ShutdownService {
private static final Logger LOG = LoggerFactory.getLogger(PulsarService.class);
private static final double GRACEFUL_SHUTDOWN_TIMEOUT_RATIO_OF_TOTAL_TIMEOUT = 0.5d;
+ private static final int DEFAULT_MONOTONIC_CLOCK_GRANULARITY_MILLIS = 8;
private final ServiceConfiguration config;
private NamespaceService nsService = null;
private ManagedLedgerStorage managedLedgerClientFactory = null;
@@ -271,6 +274,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
private TransactionPendingAckStoreProvider transactionPendingAckStoreProvider;
private final ExecutorProvider transactionExecutorProvider;
+ private final DefaultMonotonicSnapshotClock monotonicSnapshotClock;
public enum State {
Init, Started, Closing, Closed
@@ -348,6 +352,9 @@ 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);
}
public MetadataStore createConfigurationMetadataStore(PulsarMetadataEventSynchronizer synchronizer)
@@ -596,6 +603,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
brokerClientSharedInternalExecutorProvider.shutdownNow();
brokerClientSharedScheduledExecutorProvider.shutdownNow();
brokerClientSharedTimer.stop();
+ monotonicSnapshotClock.close();
asyncCloseFutures.add(EventLoopUtil.shutdownGracefully(ioEventLoopGroup));
@@ -1777,6 +1785,10 @@ public class PulsarService implements AutoCloseable, ShutdownService {
return brokerService.getListenPortTls();
}
+ public MonotonicSnapshotClock getMonotonicSnapshotClock() {
+ return monotonicSnapshotClock;
+ }
+
public static WorkerConfig initializeWorkerConfigFromBrokerConfig(ServiceConfiguration brokerConfig,
String workerConfigFile) throws IOException {
WorkerConfig workerConfig = WorkerConfig.load(workerConfigFile);
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
new file mode 100644
index 00000000000..ac9a1f03e59
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucket.java
@@ -0,0 +1,344 @@
+/*
+ * 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.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+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.
+ * <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
+ * tokens available, indicating a need for throttling.
+ * 3. In case of throttling, the application should throttle in a way that is suitable for the use case
+ * and then call the "calculateThrottlingDuration" method to calculate the duration of the required pause.
+ * 4. After the pause duration, the application should verify if there are any available tokens by invoking the
+ * containsTokens method. If tokens are available, the application should cease throttling. However, if tokens are
+ * not available, the application should maintain the throttling and recompute the throttling duration. In a
+ * concurrent environment, it is advisable to use a throttling queue to ensure fair distribution of resources across
+ * throttled connections or clients. Once the throttling duration has elapsed, the application should select the next
+ * 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>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();
+ 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
+ // 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;
+ }
+
+ // 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");
+
+ private static final AtomicLongFieldUpdater<AsyncTokenBucket> REMAINDER_NANOS_UPDATER =
+ AtomicLongFieldUpdater.newUpdater(AsyncTokenBucket.class, "remainderNanos");
+
+ /**
+ * This field represents the number of tokens in the bucket. It is eventually consistent, as the
+ * pendingConsumedTokens are subtracted from the total number of tokens at most once during each "tick" or
+ * "increment", when time advances according to the configured resolution.
+ */
+ protected volatile long tokens;
+ /**
+ * This field represents the last time the tokens were updated, in nanoseconds.
+ * The configured clockSource is used to obtain the current nanoseconds.
+ * 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.
+ */
+ protected final long resolutionNanos;
+ /**
+ * This field is used to obtain the current monotonic clock time in nanoseconds.
+ */
+ private final MonotonicSnapshotClock 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
+ * which has a complex solution to prevent the CAS loop content problem.
+ */
+ private final LongAdder pendingConsumedTokens = new LongAdder();
+
+ protected AsyncTokenBucket(MonotonicSnapshotClock clockSource, long resolutionNanos) {
+ this.clockSource = clockSource;
+ this.resolutionNanos = resolutionNanos;
+ }
+
+ public static FinalRateAsyncTokenBucketBuilder builder() {
+ return new FinalRateAsyncTokenBucketBuilder();
+ }
+
+ public static DynamicRateAsyncTokenBucketBuilder builderForDynamicRate() {
+ return new DynamicRateAsyncTokenBucketBuilder();
+ }
+
+ protected abstract long getRatePeriodNanos();
+
+ 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 forceUpdateTokens 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.
+ *
+ * @param consumeTokens number of tokens to consume, can be 0 to update the tokens balance
+ * @param forceUpdateTokens if true, the tokens are updated even if the configured resolution hasn't passed
+ * @return the current number of tokens in the bucket or Long.MIN_VALUE when the number of tokens is unknown due
+ * to eventual consistency
+ */
+ private long consumeTokensAndMaybeUpdateTokensBalance(long consumeTokens, boolean forceUpdateTokens) {
+ if (consumeTokens < 0) {
+ throw new IllegalArgumentException("consumeTokens must be >= 0");
+ }
+ long currentNanos = clockSource.getTickNanos(forceUpdateTokens);
+ // check if the tokens should be updated immediately
+ if (shouldUpdateTokensImmediately(currentNanos, forceUpdateTokens)) {
+ // calculate the number of new tokens since the last update
+ long newTokens = calculateNewTokensSinceLastUpdate(currentNanos);
+ // calculate the total amount of tokens to consume in this update
+ // flush the pendingConsumedTokens by calling "sumThenReset"
+ long totalConsumedTokens = consumeTokens + pendingConsumedTokens.sumThenReset();
+ // update the tokens and return the current token value
+ return TOKENS_UPDATER.updateAndGet(this,
+ currentTokens ->
+ // after adding new tokens, limit the tokens to the capacity
+ Math.min(currentTokens + newTokens, getCapacity())
+ // subtract the consumed tokens
+ - totalConsumedTokens);
+ } else {
+ // eventual consistent fast path, tokens are not updated immediately
+
+ // 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;
+ }
+ }
+
+ /**
+ * 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 forceUpdateTokens is true, the tokens will be updated immediately.
+ *
+ * @param currentNanos the current monotonic clock time in nanoseconds
+ * @param forceUpdateTokens if true, the tokens will be updated immediately
+ * @return true if the tokens should be updated immediately, false otherwise
+ */
+ private boolean shouldUpdateTokensImmediately(long currentNanos, boolean forceUpdateTokens) {
+ long currentIncrement = resolutionNanos != 0 ? currentNanos / resolutionNanos : 0;
+ long currentLastIncrement = lastIncrement;
+ return currentIncrement == 0
+ || (currentIncrement > currentLastIncrement
+ && LAST_INCREMENT_UPDATER.compareAndSet(this, currentLastIncrement, currentIncrement))
+ || forceUpdateTokens;
+ }
+
+ /**
+ * 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.
+ *
+ * @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) {
+ long newTokens;
+ long previousLastNanos = LAST_NANOS_UPDATER.getAndSet(this, currentNanos);
+ if (previousLastNanos == 0) {
+ newTokens = 0;
+ } else {
+ long durationNanos = currentNanos - previousLastNanos + REMAINDER_NANOS_UPDATER.getAndSet(this, 0);
+ long currentRate = getRate();
+ long currentRatePeriodNanos = getRatePeriodNanos();
+ // new tokens is the amount of tokens that are created in the duration since the last update
+ // with the configured rate
+ newTokens = (durationNanos * currentRate) / currentRatePeriodNanos;
+ // carry forward the remainder nanos so that the rounding error is eliminated
+ long remainderNanos = durationNanos - ((newTokens * currentRatePeriodNanos) / currentRate);
+ if (remainderNanos > 0) {
+ REMAINDER_NANOS_UPDATER.addAndGet(this, remainderNanos);
+ }
+ }
+ return newTokens;
+ }
+
+ /**
+ * Eventually consume tokens from the bucket.
+ * The number of tokens is eventually consistent with the configured granularity of resolutionNanos.
+ *
+ * @param consumeTokens the number of tokens to consume
+ */
+ public void consumeTokens(long consumeTokens) {
+ consumeTokensAndMaybeUpdateTokensBalance(consumeTokens, false);
+ }
+
+ /**
+ * 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.
+ *
+ * @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.
+ */
+ 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;
+ }
+ }
+
+ /**
+ * Returns the current token balance. When forceUpdateTokens is true, the tokens balance is updated before
+ * returning. If forceUpdateTokens is false, the tokens balance could be updated if the last updated happened
+ * more than resolutionNanos nanoseconds ago.
+ *
+ * @param forceUpdateTokens if true, the tokens balance is updated before returning
+ * @return the current token balance
+ */
+ protected long tokens(boolean forceUpdateTokens) {
+ long currentTokens = consumeTokensAndMaybeUpdateTokensBalance(0, forceUpdateTokens);
+ 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;
+ }
+ }
+
+ /**
+ * 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.
+ */
+ public long calculateThrottlingDuration() {
+ long currentTokens = consumeTokensAndMaybeUpdateTokensBalance(0, true);
+ if (currentTokens == Long.MIN_VALUE) {
+ throw new IllegalArgumentException(
+ "Unexpected result from updateAndConsumeTokens with forceUpdateTokens set to true");
+ }
+ if (currentTokens > 0) {
+ return 0L;
+ }
+ // currentTokens is negative, so subtracting a negative value results in adding the absolute value (-(-x) -> +x)
+ long needTokens = getTargetAmountOfTokensAfterThrottling() - currentTokens;
+ return (needTokens * getRatePeriodNanos()) / getRate();
+ }
+
+ 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.
+ */
+ public final long getTokens() {
+ return tokens(false);
+ }
+
+ public abstract long getRate();
+
+ /**
+ * 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.
+ *
+ * @return true if the bucket contains tokens, false otherwise
+ */
+ public boolean containsTokens() {
+ return containsTokens(false);
+ }
+
+ /**
+ * Checks if the bucket contains tokens.
+ * The token balance is updated before the comparison if the configured resolutionNanos has passed since the last
+ * update. The token balance is also updated when forceUpdateTokens is true.
+ * It's possible that the returned result is not definite since the token balance is eventually consistent.
+ *
+ * @param forceUpdateTokens if true, the token balance is updated before the comparison
+ * @return true if the bucket contains tokens, false otherwise
+ */
+ public boolean containsTokens(boolean forceUpdateTokens) {
+ return tokens(forceUpdateTokens) > 0;
+ }
+
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBuilder.java
similarity index 52%
copy from pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java
copy to pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBuilder.java
index 8dfed77814b..ee256d5a37d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucketBuilder.java
@@ -16,17 +16,30 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.broker.service;
-import static org.testng.Assert.assertTrue;
-import org.testng.annotations.Test;
+package org.apache.pulsar.broker.qos;
-public class PublishRateLimiterDisableTest {
+// CHECKSTYLE.OFF: ClassTypeParameterName
+public abstract class AsyncTokenBucketBuilder<SELF extends AsyncTokenBucketBuilder<SELF>> {
+ protected MonotonicSnapshotClock clock = AsyncTokenBucket.DEFAULT_SNAPSHOT_CLOCK;
+ protected long resolutionNanos = AsyncTokenBucket.defaultResolutionNanos;
- // GH issue #10603
- @Test
- void shouldAlwaysAllowAcquire() {
- PublishRateLimiterDisable publishRateLimiter = PublishRateLimiterDisable.DISABLED_RATE_LIMITER;
- assertTrue(publishRateLimiter.tryAcquire(Integer.MAX_VALUE, Long.MAX_VALUE));
+ protected AsyncTokenBucketBuilder() {
}
-}
\ No newline at end of file
+
+ protected SELF self() {
+ return (SELF) this;
+ }
+
+ public SELF clock(MonotonicSnapshotClock clock) {
+ this.clock = clock;
+ return self();
+ }
+
+ public SELF resolutionNanos(long resolutionNanos) {
+ this.resolutionNanos = resolutionNanos;
+ return self();
+ }
+
+ public abstract AsyncTokenBucket build();
+}
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
new file mode 100644
index 00000000000..df3843921ed
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java
@@ -0,0 +1,89 @@
+/*
+ * 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.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Default implementation of {@link MonotonicSnapshotClock}.
+ *
+ * Starts a daemon thread that updates the snapshot value periodically with a configured interval. The close method
+ * should be called to stop the thread.
+ */
+public class DefaultMonotonicSnapshotClock implements MonotonicSnapshotClock, AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultMonotonicSnapshotClock.class);
+ private final long sleepMillis;
+ private final int sleepNanos;
+ private final LongSupplier clockSource;
+ private final Thread thread;
+ 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");
+ }
+ this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
+ this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
+ this.clockSource = clockSource;
+ updateSnapshotTickNanos();
+ thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public long getTickNanos(boolean requestSnapshot) {
+ if (requestSnapshot) {
+ updateSnapshotTickNanos();
+ }
+ return snapshotTickNanos;
+ }
+
+ private void updateSnapshotTickNanos() {
+ snapshotTickNanos = clockSource.getAsLong();
+ }
+
+ private void snapshotLoop() {
+ try {
+ while (!Thread.currentThread().isInterrupted()) {
+ updateSnapshotTickNanos();
+ try {
+ Thread.sleep(sleepMillis, sleepNanos);
+ } catch (InterruptedException e) {
+ Thread.currentThread().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);
+ }
+ }
+
+ @Override
+ public void close() {
+ thread.interrupt();
+ }
+}
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
new file mode 100644
index 00000000000..8edc73d1f51
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DynamicRateAsyncTokenBucket.java
@@ -0,0 +1,68 @@
+/*
+ * 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.function.LongSupplier;
+
+/**
+ * A subclass of {@link AsyncTokenBucket} that represents a token bucket with a dynamic rate.
+ * The rate and capacity of the token bucket can change over time based on the rate function and capacity factor.
+ */
+public class DynamicRateAsyncTokenBucket extends AsyncTokenBucket {
+ private final LongSupplier rateFunction;
+ private final LongSupplier ratePeriodNanosFunction;
+ private final double capacityFactor;
+
+ private final double targetFillFactorAfterThrottling;
+
+ protected DynamicRateAsyncTokenBucket(double capacityFactor, LongSupplier rateFunction,
+ MonotonicSnapshotClock clockSource, LongSupplier ratePeriodNanosFunction,
+ long resolutionNanos, double initialTokensFactor,
+ double targetFillFactorAfterThrottling) {
+ super(clockSource, resolutionNanos);
+ this.capacityFactor = capacityFactor;
+ this.rateFunction = rateFunction;
+ this.ratePeriodNanosFunction = ratePeriodNanosFunction;
+ this.targetFillFactorAfterThrottling = targetFillFactorAfterThrottling;
+ this.tokens = (long) (rateFunction.getAsLong() * initialTokensFactor);
+ tokens(false);
+ }
+
+ @Override
+ protected long getRatePeriodNanos() {
+ return ratePeriodNanosFunction.getAsLong();
+ }
+
+ @Override
+ protected long getTargetAmountOfTokensAfterThrottling() {
+ return (long) (getRate() * targetFillFactorAfterThrottling);
+ }
+
+ @Override
+ public long getCapacity() {
+ return capacityFactor == 1.0d ? getRate() : (long) (getRate() * capacityFactor);
+ }
+
+ @Override
+ public long getRate() {
+ return rateFunction.getAsLong();
+ }
+
+}
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
new file mode 100644
index 00000000000..22270484c72
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DynamicRateAsyncTokenBucketBuilder.java
@@ -0,0 +1,72 @@
+/*
+ * 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.function.LongSupplier;
+
+/**
+ * A builder class for creating instances of {@link DynamicRateAsyncTokenBucket}.
+ */
+public class DynamicRateAsyncTokenBucketBuilder
+ extends AsyncTokenBucketBuilder<DynamicRateAsyncTokenBucketBuilder> {
+ protected LongSupplier rateFunction;
+ protected double capacityFactor = 1.0d;
+ protected double initialFillFactor = 1.0d;
+ protected LongSupplier ratePeriodNanosFunction;
+ protected double targetFillFactorAfterThrottling = 0.01d;
+
+ protected DynamicRateAsyncTokenBucketBuilder() {
+ }
+
+ public DynamicRateAsyncTokenBucketBuilder rateFunction(LongSupplier rateFunction) {
+ this.rateFunction = rateFunction;
+ return this;
+ }
+
+ public DynamicRateAsyncTokenBucketBuilder ratePeriodNanosFunction(LongSupplier ratePeriodNanosFunction) {
+ this.ratePeriodNanosFunction = ratePeriodNanosFunction;
+ return this;
+ }
+
+ public DynamicRateAsyncTokenBucketBuilder capacityFactor(double capacityFactor) {
+ this.capacityFactor = capacityFactor;
+ return this;
+ }
+
+ public DynamicRateAsyncTokenBucketBuilder initialFillFactor(double initialFillFactor) {
+ this.initialFillFactor = initialFillFactor;
+ return this;
+ }
+
+ public DynamicRateAsyncTokenBucketBuilder targetFillFactorAfterThrottling(
+ double targetFillFactorAfterThrottling) {
+ this.targetFillFactorAfterThrottling = targetFillFactorAfterThrottling;
+ return this;
+ }
+
+ @Override
+ public AsyncTokenBucket build() {
+ return new DynamicRateAsyncTokenBucket(this.capacityFactor, this.rateFunction,
+ this.clock,
+ this.ratePeriodNanosFunction, this.resolutionNanos,
+ 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
new file mode 100644
index 00000000000..627c5ee1334
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/FinalRateAsyncTokenBucket.java
@@ -0,0 +1,64 @@
+/*
+ * 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;
+
+/**
+ * A subclass of {@link AsyncTokenBucket} that represents a token bucket with a rate which is final.
+ * The rate and capacity of the token bucket are constant and do not change over time.
+ */
+class FinalRateAsyncTokenBucket extends AsyncTokenBucket {
+ private final long capacity;
+ private final long rate;
+ private final long ratePeriodNanos;
+ private final long targetAmountOfTokensAfterThrottling;
+
+ protected FinalRateAsyncTokenBucket(long capacity, long rate, MonotonicSnapshotClock clockSource,
+ long ratePeriodNanos, long resolutionNanos, long initialTokens) {
+ super(clockSource, resolutionNanos);
+ 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.tokens = initialTokens;
+ tokens(false);
+ }
+
+ @Override
+ protected final long getRatePeriodNanos() {
+ return ratePeriodNanos;
+ }
+
+ @Override
+ protected final long getTargetAmountOfTokensAfterThrottling() {
+ return targetAmountOfTokensAfterThrottling;
+ }
+
+ @Override
+ public final long getCapacity() {
+ return capacity;
+ }
+
+ @Override
+ public final long getRate() {
+ return rate;
+ }
+
+}
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
new file mode 100644
index 00000000000..ff4ed53c6c7
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/FinalRateAsyncTokenBucketBuilder.java
@@ -0,0 +1,62 @@
+/*
+ * 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;
+
+/**
+ * A builder class for creating instances of {@link FinalRateAsyncTokenBucket}.
+ */
+public class FinalRateAsyncTokenBucketBuilder
+ extends AsyncTokenBucketBuilder<FinalRateAsyncTokenBucketBuilder> {
+ protected Long capacity;
+ protected Long initialTokens;
+ protected Long rate;
+ protected long ratePeriodNanos = AsyncTokenBucket.ONE_SECOND_NANOS;
+
+ protected FinalRateAsyncTokenBucketBuilder() {
+ }
+
+ public FinalRateAsyncTokenBucketBuilder rate(long rate) {
+ this.rate = rate;
+ return this;
+ }
+
+ public FinalRateAsyncTokenBucketBuilder ratePeriodNanos(long ratePeriodNanos) {
+ this.ratePeriodNanos = ratePeriodNanos;
+ return this;
+ }
+
+ public FinalRateAsyncTokenBucketBuilder capacity(long capacity) {
+ this.capacity = capacity;
+ return this;
+ }
+
+ public FinalRateAsyncTokenBucketBuilder initialTokens(long initialTokens) {
+ this.initialTokens = initialTokens;
+ return this;
+ }
+
+ public AsyncTokenBucket build() {
+ return new FinalRateAsyncTokenBucket(this.capacity != null ? this.capacity : this.rate, this.rate,
+ this.clock,
+ this.ratePeriodNanos, this.resolutionNanos,
+ this.initialTokens != null ? this.initialTokens : this.rate
+ );
+ }
+}
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/MonotonicSnapshotClock.java
new file mode 100644
index 00000000000..8f61bd5125b
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/MonotonicSnapshotClock.java
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+/**
+ * An interface representing a clock that provides a monotonic counter in nanoseconds.
+ * The counter is guaranteed to be monotonic, ensuring it will always increase or remain constant, but never decrease.
+ *
+ * Monotonicity ensures the time will always progress forward, making it ideal for measuring elapsed time.
+ * 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 {
+ /**
+ * 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);
+}
\ No newline at end of file
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimitFunction.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/package-info.java
similarity index 84%
rename from pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimitFunction.java
rename to pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/package-info.java
index 0a4b62e9500..1078d86894e 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimitFunction.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/package-info.java
@@ -16,11 +16,7 @@
* 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();
-}
+ * Pulsar broker Quality of Service (QoS) related classes.
+ */
+package org.apache.pulsar.broker.qos;
\ 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 ef40a18ab08..0aa355eee7d 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
@@ -81,7 +81,8 @@ public class ResourceGroup {
this.setResourceGroupMonitoringClassFields();
this.setResourceGroupConfigParameters(rgConfig);
this.setDefaultResourceUsageTransportHandlers();
- this.resourceGroupPublishLimiter = new ResourceGroupPublishLimiter(rgConfig, rgs.getPulsar().getExecutor());
+ this.resourceGroupPublishLimiter = new ResourceGroupPublishLimiter(rgConfig, rgs.getPulsar()
+ .getMonotonicSnapshotClock());
log.info("attaching publish rate limiter {} to {} get {}", this.resourceGroupPublishLimiter, name,
this.getResourceGroupPublishLimiter());
}
@@ -96,7 +97,8 @@ public class ResourceGroup {
this.resourceGroupName = rgName;
this.setResourceGroupMonitoringClassFields();
this.setResourceGroupConfigParameters(rgConfig);
- this.resourceGroupPublishLimiter = new ResourceGroupPublishLimiter(rgConfig, rgs.getPulsar().getExecutor());
+ this.resourceGroupPublishLimiter = new ResourceGroupPublishLimiter(rgConfig, rgs.getPulsar()
+ .getMonotonicSnapshotClock());
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 85e00bb2f87..a733db555a3 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,53 +18,22 @@
*/
package org.apache.pulsar.broker.resourcegroup;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.broker.qos.MonotonicSnapshotClock;
import org.apache.pulsar.broker.resourcegroup.ResourceGroup.BytesAndMessagesCount;
-import org.apache.pulsar.broker.service.PublishRateLimiter;
+import org.apache.pulsar.broker.service.PublishRateLimiterImpl;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.PublishRate;
import org.apache.pulsar.common.policies.data.ResourceGroup;
-import org.apache.pulsar.common.util.RateLimitFunction;
-import org.apache.pulsar.common.util.RateLimiter;
-public class ResourceGroupPublishLimiter implements PublishRateLimiter, RateLimitFunction, AutoCloseable {
- protected volatile long publishMaxMessageRate = 0;
- protected volatile long publishMaxByteRate = 0;
- protected volatile boolean publishThrottlingEnabled = false;
- private volatile RateLimiter publishRateLimiterOnMessage;
- private volatile RateLimiter publishRateLimiterOnByte;
- private final ScheduledExecutorService scheduledExecutorService;
+public class ResourceGroupPublishLimiter extends PublishRateLimiterImpl {
+ private volatile long publishMaxMessageRate;
+ private volatile long publishMaxByteRate;
- ConcurrentHashMap<String, RateLimitFunction> rateLimitFunctionMap = new ConcurrentHashMap<>();
-
- public ResourceGroupPublishLimiter(ResourceGroup resourceGroup, ScheduledExecutorService scheduledExecutorService) {
- this.scheduledExecutorService = scheduledExecutorService;
+ public ResourceGroupPublishLimiter(ResourceGroup resourceGroup, MonotonicSnapshotClock monotonicSnapshotClock) {
+ super(monotonicSnapshotClock);
update(resourceGroup);
}
- @Override
- public void checkPublishRate() {
- // No-op
- }
-
- @Override
- public void incrementPublishCount(int numOfMessages, long msgSizeInBytes) {
- // No-op
- }
-
- @Override
- public boolean resetPublishCount() {
- return true;
- }
-
- @Override
- public boolean isPublishRateExceeded() {
- return false;
- }
-
@Override
public void update(Policies policies, String clusterName) {
// No-op
@@ -94,102 +63,12 @@ public class ResourceGroupPublishLimiter implements PublishRateLimiter, RateLimi
publishRateInMsgs = resourceGroup.getPublishRateInMsgs() == null
? -1 : resourceGroup.getPublishRateInMsgs();
}
-
update(publishRateInMsgs, publishRateInBytes);
}
public void update(long publishRateInMsgs, long publishRateInBytes) {
- replaceLimiters(() -> {
- if (publishRateInMsgs > 0 || publishRateInBytes > 0) {
- this.publishThrottlingEnabled = true;
- this.publishMaxMessageRate = Math.max(publishRateInMsgs, 0);
- this.publishMaxByteRate = Math.max(publishRateInBytes, 0);
- if (this.publishMaxMessageRate > 0) {
- publishRateLimiterOnMessage = RateLimiter.builder()
- .scheduledExecutorService(scheduledExecutorService)
- .permits(publishMaxMessageRate)
- .rateTime(1L)
- .timeUnit(TimeUnit.SECONDS)
- .rateLimitFunction(this::apply)
- .build();
- }
- if (this.publishMaxByteRate > 0) {
- publishRateLimiterOnByte =
- RateLimiter.builder()
- .scheduledExecutorService(scheduledExecutorService)
- .permits(publishMaxByteRate)
- .rateTime(1L)
- .timeUnit(TimeUnit.SECONDS)
- .rateLimitFunction(this::apply)
- .build();
- }
- } else {
- this.publishMaxMessageRate = 0;
- this.publishMaxByteRate = 0;
- this.publishThrottlingEnabled = false;
- publishRateLimiterOnMessage = null;
- publishRateLimiterOnByte = null;
- }
- });
- }
-
- public boolean tryAcquire(int numbers, long bytes) {
- return (publishRateLimiterOnMessage == null || publishRateLimiterOnMessage.tryAcquire(numbers))
- && (publishRateLimiterOnByte == null || publishRateLimiterOnByte.tryAcquire(bytes));
- }
-
- public void registerRateLimitFunction(String name, RateLimitFunction func) {
- rateLimitFunctionMap.put(name, func);
- }
-
- public void unregisterRateLimitFunction(String name) {
- rateLimitFunctionMap.remove(name);
- }
-
- private void replaceLimiters(Runnable updater) {
- RateLimiter previousPublishRateLimiterOnMessage = publishRateLimiterOnMessage;
- publishRateLimiterOnMessage = null;
- RateLimiter previousPublishRateLimiterOnByte = publishRateLimiterOnByte;
- publishRateLimiterOnByte = null;
- try {
- if (updater != null) {
- updater.run();
- }
- } finally {
- // Close previous limiters to prevent resource leakages.
- // Delay closing of previous limiters after new ones are in place so that updating the limiter
- // doesn't cause unavailability.
- if (previousPublishRateLimiterOnMessage != null) {
- previousPublishRateLimiterOnMessage.close();
- }
- if (previousPublishRateLimiterOnByte != null) {
- previousPublishRateLimiterOnByte.close();
- }
- }
- }
-
- @Override
- public void close() {
- // Unblock any producers, consumers waiting first.
- // This needs to be done before replacing the filters to null
- this.apply();
- replaceLimiters(null);
- }
-
- @Override
- public void apply() {
- // Make sure that both the rate limiters are applied before opening the flood gates.
- RateLimiter currentTopicPublishRateLimiterOnMessage = publishRateLimiterOnMessage;
- RateLimiter currentTopicPublishRateLimiterOnByte = publishRateLimiterOnByte;
- if ((currentTopicPublishRateLimiterOnMessage != null
- && currentTopicPublishRateLimiterOnMessage.getAvailablePermits() <= 0)
- || (currentTopicPublishRateLimiterOnByte != null
- && currentTopicPublishRateLimiterOnByte.getAvailablePermits() <= 0)) {
- return;
- }
-
- for (Map.Entry<String, RateLimitFunction> entry: rateLimitFunctionMap.entrySet()) {
- entry.getValue().apply();
- }
+ this.publishMaxMessageRate = publishRateInMsgs;
+ this.publishMaxByteRate = publishRateInBytes;
+ updateTokenBuckets(publishRateInMsgs, publishRateInBytes);
}
}
\ No newline at end of file
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java
index d3f8eb7613a..48419d7c451 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java
@@ -173,7 +173,6 @@ public class ResourceGroupService implements AutoCloseable{
throw new PulsarAdminException(errMesg);
}
- rg.resourceGroupPublishLimiter.close();
rg.resourceGroupPublishLimiter = null;
resourceGroupsMap.remove(name);
}
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 b36389ab2da..ea68d588896 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
@@ -315,10 +315,10 @@ public abstract class AbstractBaseDispatcher extends EntryFilterSupport implemen
|| (cursor != null && !cursor.isActive())) {
long permits = dispatchThrottlingOnBatchMessageEnabled ? totalEntries : totalMessagesSent;
topic.getBrokerDispatchRateLimiter().ifPresent(rateLimiter ->
- rateLimiter.tryDispatchPermit(permits, totalBytesSent));
+ rateLimiter.consumeDispatchQuota(permits, totalBytesSent));
topic.getDispatchRateLimiter().ifPresent(rateLimter ->
- rateLimter.tryDispatchPermit(permits, totalBytesSent));
- getRateLimiter().ifPresent(rateLimiter -> rateLimiter.tryDispatchPermit(permits, totalBytesSent));
+ rateLimter.consumeDispatchQuota(permits, totalBytesSent));
+ getRateLimiter().ifPresent(rateLimiter -> rateLimiter.consumeDispatchQuota(permits, totalBytesSent));
}
}
@@ -356,16 +356,6 @@ public abstract class AbstractBaseDispatcher extends EntryFilterSupport implemen
protected abstract void reScheduleRead();
- protected boolean reachDispatchRateLimit(DispatchRateLimiter dispatchRateLimiter) {
- if (dispatchRateLimiter.isDispatchRateLimitingEnabled()) {
- if (!dispatchRateLimiter.hasMessageDispatchPermit()) {
- reScheduleRead();
- return true;
- }
- }
- return false;
- }
-
protected Pair<Integer, Long> updateMessagesToRead(DispatchRateLimiter dispatchRateLimiter,
int messagesToRead, long bytesToRead) {
// update messagesToRead according to available dispatch rate limit.
@@ -376,11 +366,11 @@ public abstract class AbstractBaseDispatcher extends EntryFilterSupport implemen
protected static Pair<Integer, Long> computeReadLimits(int messagesToRead, int availablePermitsOnMsg,
long bytesToRead, long availablePermitsOnByte) {
- if (availablePermitsOnMsg > 0) {
+ if (availablePermitsOnMsg >= 0) {
messagesToRead = Math.min(messagesToRead, availablePermitsOnMsg);
}
- if (availablePermitsOnByte > 0) {
+ if (availablePermitsOnByte >= 0) {
bytesToRead = Math.min(bytesToRead, availablePermitsOnByte);
}
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 08022dd9f76..837f073b00d 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
@@ -118,9 +118,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
protected volatile Boolean isAllowAutoUpdateSchema;
- protected volatile PublishRateLimiter topicPublishRateLimiter = PublishRateLimiter.DISABLED_RATE_LIMITER;
- private final Object topicPublishRateLimiterLock = new Object();
-
+ protected volatile PublishRateLimiter topicPublishRateLimiter;
protected volatile ResourceGroupPublishLimiter resourceGroupPublishLimiter;
protected boolean preciseTopicPublishRateLimitingEnable;
@@ -158,6 +156,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
protected final LongAdder bytesOutFromRemovedSubscriptions = new LongAdder();
protected volatile Pair<String, List<EntryFilter>> entryFilters;
protected volatile boolean transferring = false;
+ private volatile List<PublishRateLimiter> activeRateLimiters;
public AbstractTopic(String topic, BrokerService brokerService) {
this.topic = topic;
@@ -172,6 +171,8 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
this.lastActive = System.nanoTime();
this.preciseTopicPublishRateLimitingEnable = config.isPreciseTopicPublishRateLimiterEnable();
+ topicPublishRateLimiter = new PublishRateLimiterImpl(brokerService.getPulsar().getMonotonicSnapshotClock());
+ updateActiveRateLimiters();
}
public SubscribeRate getSubscribeRate() {
@@ -565,16 +566,6 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
return null;
}
- @Override
- public void disableCnxAutoRead() {
- producers.values().forEach(producer -> producer.getCnx().disableCnxAutoRead());
- }
-
- @Override
- public void enableCnxAutoRead() {
- producers.values().forEach(producer -> producer.getCnx().enableCnxAutoRead());
- }
-
protected boolean hasLocalProducers() {
if (producers.isEmpty()) {
return false;
@@ -898,57 +889,35 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
.register();
@Override
- public void checkTopicPublishThrottlingRate() {
- this.topicPublishRateLimiter.checkPublishRate();
- }
+ public void incrementPublishCount(Producer producer, int numOfMessages, long msgSizeInBytes) {
+ handlePublishThrottling(producer, numOfMessages, msgSizeInBytes);
- @Override
- public void incrementPublishCount(int numOfMessages, long msgSizeInBytes) {
- // increase topic publish rate limiter
- this.topicPublishRateLimiter.incrementPublishCount(numOfMessages, msgSizeInBytes);
- // increase broker publish rate limiter
- getBrokerPublishRateLimiter().incrementPublishCount(numOfMessages, msgSizeInBytes);
// increase counters
bytesInCounter.add(msgSizeInBytes);
msgInCounter.add(numOfMessages);
}
- @Override
- public void resetTopicPublishCountAndEnableReadIfRequired() {
- // broker rate not exceeded. and completed topic limiter reset.
- if (!getBrokerPublishRateLimiter().isPublishRateExceeded() && topicPublishRateLimiter.resetPublishCount()) {
- enableProducerReadForPublishRateLimiting();
+ private void handlePublishThrottling(Producer producer, int numOfMessages, long msgSizeInBytes) {
+ // consume tokens from rate limiters and possibly throttle the connection that published the message
+ // if it's publishing too fast. Each connection will be throttled lazily when they publish messages.
+ for (PublishRateLimiter rateLimiter : activeRateLimiters) {
+ rateLimiter.handlePublishThrottling(producer, numOfMessages, msgSizeInBytes);
}
}
- public void updateDispatchRateLimiter() {
- }
-
- @Override
- public void resetBrokerPublishCountAndEnableReadIfRequired(boolean doneBrokerReset) {
- // topic rate not exceeded, and completed broker limiter reset.
- if (!topicPublishRateLimiter.isPublishRateExceeded() && doneBrokerReset) {
- enableProducerReadForPublishRateLimiting();
+ private void updateActiveRateLimiters() {
+ List<PublishRateLimiter> updatedRateLimiters = new ArrayList<>();
+ updatedRateLimiters.add(this.topicPublishRateLimiter);
+ updatedRateLimiters.add(getBrokerPublishRateLimiter());
+ if (isResourceGroupRateLimitingEnabled()) {
+ updatedRateLimiters.add(resourceGroupPublishLimiter);
}
+ activeRateLimiters = updatedRateLimiters.stream().filter(Objects::nonNull).toList();
}
- /**
- * it sets cnx auto-readable if producer's cnx is disabled due to publish-throttling.
- */
- protected void enableProducerReadForPublishRateLimiting() {
- if (producers != null) {
- producers.values().forEach(producer -> {
- producer.getCnx().cancelPublishRateLimiting();
- producer.getCnx().enableCnxAutoRead();
- });
- }
+ public void updateDispatchRateLimiter() {
}
- protected void disableProducerRead() {
- if (producers != null) {
- producers.values().forEach(producer -> producer.getCnx().disableCnxAutoRead());
- }
- }
protected void checkTopicFenced() throws BrokerServiceException {
if (isFenced) {
@@ -1099,35 +1068,6 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
return usageCount;
}
- @Override
- public boolean isPublishRateExceeded() {
- // either topic or broker publish rate exceeded.
- return this.topicPublishRateLimiter.isPublishRateExceeded()
- || getBrokerPublishRateLimiter().isPublishRateExceeded();
- }
-
- @Override
- public boolean isResourceGroupPublishRateExceeded(int numMessages, int bytes) {
- return this.resourceGroupRateLimitingEnabled
- && !this.resourceGroupPublishLimiter.tryAcquire(numMessages, bytes);
- }
-
- @Override
- public boolean isResourceGroupRateLimitingEnabled() {
- return this.resourceGroupRateLimitingEnabled;
- }
-
- @Override
- public boolean isTopicPublishRateExceeded(int numberMessages, int bytes) {
- // whether topic publish rate exceed if precise rate limit is enable
- return preciseTopicPublishRateLimitingEnable && !this.topicPublishRateLimiter.tryAcquire(numberMessages, bytes);
- }
-
- @Override
- public boolean isBrokerPublishRateExceeded() {
- // whether broker publish rate exceed
- return getBrokerPublishRateLimiter().isPublishRateExceeded();
- }
public PublishRateLimiter getTopicPublishRateLimiter() {
return topicPublishRateLimiter;
@@ -1169,19 +1109,15 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
if (resourceGroup != null) {
this.resourceGroupRateLimitingEnabled = true;
this.resourceGroupPublishLimiter = resourceGroup.getResourceGroupPublishLimiter();
- this.resourceGroupPublishLimiter.registerRateLimitFunction(this.getName(), this::enableCnxAutoRead);
log.info("Using resource group {} rate limiter for topic {}", rgName, topic);
- return;
}
} else {
if (this.resourceGroupRateLimitingEnabled) {
- this.resourceGroupPublishLimiter.unregisterRateLimitFunction(this.getName());
this.resourceGroupPublishLimiter = null;
this.resourceGroupRateLimitingEnabled = false;
}
- /* Namespace detached from resource group. Enable the producer read */
- enableProducerReadForPublishRateLimiting();
}
+ updateActiveRateLimiters();
}
public void updateEntryFilters() {
@@ -1292,37 +1228,15 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
* update topic publish dispatcher for this topic.
*/
public void updatePublishDispatcher() {
- synchronized (topicPublishRateLimiterLock) {
- PublishRate publishRate = topicPolicies.getPublishRate().get();
- if (publishRate.publishThrottlingRateInByte > 0 || publishRate.publishThrottlingRateInMsg > 0) {
- log.info("Enabling publish rate limiting {} on topic {}", publishRate, getName());
- if (!preciseTopicPublishRateLimitingEnable) {
- this.brokerService.setupTopicPublishRateLimiterMonitor();
- }
-
- if (this.topicPublishRateLimiter == null
- || this.topicPublishRateLimiter == PublishRateLimiter.DISABLED_RATE_LIMITER) {
- // create new rateLimiter if rate-limiter is disabled
- if (preciseTopicPublishRateLimitingEnable) {
- this.topicPublishRateLimiter = new PrecisePublishLimiter(publishRate,
- () -> this.enableCnxAutoRead(), brokerService.pulsar().getExecutor());
- } else {
- this.topicPublishRateLimiter = new PublishRateLimiterImpl(publishRate);
- }
- } else {
- this.topicPublishRateLimiter.update(publishRate);
- }
- } else {
- if (log.isDebugEnabled()) {
- log.debug("Disabling publish throttling for {}", this.topic);
- }
- if (topicPublishRateLimiter != null) {
- topicPublishRateLimiter.close();
- }
- this.topicPublishRateLimiter = PublishRateLimiter.DISABLED_RATE_LIMITER;
- enableProducerReadForPublishRateLimiting();
+ PublishRate publishRate = topicPolicies.getPublishRate().get();
+ if (publishRate.publishThrottlingRateInByte > 0 || publishRate.publishThrottlingRateInMsg > 0) {
+ log.info("Enabling publish rate limiting {} on topic {}", publishRate, getName());
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug("Disabling publish throttling for {}", this.topic);
}
}
+ this.topicPublishRateLimiter.update(publishRate);
}
// subscriptionTypesEnabled is dynamic and can be updated online.
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 b3a7e84fc09..b15ffc755a6 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
@@ -26,6 +26,7 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.pulsar.common.naming.SystemTopicNames.isTransactionInternalName;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Queues;
+import com.google.common.util.concurrent.RateLimiter;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
@@ -169,7 +170,6 @@ import org.apache.pulsar.common.stats.Metrics;
import org.apache.pulsar.common.util.FieldParser;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.GracefulExecutorServicesShutdown;
-import org.apache.pulsar.common.util.RateLimiter;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
import org.apache.pulsar.common.util.netty.ChannelFutures;
@@ -242,10 +242,8 @@ public class BrokerService implements Closeable {
private final ScheduledExecutorService messageExpiryMonitor;
private final ScheduledExecutorService compactionMonitor;
private final ScheduledExecutorService consumedLedgersMonitor;
- protected final PublishRateLimiterMonitor topicPublishRateLimiterMonitor;
- protected final PublishRateLimiterMonitor brokerPublishRateLimiterMonitor;
private ScheduledExecutorService deduplicationSnapshotMonitor;
- protected volatile PublishRateLimiter brokerPublishRateLimiter = PublishRateLimiter.DISABLED_RATE_LIMITER;
+ protected final PublishRateLimiter brokerPublishRateLimiter;
protected volatile DispatchRateLimiter brokerDispatchRateLimiter = null;
private DistributedIdGenerator producerNameGenerator;
@@ -295,6 +293,7 @@ public class BrokerService implements Closeable {
public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws Exception {
this.pulsar = pulsar;
+ this.brokerPublishRateLimiter = new PublishRateLimiterImpl(pulsar.getMonotonicSnapshotClock());
this.preciseTopicPublishRateLimitingEnable =
pulsar.getConfiguration().isPreciseTopicPublishRateLimiterEnable();
this.managedLedgerFactory = pulsar.getManagedLedgerFactory();
@@ -358,10 +357,6 @@ public class BrokerService implements Closeable {
.name("pulsar-consumed-ledgers-monitor")
.numThreads(1)
.build();
- this.topicPublishRateLimiterMonitor =
- new PublishRateLimiterMonitor("pulsar-topic-publish-rate-limiter-monitor");
- this.brokerPublishRateLimiterMonitor =
- new PublishRateLimiterMonitor("pulsar-broker-publish-rate-limiter-monitor");
this.backlogQuotaManager = new BacklogQuotaManager(pulsar);
this.backlogQuotaChecker = OrderedScheduler.newSchedulerBuilder()
.name("pulsar-backlog-quota-checker")
@@ -669,87 +664,6 @@ public class BrokerService implements Closeable {
}
- /**
- * Schedules and monitors publish-throttling for all owned topics that has publish-throttling configured. It also
- * disables and shutdowns publish-rate-limiter monitor task if broker disables it.
- */
- public void setupTopicPublishRateLimiterMonitor() {
- // set topic PublishRateLimiterMonitor
- long topicTickTimeMs = pulsar().getConfiguration().getTopicPublisherThrottlingTickTimeMillis();
- if (topicTickTimeMs > 0) {
- topicPublishRateLimiterMonitor.startOrUpdate(topicTickTimeMs,
- this::checkTopicPublishThrottlingRate, this::refreshTopicPublishRate);
- } else {
- // disable publish-throttling for all topics
- topicPublishRateLimiterMonitor.stop();
- }
- }
-
- /**
- * Schedules and monitors publish-throttling for broker that has publish-throttling configured. It also
- * disables and shutdowns publish-rate-limiter monitor for broker task if broker disables it.
- */
- public void setupBrokerPublishRateLimiterMonitor() {
- // set broker PublishRateLimiterMonitor
- long brokerTickTimeMs = pulsar().getConfiguration().getBrokerPublisherThrottlingTickTimeMillis();
- if (brokerTickTimeMs > 0) {
- brokerPublishRateLimiterMonitor.startOrUpdate(brokerTickTimeMs,
- this::checkBrokerPublishThrottlingRate, this::refreshBrokerPublishRate);
- } else {
- // disable publish-throttling for broker.
- brokerPublishRateLimiterMonitor.stop();
- }
- }
-
- protected static class PublishRateLimiterMonitor {
- private final String name;
- private ScheduledExecutorService scheduler = null;
- private long tickTimeMs = 0;
- private Runnable refreshTask;
-
- public PublishRateLimiterMonitor(String name) {
- this.name = name;
- }
-
- synchronized void startOrUpdate(long tickTimeMs, Runnable checkTask, Runnable refreshTask) {
- if (this.scheduler != null) {
- // we have old task running.
- if (this.tickTimeMs == tickTimeMs) {
- // tick time not changed.
- return;
- }
- stop();
- }
- //start monitor.
- scheduler = OrderedScheduler.newSchedulerBuilder()
- .name(name)
- .numThreads(1)
- .build();
- // schedule task that sums up publish-rate across all cnx on a topic ,
- // and check the rate limit exceeded or not.
- scheduler.scheduleAtFixedRate(checkTask, tickTimeMs, tickTimeMs, TimeUnit.MILLISECONDS);
- // schedule task that refreshes rate-limiting bucket
- scheduler.scheduleAtFixedRate(refreshTask, 1, 1, TimeUnit.SECONDS);
- this.tickTimeMs = tickTimeMs;
- this.refreshTask = refreshTask;
- }
-
- synchronized void stop() {
- if (this.scheduler != null) {
- this.scheduler.shutdownNow();
- // make sure topics are not being throttled
- refreshTask.run();
- this.scheduler = null;
- this.tickTimeMs = 0;
- }
- }
-
- @VisibleForTesting
- protected synchronized long getTickTimeMs() {
- return tickTimeMs;
- }
- }
-
public void close() throws IOException {
try {
closeAsync().get();
@@ -881,8 +795,6 @@ public class BrokerService implements Closeable {
consumedLedgersMonitor,
backlogQuotaChecker,
topicOrderedExecutor,
- topicPublishRateLimiterMonitor.scheduler,
- brokerPublishRateLimiterMonitor.scheduler,
deduplicationSnapshotMonitor)
.handle());
@@ -972,26 +884,21 @@ public class BrokerService implements Closeable {
Set<NamespaceBundle> serviceUnits =
pulsar.getNamespaceService() != null ? pulsar.getNamespaceService().getOwnedServiceUnits() : null;
if (serviceUnits != null) {
- try (RateLimiter rateLimiter = maxConcurrentUnload > 0 ? RateLimiter.builder()
- .scheduledExecutorService(pulsar.getExecutor())
- .rateTime(1).timeUnit(TimeUnit.SECONDS)
- .permits(maxConcurrentUnload).build() : null) {
- serviceUnits.forEach(su -> {
- if (su != null) {
- try {
- if (rateLimiter != null) {
- rateLimiter.acquire(1);
- }
- long timeout = pulsar.getConfiguration().getNamespaceBundleUnloadingTimeoutMs();
- pulsar.getNamespaceService().unloadNamespaceBundle(su, timeout, TimeUnit.MILLISECONDS,
- closeWithoutWaitingClientDisconnect).get(timeout, TimeUnit.MILLISECONDS);
- } catch (Exception e) {
- log.warn("Failed to unload namespace bundle {}", su, e);
+ RateLimiter rateLimiter = maxConcurrentUnload > 0 ? RateLimiter.create(maxConcurrentUnload) : null;
+ serviceUnits.forEach(su -> {
+ if (su != null) {
+ try {
+ if (rateLimiter != null) {
+ rateLimiter.acquire(1);
}
+ long timeout = pulsar.getConfiguration().getNamespaceBundleUnloadingTimeoutMs();
+ pulsar.getNamespaceService().unloadNamespaceBundle(su, timeout, TimeUnit.MILLISECONDS,
+ closeWithoutWaitingClientDisconnect).get(timeout, TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
+ log.warn("Failed to unload namespace bundle {}", su, e);
}
- });
- }
-
+ }
+ });
double closeTopicsTimeSeconds =
TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - closeTopicsStartTime))
/ 1000.0;
@@ -2138,26 +2045,6 @@ public class BrokerService implements Closeable {
forEachTopic(Topic::checkInactiveSubscriptions);
}
- public void checkTopicPublishThrottlingRate() {
- forEachTopic(Topic::checkTopicPublishThrottlingRate);
- }
-
- private void refreshTopicPublishRate() {
- forEachTopic(Topic::resetTopicPublishCountAndEnableReadIfRequired);
- }
-
- public void checkBrokerPublishThrottlingRate() {
- brokerPublishRateLimiter.checkPublishRate();
- if (brokerPublishRateLimiter.isPublishRateExceeded()) {
- forEachTopic(topic -> ((AbstractTopic) topic).disableProducerRead());
- }
- }
-
- private void refreshBrokerPublishRate() {
- boolean doneReset = brokerPublishRateLimiter.resetPublishCount();
- forEachTopic(topic -> topic.resetBrokerPublishCountAndEnableReadIfRequired(doneReset));
- }
-
/**
* Iterates over all loaded topics in the broker.
*/
@@ -2730,12 +2617,6 @@ public class BrokerService implements Closeable {
updateReplicatorMessageDispatchRate();
});
- // add listener to notify broker publish-rate monitoring
- registerConfigurationListener("brokerPublisherThrottlingTickTimeMillis",
- (publisherThrottlingTickTimeMillis) -> {
- setupBrokerPublishRateLimiterMonitor();
- });
-
// add listener to update topic publish-rate dynamic config
registerConfigurationListener("maxPublishRatePerTopicInMessages",
maxPublishRatePerTopicInMessages -> updateMaxPublishRatePerTopicInMessages()
@@ -2764,13 +2645,6 @@ public class BrokerService implements Closeable {
registerConfigurationListener("dispatchThrottlingRateInByte",
(dispatchThrottlingRateInByte) ->
updateBrokerDispatchThrottlingMaxRate());
- // add listener to notify topic publish-rate monitoring
- if (!preciseTopicPublishRateLimitingEnable) {
- registerConfigurationListener("topicPublisherThrottlingTickTimeMillis",
- (publisherThrottlingTickTimeMillis) -> {
- setupTopicPublishRateLimiterMonitor();
- });
- }
// add listener to notify topic subscriptionTypesEnabled changed.
registerConfigurationListener("subscriptionTypesEnabled", this::updateBrokerSubscriptionTypesEnabled);
@@ -2850,29 +2724,11 @@ public class BrokerService implements Closeable {
private void updateBrokerPublisherThrottlingMaxRate() {
int currentMaxMessageRate = pulsar.getConfiguration().getBrokerPublisherThrottlingMaxMessageRate();
long currentMaxByteRate = pulsar.getConfiguration().getBrokerPublisherThrottlingMaxByteRate();
- int brokerTickMs = pulsar.getConfiguration().getBrokerPublisherThrottlingTickTimeMillis();
-
- // not enable
- if (brokerTickMs <= 0 || (currentMaxByteRate <= 0 && currentMaxMessageRate <= 0)) {
- if (brokerPublishRateLimiter != PublishRateLimiter.DISABLED_RATE_LIMITER) {
- refreshBrokerPublishRate();
- brokerPublishRateLimiter = PublishRateLimiter.DISABLED_RATE_LIMITER;
- }
- return;
- }
final PublishRate publishRate = new PublishRate(currentMaxMessageRate, currentMaxByteRate);
log.info("Update broker publish rate limiting {}", publishRate);
- // lazy init broker Publish-rateLimiting monitoring if not initialized yet
- this.setupBrokerPublishRateLimiterMonitor();
- if (brokerPublishRateLimiter == null
- || brokerPublishRateLimiter == PublishRateLimiter.DISABLED_RATE_LIMITER) {
- // create new rateLimiter if rate-limiter is disabled
- brokerPublishRateLimiter = new PublishRateLimiterImpl(publishRate);
- } else {
- brokerPublishRateLimiter.update(publishRate);
- }
+ brokerPublishRateLimiter.update(publishRate);
}
private void updateTopicMessageDispatchRate() {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PrecisePublishLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PrecisePublishLimiter.java
deleted file mode 100644
index ce14f6d7dd7..00000000000
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PrecisePublishLimiter.java
+++ /dev/null
@@ -1,162 +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.service;
-
-import java.util.concurrent.ScheduledExecutorService;
-import org.apache.pulsar.common.policies.data.Policies;
-import org.apache.pulsar.common.policies.data.PublishRate;
-import org.apache.pulsar.common.util.RateLimitFunction;
-import org.apache.pulsar.common.util.RateLimiter;
-
-public class PrecisePublishLimiter implements PublishRateLimiter {
- protected volatile int publishMaxMessageRate = 0;
- protected volatile long publishMaxByteRate = 0;
- // precise mode for publish rate limiter
- private volatile RateLimiter topicPublishRateLimiterOnMessage;
- private volatile RateLimiter topicPublishRateLimiterOnByte;
- private final RateLimitFunction rateLimitFunction;
- private final ScheduledExecutorService scheduledExecutorService;
-
- public PrecisePublishLimiter(Policies policies, String clusterName, RateLimitFunction rateLimitFunction) {
- this.rateLimitFunction = rateLimitFunction;
- update(policies, clusterName);
- this.scheduledExecutorService = null;
- }
-
- public PrecisePublishLimiter(PublishRate publishRate, RateLimitFunction rateLimitFunction) {
- this(publishRate, rateLimitFunction, null);
- }
-
- public PrecisePublishLimiter(PublishRate publishRate, RateLimitFunction rateLimitFunction,
- ScheduledExecutorService scheduledExecutorService) {
- this.rateLimitFunction = rateLimitFunction;
- update(publishRate);
- this.scheduledExecutorService = scheduledExecutorService;
- }
-
- @Override
- public void checkPublishRate() {
- // No-op
- }
-
- @Override
- public void incrementPublishCount(int numOfMessages, long msgSizeInBytes) {
- // No-op
- }
-
- @Override
- public boolean resetPublishCount() {
- return true;
- }
-
- @Override
- public boolean isPublishRateExceeded() {
- return false;
- }
-
- // If all rate limiters are not exceeded, re-enable auto read from socket.
- private void tryReleaseConnectionThrottle() {
- RateLimiter currentTopicPublishRateLimiterOnMessage = topicPublishRateLimiterOnMessage;
- RateLimiter currentTopicPublishRateLimiterOnByte = topicPublishRateLimiterOnByte;
- if ((currentTopicPublishRateLimiterOnMessage != null
- && currentTopicPublishRateLimiterOnMessage.getAvailablePermits() <= 0)
- || (currentTopicPublishRateLimiterOnByte != null
- && currentTopicPublishRateLimiterOnByte.getAvailablePermits() <= 0)) {
- return;
- }
- this.rateLimitFunction.apply();
- }
-
- @Override
- public void update(Policies policies, String clusterName) {
- final PublishRate maxPublishRate = policies.publishMaxMessageRate != null
- ? policies.publishMaxMessageRate.get(clusterName)
- : null;
- this.update(maxPublishRate);
- }
-
- public void update(PublishRate maxPublishRate) {
- replaceLimiters(() -> {
- if (maxPublishRate != null
- && (maxPublishRate.publishThrottlingRateInMsg > 0
- || maxPublishRate.publishThrottlingRateInByte > 0)) {
- this.publishMaxMessageRate = Math.max(maxPublishRate.publishThrottlingRateInMsg, 0);
- this.publishMaxByteRate = Math.max(maxPublishRate.publishThrottlingRateInByte, 0);
- if (this.publishMaxMessageRate > 0) {
- topicPublishRateLimiterOnMessage =
- RateLimiter.builder()
- .scheduledExecutorService(scheduledExecutorService)
- .permits(publishMaxMessageRate)
- .rateLimitFunction(this::tryReleaseConnectionThrottle)
- .isDispatchOrPrecisePublishRateLimiter(true)
- .build();
- }
- if (this.publishMaxByteRate > 0) {
- topicPublishRateLimiterOnByte = RateLimiter.builder()
- .scheduledExecutorService(scheduledExecutorService)
- .permits(publishMaxByteRate)
- .rateLimitFunction(this::tryReleaseConnectionThrottle)
- .isDispatchOrPrecisePublishRateLimiter(true)
- .build();
- }
- } else {
- this.publishMaxMessageRate = 0;
- this.publishMaxByteRate = 0;
- }
- });
- }
-
- @Override
- public boolean tryAcquire(int numbers, long bytes) {
- RateLimiter currentTopicPublishRateLimiterOnMessage = topicPublishRateLimiterOnMessage;
- RateLimiter currentTopicPublishRateLimiterOnByte = topicPublishRateLimiterOnByte;
- return (currentTopicPublishRateLimiterOnMessage == null
- || currentTopicPublishRateLimiterOnMessage.tryAcquire(numbers))
- && (currentTopicPublishRateLimiterOnByte == null
- || currentTopicPublishRateLimiterOnByte.tryAcquire(bytes));
- }
-
- @Override
- public void close() {
- rateLimitFunction.apply();
- replaceLimiters(null);
- }
-
- private void replaceLimiters(Runnable updater) {
- RateLimiter previousTopicPublishRateLimiterOnMessage = topicPublishRateLimiterOnMessage;
- topicPublishRateLimiterOnMessage = null;
- RateLimiter previousTopicPublishRateLimiterOnByte = topicPublishRateLimiterOnByte;
- topicPublishRateLimiterOnByte = null;
- try {
- if (updater != null) {
- updater.run();
- }
- } finally {
- // Close previous limiters to prevent resource leakages.
- // Delay closing of previous limiters after new ones are in place so that updating the limiter
- // doesn't cause unavailability.
- if (previousTopicPublishRateLimiterOnMessage != null) {
- previousTopicPublishRateLimiterOnMessage.close();
- }
- if (previousTopicPublishRateLimiterOnByte != null) {
- previousTopicPublishRateLimiterOnByte.close();
- }
- }
- }
-}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
index b0b4fe98b0c..7e4459505a5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
@@ -321,7 +321,7 @@ public class Producer {
// barrier
pendingPublishAcksUpdater.lazySet(this, pendingPublishAcks + 1);
// increment publish-count
- this.getTopic().incrementPublishCount(batchSize, msgSize);
+ this.getTopic().incrementPublishCount(this, batchSize, msgSize);
}
private void publishOperationCompleted() {
@@ -853,4 +853,27 @@ public class Producer {
private static final Logger log = LoggerFactory.getLogger(Producer.class);
+ /**
+ * This method increments a counter that is used to control the throttling of a connection.
+ * The connection's read operations are paused when the counter's value is greater than 0, indicating that
+ * throttling is in effect.
+ * It's important to note that after calling this method, it is the caller's responsibility to ensure that the
+ * counter is decremented by calling the {@link #decrementThrottleCount()} method when throttling is no longer
+ * needed on the connection.
+ */
+ public void incrementThrottleCount() {
+ cnx.incrementThrottleCount();
+ }
+
+ /**
+ * This method decrements a counter that is used to control the throttling of a connection.
+ * The connection's read operations are resumed when the counter's value is 0, indicating that
+ * throttling is no longer in effect.
+ * It's important to note that before calling this method, the caller should have previously
+ * incremented the counter by calling the {@link #incrementThrottleCount()} method when throttling
+ * was needed on the connection.
+ */
+ public void decrementThrottleCount() {
+ cnx.decrementThrottleCount();
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiter.java
index fde2e7cb56d..cdc2f448abd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiter.java
@@ -21,35 +21,19 @@ package org.apache.pulsar.broker.service;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.PublishRate;
-public interface PublishRateLimiter extends AutoCloseable {
-
- PublishRateLimiter DISABLED_RATE_LIMITER = PublishRateLimiterDisable.DISABLED_RATE_LIMITER;
+public interface PublishRateLimiter {
/**
- * checks and update state of current publish and marks if it has exceeded the rate-limiting threshold.
- */
- void checkPublishRate();
-
- /**
- * increments current publish count.
+ * Consumes publishing quota and handles throttling.
+ * <p>
+ * The rate limiter implementation calls {@link Producer#incrementThrottleCount()} to indicate
+ * that the producer should be throttled. The rate limiter must schedule a call to
+ * {@link Producer#decrementThrottleCount()} after a throttling period that it calculates.
*
- * @param numOfMessages
- * @param msgSizeInBytes
- */
- void incrementPublishCount(int numOfMessages, long msgSizeInBytes);
-
- /**
- * reset current publish count.
- *
- * @return
- */
- boolean resetPublishCount();
-
- /**
- * returns true if current publish has reached the rate-limiting threshold.
- * @return
+ * @param numOfMessages number of messages to publish
+ * @param msgSizeInBytes size of messages in bytes to publish
*/
- boolean isPublishRateExceeded();
+ void handlePublishThrottling(Producer producer, int numOfMessages, long msgSizeInBytes);
/**
* updates rate-limiting threshold based on policies.
@@ -63,17 +47,4 @@ public interface PublishRateLimiter extends AutoCloseable {
* @param maxPublishRate
*/
void update(PublishRate maxPublishRate);
-
- /**
- * try to acquire permit.
- *
- * @param numbers
- * @param bytes
- */
- boolean tryAcquire(int numbers, long bytes);
-
- /**
- * Close the limiter.
- */
- void close();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterDisable.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterDisable.java
deleted file mode 100644
index fdc13ed8f4e..00000000000
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterDisable.java
+++ /dev/null
@@ -1,69 +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.service;
-
-import org.apache.pulsar.common.policies.data.Policies;
-import org.apache.pulsar.common.policies.data.PublishRate;
-
-public class PublishRateLimiterDisable implements PublishRateLimiter {
-
- public static final PublishRateLimiterDisable DISABLED_RATE_LIMITER = new PublishRateLimiterDisable();
-
- @Override
- public void checkPublishRate() {
- // No-op
- }
-
- @Override
- public void incrementPublishCount(int numOfMessages, long msgSizeInBytes) {
- // No-op
- }
-
- @Override
- public boolean resetPublishCount() {
- // No-op
- return false;
- }
-
- @Override
- public boolean isPublishRateExceeded() {
- return false;
- }
-
- @Override
- public void update(Policies policies, String clusterName) {
- // No-op
- }
-
- @Override
- public void update(PublishRate maxPublishRate) {
- // No-op
- }
-
- @Override
- public boolean tryAcquire(int numbers, long bytes) {
- // Always allow
- return true;
- }
-
- @Override
- public void close() {
- // No-op
- }
-}
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 1947acd87fc..8255d9b6931 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
@@ -16,70 +16,135 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.pulsar.broker.service;
-import java.util.concurrent.atomic.LongAdder;
+import com.google.common.annotations.VisibleForTesting;
+import io.netty.channel.EventLoopGroup;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.pulsar.broker.qos.AsyncTokenBucket;
+import org.apache.pulsar.broker.qos.MonotonicSnapshotClock;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.PublishRate;
+import org.jctools.queues.MessagePassingQueue;
+import org.jctools.queues.MpscUnboundedArrayQueue;
public class PublishRateLimiterImpl implements PublishRateLimiter {
- protected volatile int publishMaxMessageRate = 0;
- protected volatile long publishMaxByteRate = 0;
- protected volatile boolean publishThrottlingEnabled = false;
- protected volatile boolean publishRateExceeded = false;
- protected volatile LongAdder currentPublishMsgCount = new LongAdder();
- protected volatile LongAdder currentPublishByteCount = new LongAdder();
-
- public PublishRateLimiterImpl(Policies policies, String clusterName) {
- update(policies, clusterName);
- }
+ private volatile AsyncTokenBucket tokenBucketOnMessage;
+ private volatile AsyncTokenBucket tokenBucketOnByte;
+ private final MonotonicSnapshotClock monotonicSnapshotClock;
- public PublishRateLimiterImpl(PublishRate maxPublishRate) {
- update(maxPublishRate);
+ 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;
}
+ /**
+ * {@inheritDoc}
+ */
@Override
- public void checkPublishRate() {
- if (this.publishThrottlingEnabled && !publishRateExceeded) {
- if (this.publishMaxByteRate > 0) {
- long currentPublishByteRate = this.currentPublishByteCount.sum();
- if (currentPublishByteRate > this.publishMaxByteRate) {
- publishRateExceeded = true;
- return;
- }
- }
-
- if (this.publishMaxMessageRate > 0) {
- long currentPublishMsgRate = this.currentPublishMsgCount.sum();
- if (currentPublishMsgRate > this.publishMaxMessageRate) {
- publishRateExceeded = true;
- }
- }
+ public void handlePublishThrottling(Producer producer, int numOfMessages,
+ long msgSizeInBytes) {
+ boolean shouldThrottle = false;
+ AsyncTokenBucket currentTokenBucketOnMessage = tokenBucketOnMessage;
+ if (currentTokenBucketOnMessage != null) {
+ // consume tokens from the token bucket for messages
+ // we should throttle if it returns false since the token bucket is empty in that case
+ shouldThrottle = !currentTokenBucketOnMessage.consumeTokensAndCheckIfContainsTokens(numOfMessages);
+ }
+ AsyncTokenBucket currentTokenBucketOnByte = tokenBucketOnByte;
+ if (currentTokenBucketOnByte != null) {
+ // consume tokens from the token bucket for bytes
+ // we should throttle if it returns false since the token bucket is empty in that case
+ shouldThrottle |= !currentTokenBucketOnByte.consumeTokensAndCheckIfContainsTokens(msgSizeInBytes);
+ }
+ if (shouldThrottle) {
+ // throttle the producer by incrementing the throttle count
+ producer.incrementThrottleCount();
+ // schedule decrementing the throttle count to possibly unthrottle the producer after the
+ // throttling period
+ scheduleDecrementThrottleCount(producer);
}
}
- @Override
- public void incrementPublishCount(int numOfMessages, long msgSizeInBytes) {
- if (this.publishThrottlingEnabled) {
- this.currentPublishMsgCount.add(numOfMessages);
- this.currentPublishByteCount.add(msgSizeInBytes);
+ private void scheduleDecrementThrottleCount(Producer producer) {
+ // add the producer to the queue of producers to be unthrottled
+ unthrottlingQueue.offer(producer);
+ // schedule unthrottling when the throttling count is incremented to 1
+ // this is to avoid scheduling unthrottling multiple times for concurrent producers
+ if (throttledProducersCount.incrementAndGet() == 1) {
+ EventLoopGroup executor = producer.getCnx().getBrokerService().executor();
+ scheduleUnthrottling(executor, calculateThrottlingDurationNanos());
}
}
- @Override
- public boolean resetPublishCount() {
- if (this.publishThrottlingEnabled) {
- this.currentPublishMsgCount.reset();
- this.currentPublishByteCount.reset();
- this.publishRateExceeded = false;
- return true;
+ /**
+ * Schedules the unthrottling operation after a throttling period.
+ *
+ * This method will usually be called only once at a time. However, in a multi-threaded environment,
+ * it's possible for concurrent threads to call this method simultaneously. This is acceptable and does not
+ * disrupt the functionality, as the method is designed to handle such scenarios gracefully.
+ *
+ * The solution avoids using locks and this nonblocking approach requires allowing concurrent calls to this method.
+ * The implementation intends to prevent skipping of scheduling as a result of a race condition, which could
+ * result in a producer never being unthrottled.
+ *
+ * The solution for skipping of scheduling is to allow 2 threads to schedule unthrottling when the throttling
+ * count is exactly 1 when unthrottleQueuedProducers checks whether there's a need to reschedule. There might
+ * be another thread that added it and also scheduled unthrottling. This is acceptable and intended for resolving
+ * the race condition.
+ *
+ * @param executor The executor service used to schedule the unthrottling operation.
+ * @param delayNanos
+ */
+ private void scheduleUnthrottling(ScheduledExecutorService executor, long delayNanos) {
+ executor.schedule(() -> this.unthrottleQueuedProducers(executor), delayNanos,
+ TimeUnit.NANOSECONDS);
+ }
+
+ private long calculateThrottlingDurationNanos() {
+ AsyncTokenBucket currentTokenBucketOnMessage = tokenBucketOnMessage;
+ long throttlingDurationNanos = 0L;
+ if (currentTokenBucketOnMessage != null) {
+ throttlingDurationNanos = currentTokenBucketOnMessage.calculateThrottlingDuration();
}
- return false;
+ AsyncTokenBucket currentTokenBucketOnByte = tokenBucketOnByte;
+ if (currentTokenBucketOnByte != null) {
+ throttlingDurationNanos = Math.max(throttlingDurationNanos,
+ currentTokenBucketOnByte.calculateThrottlingDuration());
+ }
+ return throttlingDurationNanos;
}
- @Override
- public boolean isPublishRateExceeded() {
- return publishRateExceeded;
+ private void unthrottleQueuedProducers(ScheduledExecutorService executor) {
+ if (!processingQueuedProducers.compareAndSet(false, true)) {
+ // another thread is already processing unthrottling
+ return;
+ }
+ try {
+ Producer producer;
+ long throttlingDuration = 0L;
+ // unthrottle as many producers as possible while there are token available
+ while ((throttlingDuration = calculateThrottlingDurationNanos()) == 0L
+ && (producer = unthrottlingQueue.poll()) != null) {
+ producer.decrementThrottleCount();
+ throttledProducersCount.decrementAndGet();
+ }
+ // if there are still producers to be unthrottled, schedule unthrottling again
+ // after another throttling period
+ if (throttledProducersCount.get() > 0) {
+ scheduleUnthrottling(executor, throttlingDuration);
+ }
+ } finally {
+ processingQueuedProducers.set(false);
+ }
}
@Override
@@ -91,26 +156,36 @@ public class PublishRateLimiterImpl implements PublishRateLimiter {
}
public void update(PublishRate maxPublishRate) {
- if (maxPublishRate != null
- && (maxPublishRate.publishThrottlingRateInMsg > 0 || maxPublishRate.publishThrottlingRateInByte > 0)) {
- this.publishThrottlingEnabled = true;
- this.publishMaxMessageRate = Math.max(maxPublishRate.publishThrottlingRateInMsg, 0);
- this.publishMaxByteRate = Math.max(maxPublishRate.publishThrottlingRateInByte, 0);
+ if (maxPublishRate != null) {
+ updateTokenBuckets(maxPublishRate.publishThrottlingRateInMsg, maxPublishRate.publishThrottlingRateInByte);
} else {
- this.publishMaxMessageRate = 0;
- this.publishMaxByteRate = 0;
- this.publishThrottlingEnabled = false;
+ tokenBucketOnMessage = null;
+ tokenBucketOnByte = null;
}
- resetPublishCount();
}
- @Override
- public boolean tryAcquire(int numbers, long bytes) {
- return false;
+ protected void updateTokenBuckets(long publishThrottlingRateInMsg, long publishThrottlingRateInByte) {
+ if (publishThrottlingRateInMsg > 0) {
+ tokenBucketOnMessage =
+ AsyncTokenBucket.builder().rate(publishThrottlingRateInMsg).clock(monotonicSnapshotClock).build();
+ } else {
+ tokenBucketOnMessage = null;
+ }
+ if (publishThrottlingRateInByte > 0) {
+ tokenBucketOnByte =
+ AsyncTokenBucket.builder().rate(publishThrottlingRateInByte).clock(monotonicSnapshotClock).build();
+ } else {
+ tokenBucketOnByte = null;
+ }
}
- @Override
- public void close() {
- // no-op
+ @VisibleForTesting
+ public AsyncTokenBucket getTokenBucketOnMessage() {
+ return tokenBucketOnMessage;
+ }
+
+ @VisibleForTesting
+ public AsyncTokenBucket getTokenBucketOnByte() {
+ return tokenBucketOnByte;
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index 2466f86720f..2baa55b80e7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -41,7 +41,6 @@ import io.netty.handler.ssl.SslHandler;
import io.netty.util.concurrent.FastThreadLocal;
import io.netty.util.concurrent.Promise;
import io.netty.util.concurrent.ScheduledFuture;
-import io.prometheus.client.Gauge;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
@@ -71,8 +70,6 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
-import org.apache.commons.lang3.mutable.MutableInt;
-import org.apache.commons.lang3.mutable.MutableLong;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.TransactionMetadataStoreService;
@@ -228,11 +225,8 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
private final int maxMessageSize;
private boolean preciseDispatcherFlowControl;
- private boolean preciseTopicPublishRateLimitingEnable;
private boolean encryptionRequireOnProducer;
- // Flag to manage throttling-rate by atomically enable/disable read-channel.
- private volatile boolean autoReadDisabledRateLimiting = false;
private FeatureFlags features;
private PulsarCommandSender commandSender;
@@ -241,23 +235,52 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
private static final KeySharedMeta emptyKeySharedMeta = new KeySharedMeta()
.setKeySharedMode(KeySharedMode.AUTO_SPLIT);
- // Flag to manage throttling-publish-buffer by atomically enable/disable read-channel.
- private boolean autoReadDisabledPublishBufferLimiting = false;
private final long maxPendingBytesPerThread;
private final long resumeThresholdPendingBytesPerThread;
private final long connectionLivenessCheckTimeoutMillis;
- // Number of bytes pending to be published from a single specific IO thread.
- private static final FastThreadLocal<MutableLong> pendingBytesPerThread = new FastThreadLocal<MutableLong>() {
- @Override
- protected MutableLong initialValue() throws Exception {
- return new MutableLong();
+ // Tracks and limits number of bytes pending to be published from a single specific IO thread.
+ static final class PendingBytesPerThreadTracker {
+ private static final FastThreadLocal<PendingBytesPerThreadTracker> pendingBytesPerThread =
+ new FastThreadLocal<>() {
+ @Override
+ protected PendingBytesPerThreadTracker initialValue() throws Exception {
+ return new PendingBytesPerThreadTracker();
+ }
+ };
+
+ private long pendingBytes;
+ private boolean limitExceeded;
+
+ public static PendingBytesPerThreadTracker getInstance() {
+ return pendingBytesPerThread.get();
}
- };
+
+ public void incrementPublishBytes(long bytes, long maxPendingBytesPerThread) {
+ pendingBytes += bytes;
+ // when the limit is exceeded we throttle all connections that are sharing the same thread
+ if (maxPendingBytesPerThread > 0 && pendingBytes > maxPendingBytesPerThread
+ && !limitExceeded) {
+ limitExceeded = true;
+ cnxsPerThread.get().forEach(cnx -> cnx.throttleTracker.setPublishBufferLimiting(true));
+ }
+ }
+
+ public void decrementPublishBytes(long bytes, long resumeThresholdPendingBytesPerThread) {
+ pendingBytes -= bytes;
+ // when the limit has been exceeded, and we are below the resume threshold
+ // we resume all connections sharing the same thread
+ if (limitExceeded && pendingBytes <= resumeThresholdPendingBytesPerThread) {
+ limitExceeded = false;
+ cnxsPerThread.get().forEach(cnx -> cnx.throttleTracker.setPublishBufferLimiting(false));
+ }
+ }
+ }
+
// A set of connections tied to the current thread
- private static final FastThreadLocal<Set<ServerCnx>> cnxsPerThread = new FastThreadLocal<Set<ServerCnx>>() {
+ private static final FastThreadLocal<Set<ServerCnx>> cnxsPerThread = new FastThreadLocal<>() {
@Override
protected Set<ServerCnx> initialValue() throws Exception {
return Collections.newSetFromMap(new IdentityHashMap<>());
@@ -268,6 +291,9 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
Start, Connected, Failed, Connecting
}
+ private final ServerCnxThrottleTracker throttleTracker = new ServerCnxThrottleTracker(this);
+
+
public ServerCnx(PulsarService pulsar) {
this(pulsar, null);
}
@@ -302,7 +328,6 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
this.maxPendingSendRequests = conf.getMaxPendingPublishRequestsPerConnection();
this.resumeReadsThreshold = maxPendingSendRequests / 2;
this.preciseDispatcherFlowControl = conf.isPreciseDispatcherFlowControl();
- this.preciseTopicPublishRateLimitingEnable = conf.isPreciseTopicPublishRateLimiterEnable();
this.encryptionRequireOnProducer = conf.isEncryptionRequireOnProducer();
// Assign a portion of max-pending bytes to each IO thread
this.maxPendingBytesPerThread = conf.getMaxMessagePublishBufferSizeInMB() * 1024L * 1024L
@@ -1792,7 +1817,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
}
}
- startSendOperation(producer, headersAndPayload.readableBytes(), send.getNumMessages());
+ increasePendingSendRequestsAndPublishBytes(headersAndPayload.readableBytes());
if (send.hasTxnidMostBits() && send.hasTxnidLeastBits()) {
TxnID txnID = new TxnID(send.getTxnidMostBits(), send.getTxnidLeastBits());
@@ -3167,64 +3192,20 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
return ctx.channel().isWritable();
}
- private static final Gauge throttledConnections = Gauge.build()
- .name("pulsar_broker_throttled_connections")
- .help("Counter of connections throttled because of per-connection limit")
- .register();
-
- private static final Gauge throttledConnectionsGlobal = Gauge.build()
- .name("pulsar_broker_throttled_connections_global_limit")
- .help("Counter of connections throttled because of per-connection limit")
- .register();
-
- public void startSendOperation(Producer producer, int msgSize, int numMessages) {
- boolean isPublishRateExceeded = false;
- if (preciseTopicPublishRateLimitingEnable) {
- boolean isPreciseTopicPublishRateExceeded =
- producer.getTopic().isTopicPublishRateExceeded(numMessages, msgSize);
- if (isPreciseTopicPublishRateExceeded) {
- producer.getTopic().disableCnxAutoRead();
- return;
- }
- isPublishRateExceeded = producer.getTopic().isBrokerPublishRateExceeded();
- } else {
- if (producer.getTopic().isResourceGroupRateLimitingEnabled()) {
- final boolean resourceGroupPublishRateExceeded =
- producer.getTopic().isResourceGroupPublishRateExceeded(numMessages, msgSize);
- if (resourceGroupPublishRateExceeded) {
- producer.getTopic().disableCnxAutoRead();
- return;
- }
- }
- isPublishRateExceeded = producer.getTopic().isPublishRateExceeded();
- }
-
- if (++pendingSendRequest == maxPendingSendRequests || isPublishRateExceeded) {
- // When the quota of pending send requests is reached, stop reading from socket to cause backpressure on
- // client connection, possibly shared between multiple producers
- disableCnxAutoRead();
- autoReadDisabledRateLimiting = isPublishRateExceeded;
- throttledConnections.inc();
- }
-
- if (pendingBytesPerThread.get().addAndGet(msgSize) >= maxPendingBytesPerThread
- && !autoReadDisabledPublishBufferLimiting
- && maxPendingBytesPerThread > 0) {
- // Disable reading from all the connections associated with this thread
- MutableInt pausedConnections = new MutableInt();
- cnxsPerThread.get().forEach(cnx -> {
- if (cnx.hasProducers() && !cnx.autoReadDisabledPublishBufferLimiting) {
- cnx.disableCnxAutoRead();
- cnx.autoReadDisabledPublishBufferLimiting = true;
- pausedConnections.increment();
- }
- });
-
- getBrokerService().pausedConnections(pausedConnections.intValue());
+ // handle throttling based on pending send requests in the same connection
+ // or the pending publish bytes
+ private void increasePendingSendRequestsAndPublishBytes(int msgSize) {
+ if (++pendingSendRequest == maxPendingSendRequests) {
+ throttleTracker.setPendingSendRequestsExceeded(true);
}
+ PendingBytesPerThreadTracker.getInstance().incrementPublishBytes(msgSize, maxPendingBytesPerThread);
}
- private void recordRateLimitMetrics(ConcurrentLongHashMap<CompletableFuture<Producer>> producers) {
+
+ /**
+ * Increase the throttling metric for the topic when a producer is throttled.
+ */
+ void increasePublishLimitedTimesForTopics() {
producers.forEach((key, producerFuture) -> {
if (producerFuture != null && producerFuture.isDone()) {
Producer p = producerFuture.getNow(null);
@@ -3237,65 +3218,17 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
@Override
public void completedSendOperation(boolean isNonPersistentTopic, int msgSize) {
- if (pendingBytesPerThread.get().addAndGet(-msgSize) < resumeThresholdPendingBytesPerThread
- && autoReadDisabledPublishBufferLimiting) {
- // Re-enable reading on all the blocked connections
- MutableInt resumedConnections = new MutableInt();
- cnxsPerThread.get().forEach(cnx -> {
- if (cnx.autoReadDisabledPublishBufferLimiting) {
- cnx.autoReadDisabledPublishBufferLimiting = false;
- cnx.enableCnxAutoRead();
- resumedConnections.increment();
- }
- });
-
- getBrokerService().resumedConnections(resumedConnections.intValue());
- }
+ PendingBytesPerThreadTracker.getInstance().decrementPublishBytes(msgSize, resumeThresholdPendingBytesPerThread);
if (--pendingSendRequest == resumeReadsThreshold) {
- enableCnxAutoRead();
+ throttleTracker.setPendingSendRequestsExceeded(false);
}
+
if (isNonPersistentTopic) {
nonPersistentPendingMessages--;
}
}
- @Override
- public void enableCnxAutoRead() {
- // we can add check (&& pendingSendRequest < MaxPendingSendRequests) here but then it requires
- // pendingSendRequest to be volatile and it can be expensive while writing. also this will be called on if
- // throttling is enable on the topic. so, avoid pendingSendRequest check will be fine.
- if (ctx != null && !ctx.channel().config().isAutoRead()
- && !autoReadDisabledRateLimiting && !autoReadDisabledPublishBufferLimiting) {
- // Resume reading from socket if pending-request is not reached to threshold
- ctx.channel().config().setAutoRead(true);
- throttledConnections.dec();
- }
- }
-
- @Override
- public void disableCnxAutoRead() {
- if (ctx != null && ctx.channel().config().isAutoRead()) {
- ctx.channel().config().setAutoRead(false);
- recordRateLimitMetrics(producers);
- }
- }
-
- @Override
- public void cancelPublishRateLimiting() {
- if (autoReadDisabledRateLimiting) {
- autoReadDisabledRateLimiting = false;
- }
- }
-
- @Override
- public void cancelPublishBufferLimiting() {
- if (autoReadDisabledPublishBufferLimiting) {
- autoReadDisabledPublishBufferLimiting = false;
- throttledConnectionsGlobal.dec();
- }
- }
-
private <T> ServerError getErrorCode(CompletableFuture<T> future) {
return getErrorCodeWithErrorLog(future, false, null);
}
@@ -3446,11 +3379,6 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
return proxyVersion;
}
- @VisibleForTesting
- void setAutoReadDisabledRateLimiting(boolean isLimiting) {
- this.autoReadDisabledRateLimiting = isLimiting;
- }
-
@Override
public boolean isPreciseDispatcherFlowControl() {
return preciseDispatcherFlowControl;
@@ -3615,4 +3543,20 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
protected void setAuthRole(String authRole) {
this.authRole = authRole;
}
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void incrementThrottleCount() {
+ throttleTracker.incrementThrottleCount();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void decrementThrottleCount() {
+ throttleTracker.decrementThrottleCount();
+ }
}
\ No newline at end of file
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java
new file mode 100644
index 00000000000..f223d6eee37
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java
@@ -0,0 +1,146 @@
+/*
+ * 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 io.prometheus.client.Gauge;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Tracks the state of throttling for a connection. The throttling happens by pausing reads by setting
+ * Netty {@link io.netty.channel.ChannelConfig#setAutoRead(boolean)} to false for the channel (connection).
+ * <p>
+ * There can be multiple rate limiters that can throttle a connection. Each rate limiter will independently
+ * call the {@link #incrementThrottleCount()} and {@link #decrementThrottleCount()} methods to signal that the
+ * connection should be throttled or not. The connection will be throttled if the counter is greater than 0.
+ * <p>
+ * Besides the rate limiters, the connection can also be throttled if the number of pending publish requests exceeds
+ * a configured threshold. This throttling is toggled with the {@link #setPendingSendRequestsExceeded} method.
+ * There's also per-thread memory limits which could throttle the connection. This throttling is toggled with the
+ * {@link #setPublishBufferLimiting} method. Internally, these two methods will call the
+ * {@link #incrementThrottleCount()} and {@link #decrementThrottleCount()} methods when the state changes.
+ */
+@Slf4j
+final class ServerCnxThrottleTracker {
+ private static final Gauge throttledConnections = Gauge.build()
+ .name("pulsar_broker_throttled_connections")
+ .help("Counter of connections throttled because of per-connection limit")
+ .register();
+
+ private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(
+ ServerCnxThrottleTracker.class, "throttleCount");
+
+ private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
+ PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(
+ ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
+ private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> PUBLISH_BUFFER_LIMITING_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(
+ ServerCnxThrottleTracker.class, "publishBufferLimiting");
+ private final ServerCnx serverCnx;
+ private volatile int throttleCount;
+ private volatile int pendingSendRequestsExceeded;
+ private volatile int publishBufferLimiting;
+
+ public ServerCnxThrottleTracker(ServerCnx serverCnx) {
+ this.serverCnx = serverCnx;
+
+ }
+
+ /**
+ * See {@link Producer#incrementThrottleCount()} for documentation.
+ */
+ public void incrementThrottleCount() {
+ int currentThrottleCount = THROTTLE_COUNT_UPDATER.incrementAndGet(this);
+ if (currentThrottleCount == 1) {
+ changeAutoRead(false);
+ }
+ }
+
+ /**
+ * See {@link Producer#decrementThrottleCount()} for documentation.
+ */
+ public void decrementThrottleCount() {
+ int currentThrottleCount = THROTTLE_COUNT_UPDATER.decrementAndGet(this);
+ if (currentThrottleCount == 0) {
+ changeAutoRead(true);
+ }
+ }
+
+ private void changeAutoRead(boolean autoRead) {
+ if (isChannelActive()) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Setting auto read to {}", serverCnx.ctx().channel(), autoRead);
+ }
+ // change the auto read flag on the channel
+ serverCnx.ctx().channel().config().setAutoRead(autoRead);
+ }
+ // update the metrics that track throttling
+ if (autoRead) {
+ serverCnx.getBrokerService().resumedConnections(1);
+ } else if (isChannelActive()) {
+ serverCnx.increasePublishLimitedTimesForTopics();
+ serverCnx.getBrokerService().pausedConnections(1);
+ }
+ }
+
+ private boolean isChannelActive() {
+ return serverCnx.isActive() && serverCnx.ctx() != null && serverCnx.ctx().channel().isActive();
+ }
+
+ public void setPublishBufferLimiting(boolean throttlingEnabled) {
+ changeThrottlingFlag(PUBLISH_BUFFER_LIMITING_UPDATER, throttlingEnabled);
+ }
+
+ public void setPendingSendRequestsExceeded(boolean throttlingEnabled) {
+ boolean changed = changeThrottlingFlag(PENDING_SEND_REQUESTS_EXCEEDED_UPDATER, throttlingEnabled);
+ if (changed) {
+ // update the metrics that track throttling due to pending send requests
+ if (throttlingEnabled) {
+ throttledConnections.inc();
+ } else {
+ throttledConnections.dec();
+ }
+ }
+ }
+
+ private boolean changeThrottlingFlag(AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> throttlingFlagFieldUpdater,
+ boolean throttlingEnabled) {
+ // don't change a throttling flag if the channel is not active
+ if (!isChannelActive()) {
+ return false;
+ }
+ if (throttlingFlagFieldUpdater.compareAndSet(this, booleanToInt(!throttlingEnabled),
+ booleanToInt(throttlingEnabled))) {
+ if (throttlingEnabled) {
+ incrementThrottleCount();
+ } else {
+ decrementThrottleCount();
+ }
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ private static int booleanToInt(boolean value) {
+ return value ? 1 : 0;
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
index 0cf3c55d9d3..343aef09c1c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
@@ -219,32 +219,12 @@ public interface Topic {
void checkMessageDeduplicationInfo();
- void checkTopicPublishThrottlingRate();
-
- void incrementPublishCount(int numOfMessages, long msgSizeInBytes);
-
- void resetTopicPublishCountAndEnableReadIfRequired();
-
- void resetBrokerPublishCountAndEnableReadIfRequired(boolean doneReset);
-
- boolean isPublishRateExceeded();
-
- boolean isTopicPublishRateExceeded(int msgSize, int numMessages);
-
- boolean isResourceGroupRateLimitingEnabled();
-
- boolean isResourceGroupPublishRateExceeded(int msgSize, int numMessages);
-
- boolean isBrokerPublishRateExceeded();
+ void incrementPublishCount(Producer producer, int numOfMessages, long msgSizeInBytes);
boolean shouldProducerMigrate();
boolean isReplicationBacklogExist();
- void disableCnxAutoRead();
-
- void enableCnxAutoRead();
-
CompletableFuture<Void> onPoliciesUpdate(Policies data);
CompletableFuture<Void> checkBacklogQuotaExceeded(String producerName, BacklogQuotaType backlogQuotaType);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java
index a644bb70de8..6db085bd90a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java
@@ -61,14 +61,6 @@ public interface TransportCnx {
void closeProducer(Producer producer);
void closeProducer(Producer producer, Optional<BrokerLookupData> assignedBrokerLookupData);
- void cancelPublishRateLimiting();
-
- void cancelPublishBufferLimiting();
-
- void disableCnxAutoRead();
-
- void enableCnxAutoRead();
-
void execute(Runnable runnable);
void removedConsumer(Consumer consumer);
@@ -93,4 +85,22 @@ public interface TransportCnx {
* is null if the connection liveness check is disabled.
*/
CompletableFuture<Boolean> checkConnectionLiveness();
+
+ /**
+ * Increments the counter that controls the throttling of the connection by pausing reads.
+ * The connection will be throttled while the counter is greater than 0.
+ * <p>
+ * The caller is responsible for decrementing the counter by calling {@link #decrementThrottleCount()} when the
+ * connection should no longer be throttled.
+ */
+ void incrementThrottleCount();
+
+ /**
+ * Decrements the counter that controls the throttling of the connection by pausing reads.
+ * The connection will be throttled while the counter is greater than 0.
+ * <p>
+ * This method should be called when the connection should no longer be throttled. However, the caller should have
+ * previously called {@link #incrementThrottleCount()}.
+ */
+ void decrementThrottleCount();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
index c8f2d4ce62e..00cf3a6583b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -542,12 +542,6 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
} else {
subscriptions.forEach((s, sub) -> futures.add(sub.close(false, Optional.empty())));
}
- if (topicPublishRateLimiter != null) {
- topicPublishRateLimiter.close();
- }
- if (this.resourceGroupPublishLimiter != null) {
- this.resourceGroupPublishLimiter.unregisterRateLimitFunction(this.getName());
- }
if (entryFilters != null) {
entryFilters.getRight().forEach(filter -> {
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 233972452e9..b29cbcd660d 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
@@ -21,19 +21,17 @@ package org.apache.pulsar.broker.service.persistent;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
-import java.util.function.Supplier;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.qos.AsyncTokenBucket;
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.apache.pulsar.common.util.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class DispatchRateLimiter {
-
public enum Type {
TOPIC,
SUBSCRIPTION,
@@ -47,8 +45,8 @@ public class DispatchRateLimiter {
private final Type type;
private final BrokerService brokerService;
- private RateLimiter dispatchRateLimiterOnMessage;
- private RateLimiter dispatchRateLimiterOnByte;
+ private volatile AsyncTokenBucket dispatchRateLimiterOnMessage;
+ private volatile AsyncTokenBucket dispatchRateLimiterOnByte;
public DispatchRateLimiter(PersistentTopic topic, Type type) {
this(topic, null, type);
@@ -78,7 +76,7 @@ public class DispatchRateLimiter {
* @return
*/
public long getAvailableDispatchRateLimitOnMsg() {
- return dispatchRateLimiterOnMessage == null ? -1 : dispatchRateLimiterOnMessage.getAvailablePermits();
+ return dispatchRateLimiterOnMessage == null ? -1 : Math.max(dispatchRateLimiterOnMessage.getTokens(), 0);
}
/**
@@ -87,32 +85,22 @@ public class DispatchRateLimiter {
* @return
*/
public long getAvailableDispatchRateLimitOnByte() {
- return dispatchRateLimiterOnByte == null ? -1 : dispatchRateLimiterOnByte.getAvailablePermits();
+ return dispatchRateLimiterOnByte == null ? -1 : Math.max(dispatchRateLimiterOnByte.getTokens(), 0);
}
/**
* It acquires msg and bytes permits from rate-limiter and returns if acquired permits succeed.
*
- * @param msgPermits
- * @param bytePermits
- * @return
- */
- public boolean tryDispatchPermit(long msgPermits, long bytePermits) {
- boolean acquiredMsgPermit = msgPermits <= 0 || dispatchRateLimiterOnMessage == null
- || dispatchRateLimiterOnMessage.tryAcquire(msgPermits);
- boolean acquiredBytePermit = bytePermits <= 0 || dispatchRateLimiterOnByte == null
- || dispatchRateLimiterOnByte.tryAcquire(bytePermits);
- return acquiredMsgPermit && acquiredBytePermit;
- }
-
- /**
- * checks if dispatch-rate limit is configured and if it's configured then check if permits are available or not.
- *
- * @return
+ * @param numberOfMessages
+ * @param byteSize
*/
- public boolean hasMessageDispatchPermit() {
- return (dispatchRateLimiterOnMessage == null || dispatchRateLimiterOnMessage.getAvailablePermits() > 0)
- && (dispatchRateLimiterOnByte == null || dispatchRateLimiterOnByte.getAvailablePermits() > 0);
+ public void consumeDispatchQuota(long numberOfMessages, long byteSize) {
+ if (numberOfMessages > 0 && dispatchRateLimiterOnMessage != null) {
+ dispatchRateLimiterOnMessage.consumeTokens(numberOfMessages);
+ }
+ if (byteSize > 0 && dispatchRateLimiterOnByte != null) {
+ dispatchRateLimiterOnByte.consumeTokens(byteSize);
+ }
}
/**
@@ -227,60 +215,40 @@ public class DispatchRateLimiter {
long msgRate = dispatchRate.getDispatchThrottlingRateInMsg();
long byteRate = dispatchRate.getDispatchThrottlingRateInByte();
- long ratePeriod = dispatchRate.getRatePeriodInSecond();
+ long ratePeriodNanos = TimeUnit.SECONDS.toNanos(Math.max(dispatchRate.getRatePeriodInSecond(), 1));
- Supplier<Long> permitUpdaterMsg = dispatchRate.isRelativeToPublishRate()
- ? () -> getRelativeDispatchRateInMsg(dispatchRate)
- : null;
// update msg-rateLimiter
if (msgRate > 0) {
- if (this.dispatchRateLimiterOnMessage == null) {
+ if (dispatchRate.isRelativeToPublishRate()) {
this.dispatchRateLimiterOnMessage =
- RateLimiter.builder()
- .scheduledExecutorService(brokerService.pulsar().getExecutor())
- .permits(msgRate)
- .rateTime(ratePeriod)
- .timeUnit(TimeUnit.SECONDS)
- .permitUpdater(permitUpdaterMsg)
- .isDispatchOrPrecisePublishRateLimiter(true)
+ AsyncTokenBucket.builderForDynamicRate()
+ .rateFunction(() -> getRelativeDispatchRateInMsg(dispatchRate))
+ .ratePeriodNanosFunction(() -> ratePeriodNanos)
.build();
} else {
- this.dispatchRateLimiterOnMessage.setRate(msgRate, dispatchRate.getRatePeriodInSecond(),
- TimeUnit.SECONDS, permitUpdaterMsg);
+ this.dispatchRateLimiterOnMessage =
+ AsyncTokenBucket.builder().rate(msgRate).ratePeriodNanos(ratePeriodNanos)
+ .build();
}
} else {
- // message-rate should be disable and close
- if (this.dispatchRateLimiterOnMessage != null) {
- this.dispatchRateLimiterOnMessage.close();
- this.dispatchRateLimiterOnMessage = null;
- }
+ this.dispatchRateLimiterOnMessage = null;
}
- Supplier<Long> permitUpdaterByte = dispatchRate.isRelativeToPublishRate()
- ? () -> getRelativeDispatchRateInByte(dispatchRate)
- : null;
// update byte-rateLimiter
if (byteRate > 0) {
- if (this.dispatchRateLimiterOnByte == null) {
+ if (dispatchRate.isRelativeToPublishRate()) {
this.dispatchRateLimiterOnByte =
- RateLimiter.builder()
- .scheduledExecutorService(brokerService.pulsar().getExecutor())
- .permits(byteRate)
- .rateTime(ratePeriod)
- .timeUnit(TimeUnit.SECONDS)
- .permitUpdater(permitUpdaterByte)
- .isDispatchOrPrecisePublishRateLimiter(true)
+ AsyncTokenBucket.builderForDynamicRate()
+ .rateFunction(() -> getRelativeDispatchRateInByte(dispatchRate))
+ .ratePeriodNanosFunction(() -> ratePeriodNanos)
.build();
} else {
- this.dispatchRateLimiterOnByte.setRate(byteRate, dispatchRate.getRatePeriodInSecond(),
- TimeUnit.SECONDS, permitUpdaterByte);
+ this.dispatchRateLimiterOnByte =
+ AsyncTokenBucket.builder().rate(byteRate).ratePeriodNanos(ratePeriodNanos)
+ .build();
}
} else {
- // message-rate should be disable and close
- if (this.dispatchRateLimiterOnByte != null) {
- this.dispatchRateLimiterOnByte.close();
- this.dispatchRateLimiterOnByte = null;
- }
+ this.dispatchRateLimiterOnByte = null;
}
}
@@ -323,11 +291,9 @@ public class DispatchRateLimiter {
public void close() {
// close rate-limiter
if (dispatchRateLimiterOnMessage != null) {
- dispatchRateLimiterOnMessage.close();
dispatchRateLimiterOnMessage = null;
}
if (dispatchRateLimiterOnByte != null) {
- dispatchRateLimiterOnByte.close();
dispatchRateLimiterOnByte = null;
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java
index b8287dd2c14..191b19e3171 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java
@@ -148,7 +148,7 @@ public class GeoPersistentReplicator extends PersistentReplicator {
continue;
}
- dispatchRateLimiter.ifPresent(rateLimiter -> rateLimiter.tryDispatchPermit(1, entry.getLength()));
+ dispatchRateLimiter.ifPresent(rateLimiter -> rateLimiter.consumeDispatchQuota(1, entry.getLength()));
msg.setReplicatedFrom(localCluster);
headersAndPayload.retain();
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 30643b7058e..6001561d72a 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
@@ -402,52 +402,52 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
if (serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) {
if (topic.getBrokerDispatchRateLimiter().isPresent()) {
DispatchRateLimiter brokerRateLimiter = topic.getBrokerDispatchRateLimiter().get();
- if (reachDispatchRateLimit(brokerRateLimiter)) {
+ 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);
- } else {
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(brokerRateLimiter, messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
}
}
if (topic.getDispatchRateLimiter().isPresent()) {
DispatchRateLimiter topicRateLimiter = topic.getDispatchRateLimiter().get();
- if (reachDispatchRateLimit(topicRateLimiter)) {
+ 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);
- } else {
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(topicRateLimiter, messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
}
}
if (dispatchRateLimiter.isPresent()) {
- if (reachDispatchRateLimit(dispatchRateLimiter.get())) {
+ 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);
- } else {
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(dispatchRateLimiter.get(), messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
}
}
}
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 aea9b1c9b9e..54c039d632f 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
@@ -417,52 +417,52 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
if (serviceConfig.isDispatchThrottlingOnNonBacklogConsumerEnabled() || !cursor.isActive()) {
if (topic.getBrokerDispatchRateLimiter().isPresent()) {
DispatchRateLimiter brokerRateLimiter = topic.getBrokerDispatchRateLimiter().get();
- if (reachDispatchRateLimit(brokerRateLimiter)) {
+ 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);
- } else {
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(brokerRateLimiter, messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
}
}
if (topic.getDispatchRateLimiter().isPresent()) {
DispatchRateLimiter topicRateLimiter = topic.getDispatchRateLimiter().get();
- if (reachDispatchRateLimit(topicRateLimiter)) {
+ 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);
- } else {
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(topicRateLimiter, messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
}
}
if (dispatchRateLimiter.isPresent()) {
- if (reachDispatchRateLimit(dispatchRateLimiter.get())) {
+ 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);
- } else {
- Pair<Integer, Long> calculateToRead =
- updateMessagesToRead(dispatchRateLimiter.get(), messagesToRead, bytesToRead);
- messagesToRead = calculateToRead.getLeft();
- bytesToRead = calculateToRead.getRight();
}
}
}
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 a96036b7cfe..a4ac52d5ded 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
@@ -201,8 +201,11 @@ public abstract class PersistentReplicator extends AbstractReplicator
// handle rate limit
if (dispatchRateLimiter.isPresent() && dispatchRateLimiter.get().isDispatchRateLimitingEnabled()) {
DispatchRateLimiter rateLimiter = dispatchRateLimiter.get();
+ // if dispatch-rate is in msg then read only msg according to available permit
+ long availablePermitsOnMsg = rateLimiter.getAvailableDispatchRateLimitOnMsg();
+ long availablePermitsOnByte = rateLimiter.getAvailableDispatchRateLimitOnByte();
// no permits from rate limit
- if (!rateLimiter.hasMessageDispatchPermit()) {
+ if (availablePermitsOnByte == 0 || availablePermitsOnMsg == 0) {
if (log.isDebugEnabled()) {
log.debug("[{}] message-read exceeded topic replicator message-rate {}/{},"
+ " schedule after a {}",
@@ -213,9 +216,6 @@ public abstract class PersistentReplicator extends AbstractReplicator
}
return -1;
}
-
- // if dispatch-rate is in msg then read only msg according to available permit
- long availablePermitsOnMsg = rateLimiter.getAvailableDispatchRateLimitOnMsg();
if (availablePermitsOnMsg > 0) {
availablePermits = Math.min(availablePermits, (int) availablePermitsOnMsg);
}
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 16dadbd3597..48069cf5554 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
@@ -1522,12 +1522,6 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
} else {
subscriptions.forEach((s, sub) -> futures.add(sub.close(false, Optional.empty())));
}
- if (topicPublishRateLimiter != null) {
- topicPublishRateLimiter.close();
- }
- if (this.resourceGroupPublishLimiter != null) {
- this.resourceGroupPublishLimiter.unregisterRateLimitFunction(this.getName());
- }
//close entry filters
if (entryFilters != null) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java
index fb306348bcd..493072eb0c8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java
@@ -101,7 +101,7 @@ public class ShadowReplicator extends PersistentReplicator {
continue;
}
- dispatchRateLimiter.ifPresent(rateLimiter -> rateLimiter.tryDispatchPermit(1, entry.getLength()));
+ dispatchRateLimiter.ifPresent(rateLimiter -> rateLimiter.consumeDispatchQuota(1, entry.getLength()));
msgOut.recordEvent(headersAndPayload.readableBytes());
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 58f099a9dab..b1de10e73b7 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
@@ -19,40 +19,32 @@
package org.apache.pulsar.broker.service.persistent;
-import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.base.MoreObjects;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.SubscribeRate;
-import org.apache.pulsar.common.util.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class SubscribeRateLimiter {
-
private final String topicName;
private final BrokerService brokerService;
- private ConcurrentHashMap<ConsumerIdentifier, RateLimiter> subscribeRateLimiter;
- private final ScheduledExecutorService executorService;
- private ScheduledFuture<?> resetTask;
+ private ConcurrentHashMap<ConsumerIdentifier, AsyncTokenBucket> subscribeRateLimiter;
private SubscribeRate subscribeRate;
public SubscribeRateLimiter(PersistentTopic topic) {
this.topicName = topic.getName();
this.brokerService = topic.getBrokerService();
subscribeRateLimiter = new ConcurrentHashMap<>();
- this.executorService = brokerService.pulsar().getExecutor();
// get subscribeRate from topic level policies
this.subscribeRate = topic.getSubscribeRate();
if (isSubscribeRateEnabled(this.subscribeRate)) {
- resetTask = createTask();
log.info("[{}] configured subscribe-dispatch rate at broker {}", this.topicName, subscribeRate);
}
}
@@ -64,7 +56,7 @@ public class SubscribeRateLimiter {
*/
public long getAvailableSubscribeRateLimit(ConsumerIdentifier consumerIdentifier) {
return subscribeRateLimiter.get(consumerIdentifier)
- == null ? -1 : subscribeRateLimiter.get(consumerIdentifier).getAvailablePermits();
+ == null ? -1 : subscribeRateLimiter.get(consumerIdentifier).getTokens();
}
/**
@@ -74,8 +66,15 @@ public class SubscribeRateLimiter {
*/
public synchronized boolean tryAcquire(ConsumerIdentifier consumerIdentifier) {
addSubscribeLimiterIfAbsent(consumerIdentifier);
- return subscribeRateLimiter.get(consumerIdentifier)
- == null || subscribeRateLimiter.get(consumerIdentifier).tryAcquire();
+ AsyncTokenBucket tokenBucket = subscribeRateLimiter.get(consumerIdentifier);
+ if (tokenBucket == null) {
+ return true;
+ }
+ if (!tokenBucket.containsTokens(true)) {
+ return false;
+ }
+ tokenBucket.consumeTokens(1);
+ return true;
}
/**
@@ -86,7 +85,7 @@ public class SubscribeRateLimiter {
*/
public boolean subscribeAvailable(ConsumerIdentifier consumerIdentifier) {
return (subscribeRateLimiter.get(consumerIdentifier)
- == null || subscribeRateLimiter.get(consumerIdentifier).getAvailablePermits() > 0);
+ == null || subscribeRateLimiter.get(consumerIdentifier).containsTokens());
}
/**
@@ -98,15 +97,11 @@ public class SubscribeRateLimiter {
if (subscribeRateLimiter.get(consumerIdentifier) != null || !isSubscribeRateEnabled(this.subscribeRate)) {
return;
}
-
updateSubscribeRate(consumerIdentifier, this.subscribeRate);
}
private synchronized void removeSubscribeLimiter(ConsumerIdentifier consumerIdentifier) {
- if (this.subscribeRateLimiter.get(consumerIdentifier) != null) {
- this.subscribeRateLimiter.get(consumerIdentifier).close();
- this.subscribeRateLimiter.remove(consumerIdentifier);
- }
+ this.subscribeRateLimiter.remove(consumerIdentifier);
}
/**
@@ -117,23 +112,13 @@ public class SubscribeRateLimiter {
*/
private synchronized void updateSubscribeRate(ConsumerIdentifier consumerIdentifier, SubscribeRate subscribeRate) {
long ratePerConsumer = subscribeRate.subscribeThrottlingRatePerConsumer;
- long ratePeriod = subscribeRate.ratePeriodInSecond;
+ long ratePeriodNanos = TimeUnit.SECONDS.toNanos(Math.max(subscribeRate.ratePeriodInSecond, 1));
// update subscribe-rateLimiter
if (ratePerConsumer > 0) {
- if (this.subscribeRateLimiter.get(consumerIdentifier) == null) {
- this.subscribeRateLimiter.put(consumerIdentifier,
- RateLimiter.builder()
- .scheduledExecutorService(brokerService.pulsar().getExecutor())
- .permits(ratePerConsumer)
- .rateTime(ratePeriod)
- .timeUnit(TimeUnit.SECONDS)
- .build());
- } else {
- this.subscribeRateLimiter.get(consumerIdentifier)
- .setRate(ratePerConsumer, ratePeriod, TimeUnit.SECONDS,
- null);
- }
+ AsyncTokenBucket tokenBucket =
+ AsyncTokenBucket.builder().rate(ratePerConsumer).ratePeriodNanos(ratePeriodNanos).build();
+ this.subscribeRateLimiter.put(consumerIdentifier, tokenBucket);
} else {
// subscribe-rate should be disable and close
removeSubscribeLimiter(consumerIdentifier);
@@ -145,7 +130,6 @@ public class SubscribeRateLimiter {
return;
}
this.subscribeRate = subscribeRate;
- stopResetTask();
for (ConsumerIdentifier consumerIdentifier : this.subscribeRateLimiter.keySet()) {
if (!isSubscribeRateEnabled(this.subscribeRate)) {
removeSubscribeLimiter(consumerIdentifier);
@@ -154,7 +138,6 @@ public class SubscribeRateLimiter {
}
}
if (isSubscribeRateEnabled(this.subscribeRate)) {
- this.resetTask = createTask();
log.info("[{}] configured subscribe-dispatch rate at broker {}", this.topicName, subscribeRate);
}
}
@@ -209,32 +192,9 @@ public class SubscribeRateLimiter {
}
public void close() {
- closeAndClearRateLimiters();
- stopResetTask();
- }
- private ScheduledFuture<?> createTask() {
- return executorService.scheduleAtFixedRate(catchingAndLoggingThrowables(this::closeAndClearRateLimiters),
- this.subscribeRate.ratePeriodInSecond,
- this.subscribeRate.ratePeriodInSecond,
- TimeUnit.SECONDS);
}
- private void stopResetTask() {
- if (this.resetTask != null) {
- this.resetTask.cancel(false);
- }
- }
-
- private synchronized void closeAndClearRateLimiters() {
- // close rate-limiter
- this.subscribeRateLimiter.values().forEach(rateLimiter -> {
- if (rateLimiter != null) {
- rateLimiter.close();
- }
- });
- this.subscribeRateLimiter.clear();
- }
public SubscribeRate getSubscribeRate() {
return subscribeRate;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
index 25ca3bf1444..036d4354a5f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
@@ -2023,16 +2023,12 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
final String topicName = "persistent://" + myNamespace + "/test-" + UUID.randomUUID();
pulsarClient.newProducer().topic(topicName).create().close();
- Field publishMaxMessageRate = PublishRateLimiterImpl.class.getDeclaredField("publishMaxMessageRate");
- publishMaxMessageRate.setAccessible(true);
- Field publishMaxByteRate = PublishRateLimiterImpl.class.getDeclaredField("publishMaxByteRate");
- publishMaxByteRate.setAccessible(true);
//1 use broker-level policy by default
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topicName).get().get();
PublishRateLimiterImpl publishRateLimiter = (PublishRateLimiterImpl) topic.getTopicPublishRateLimiter();
- Assert.assertEquals(publishMaxMessageRate.get(publishRateLimiter), 5);
- Assert.assertEquals(publishMaxByteRate.get(publishRateLimiter), 50L);
+ Assert.assertEquals(publishRateLimiter.getTokenBucketOnMessage().getRate(), 5);
+ Assert.assertEquals(publishRateLimiter.getTokenBucketOnByte().getRate(), 50L);
//2 set namespace-level policy
PublishRate publishMsgRate = new PublishRate(10, 100L);
@@ -2041,12 +2037,12 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
Awaitility.await()
.until(() -> {
PublishRateLimiterImpl limiter = (PublishRateLimiterImpl) topic.getTopicPublishRateLimiter();
- return (int)publishMaxMessageRate.get(limiter) == 10;
+ return (int)limiter.getTokenBucketOnMessage().getRate() == 10;
});
publishRateLimiter = (PublishRateLimiterImpl) topic.getTopicPublishRateLimiter();
- Assert.assertEquals(publishMaxMessageRate.get(publishRateLimiter), 10);
- Assert.assertEquals(publishMaxByteRate.get(publishRateLimiter), 100L);
+ Assert.assertEquals(publishRateLimiter.getTokenBucketOnMessage().getRate(), 10);
+ Assert.assertEquals(publishRateLimiter.getTokenBucketOnByte().getRate(), 100L);
//3 set topic-level policy, namespace-level policy should be overwritten
PublishRate publishMsgRate2 = new PublishRate(11, 101L);
@@ -2056,8 +2052,8 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
.until(() -> admin.topicPolicies().getPublishRate(topicName) != null);
publishRateLimiter = (PublishRateLimiterImpl) topic.getTopicPublishRateLimiter();
- Assert.assertEquals(publishMaxMessageRate.get(publishRateLimiter), 11);
- Assert.assertEquals(publishMaxByteRate.get(publishRateLimiter), 101L);
+ Assert.assertEquals(publishRateLimiter.getTokenBucketOnMessage().getRate(), 11);
+ Assert.assertEquals(publishRateLimiter.getTokenBucketOnByte().getRate(), 101L);
//4 remove topic-level policy, namespace-level policy will take effect
admin.topicPolicies().removePublishRate(topicName);
@@ -2066,8 +2062,8 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
.until(() -> admin.topicPolicies().getPublishRate(topicName) == null);
publishRateLimiter = (PublishRateLimiterImpl) topic.getTopicPublishRateLimiter();
- Assert.assertEquals(publishMaxMessageRate.get(publishRateLimiter), 10);
- Assert.assertEquals(publishMaxByteRate.get(publishRateLimiter), 100L);
+ Assert.assertEquals(publishRateLimiter.getTokenBucketOnMessage().getRate(), 10);
+ Assert.assertEquals(publishRateLimiter.getTokenBucketOnByte().getRate(), 100L);
//5 remove namespace-level policy, broker-level policy will take effect
admin.namespaces().removePublishRate(myNamespace);
@@ -2075,12 +2071,12 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
Awaitility.await()
.until(() -> {
PublishRateLimiterImpl limiter = (PublishRateLimiterImpl) topic.getTopicPublishRateLimiter();
- return (int)publishMaxMessageRate.get(limiter) == 5;
+ return (int)limiter.getTokenBucketOnMessage().getRate() == 5;
});
publishRateLimiter = (PublishRateLimiterImpl) topic.getTopicPublishRateLimiter();
- Assert.assertEquals(publishMaxMessageRate.get(publishRateLimiter), 5);
- Assert.assertEquals(publishMaxByteRate.get(publishRateLimiter), 50L);
+ Assert.assertEquals(publishRateLimiter.getTokenBucketOnMessage().getRate(), 5);
+ Assert.assertEquals(publishRateLimiter.getTokenBucketOnByte().getRate(), 50L);
}
@Test(timeOut = 20000)
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
new file mode 100644
index 00000000000..b446f9e902f
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/qos/AsyncTokenBucketTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.testng.Assert.assertEquals;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+
+public class AsyncTokenBucketTest {
+ private AtomicLong manualClockSource;
+ private MonotonicSnapshotClock clockSource;
+
+ private AsyncTokenBucket asyncTokenBucket;
+
+ @BeforeMethod
+ public void setup() {
+ manualClockSource = new AtomicLong(TimeUnit.SECONDS.toNanos(100));
+ clockSource = requestSnapshot -> manualClockSource.get();
+ }
+
+
+ private void incrementSeconds(int seconds) {
+ manualClockSource.addAndGet(TimeUnit.SECONDS.toNanos(seconds));
+ }
+
+ private void incrementMillis(long millis) {
+ manualClockSource.addAndGet(TimeUnit.MILLISECONDS.toNanos(millis));
+ }
+
+ @Test
+ void shouldAddTokensWithConfiguredRate() {
+ asyncTokenBucket =
+ AsyncTokenBucket.builder().capacity(100).rate(10).initialTokens(0).clock(clockSource).build();
+ incrementSeconds(5);
+ assertEquals(asyncTokenBucket.getTokens(), 50);
+ incrementSeconds(1);
+ assertEquals(asyncTokenBucket.getTokens(), 60);
+ incrementSeconds(4);
+ assertEquals(asyncTokenBucket.getTokens(), 100);
+
+ // No matter how long the period is, tokens do not go above capacity
+ incrementSeconds(5);
+ assertEquals(asyncTokenBucket.getTokens(), 100);
+
+ // Consume all and verify none available and then wait 1 period and check replenished
+ asyncTokenBucket.consumeTokens(100);
+ assertEquals(asyncTokenBucket.tokens(true), 0);
+ incrementSeconds(1);
+ assertEquals(asyncTokenBucket.getTokens(), 10);
+ }
+
+ @Test
+ void shouldCalculatePauseCorrectly() {
+ asyncTokenBucket =
+ AsyncTokenBucket.builder().capacity(100).rate(10).initialTokens(0).clock(clockSource)
+ .build();
+ incrementSeconds(5);
+ asyncTokenBucket.consumeTokens(100);
+ assertEquals(asyncTokenBucket.getTokens(), -50);
+ assertEquals(TimeUnit.NANOSECONDS.toMillis(asyncTokenBucket.calculateThrottlingDuration()), 5100);
+ }
+
+ @Test
+ void shouldSupportFractionsWhenUpdatingTokens() {
+ asyncTokenBucket =
+ AsyncTokenBucket.builder().capacity(100).rate(10).initialTokens(0).clock(clockSource).build();
+ incrementMillis(100);
+ assertEquals(asyncTokenBucket.getTokens(), 1);
+ }
+
+ @Test
+ void shouldSupportFractionsAndRetainLeftoverWhenUpdatingTokens() {
+ asyncTokenBucket =
+ AsyncTokenBucket.builder().capacity(100).rate(10).initialTokens(0).clock(clockSource).build();
+ for (int i = 0; i < 150; i++) {
+ incrementMillis(1);
+ }
+ assertEquals(asyncTokenBucket.getTokens(), 1);
+ incrementMillis(150);
+ assertEquals(asyncTokenBucket.getTokens(), 3);
+ }
+
+}
\ No newline at end of file
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupRateLimiterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupRateLimiterTest.java
index d4bc9eb925d..3e88af9ec0b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupRateLimiterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupRateLimiterTest.java
@@ -21,7 +21,9 @@ package org.apache.pulsar.broker.resourcegroup;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
-
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.MessageId;
@@ -34,12 +36,9 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
public class ResourceGroupRateLimiterTest extends BrokerTestBase {
+ private static final long MESSAGE_SIZE_SERIALIZED = 41L;
final String rgName = "testRG";
org.apache.pulsar.common.policies.data.ResourceGroup testAddRg =
new org.apache.pulsar.common.policies.data.ResourceGroup();
@@ -53,7 +52,6 @@ public class ResourceGroupRateLimiterTest extends BrokerTestBase {
conf.setMaxPendingPublishRequestsPerConnection(0);
super.baseSetup();
prepareData();
-
}
@AfterClass(alwaysRun = true)
@@ -123,7 +121,7 @@ public class ResourceGroupRateLimiterTest extends BrokerTestBase {
try {
// third one should succeed
- messageId = producer.sendAsync(new byte[MESSAGE_SIZE]).get(100, TimeUnit.MILLISECONDS);
+ messageId = producer.sendAsync(new byte[MESSAGE_SIZE]).get(300, TimeUnit.MILLISECONDS);
Assert.assertNotNull(messageId);
} catch (TimeoutException e) {
Assert.fail("should not fail");
@@ -133,6 +131,8 @@ public class ResourceGroupRateLimiterTest extends BrokerTestBase {
admin.namespaces().removeNamespaceResourceGroup(namespaceName);
deleteResourceGroup(rgName);
+ Thread.sleep(2000);
+
// No rate limits should be applied.
for (int i = 0; i < 5; i++) {
messageId = producer.sendAsync(new byte[MESSAGE_SIZE]).get(100, TimeUnit.MILLISECONDS);
@@ -148,7 +148,7 @@ public class ResourceGroupRateLimiterTest extends BrokerTestBase {
}
private void prepareData() {
- testAddRg.setPublishRateInBytes(Long.valueOf(MESSAGE_SIZE));
+ testAddRg.setPublishRateInBytes(Long.valueOf(MESSAGE_SIZE_SERIALIZED));
testAddRg.setPublishRateInMsgs(1);
testAddRg.setDispatchRateInMsgs(-1);
testAddRg.setDispatchRateInBytes(Long.valueOf(-1));
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 03aaf3c7bb2..0a1213ac1a8 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
@@ -118,7 +118,7 @@ public class AbstractBaseDispatcherTest {
int size = this.helper.filterEntriesForConsumer(entries, batchSizes, sendMessageInfo, null, cursor, false, null);
assertEquals(size, 0);
- verify(subscriptionDispatchRateLimiter).tryDispatchPermit(1, expectedBytePermits);
+ verify(subscriptionDispatchRateLimiter).consumeDispatchQuota(1, expectedBytePermits);
}
@Test
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 7f5d3e0edf1..39be56e3f41 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
@@ -19,12 +19,14 @@
package org.apache.pulsar.broker.service;
import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.withSettings;
import static org.testng.Assert.assertEquals;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -43,8 +45,10 @@ public class AbstractTopicTest {
subscription = mock(AbstractSubscription.class);
when(brokerService.pulsar()).thenReturn(pulsarService);
+ doReturn(pulsarService).when(brokerService).getPulsar();
when(pulsarService.getConfiguration()).thenReturn(serviceConfiguration);
when(brokerService.getBacklogQuotaManager()).thenReturn(backlogQuotaManager);
+ doReturn(AsyncTokenBucket.DEFAULT_SNAPSHOT_CLOCK).when(pulsarService).getMonotonicSnapshotClock();
topic = mock(AbstractTopic.class, withSettings()
.useConstructor("topic", brokerService)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index 35f68a7a9ca..f456a133d99 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -61,7 +61,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
@@ -1482,81 +1481,6 @@ public class BrokerServiceTest extends BrokerTestBase {
Assert.assertTrue(sb.toString().contains("test_metrics"));
}
- @Test
- public void testPublishRateLimiterMonitor() {
- BrokerService.PublishRateLimiterMonitor monitor = new BrokerService.PublishRateLimiterMonitor("test");
- AtomicInteger checkCnt = new AtomicInteger(0);
- AtomicInteger refreshCnt = new AtomicInteger(0);
- monitor.startOrUpdate(100, checkCnt::incrementAndGet, refreshCnt::incrementAndGet);
- Assert.assertEquals(monitor.getTickTimeMs(), 100);
- Awaitility.await().until(() -> checkCnt.get() > 0);
- Awaitility.await().until(() -> refreshCnt.get() > 0);
-
- monitor.startOrUpdate(500, checkCnt::incrementAndGet, refreshCnt::incrementAndGet);
- Assert.assertEquals(monitor.getTickTimeMs(), 500);
- checkCnt.set(0);
- refreshCnt.set(0);
- Awaitility.await().until(() -> checkCnt.get() > 0);
- Awaitility.await().until(() -> refreshCnt.get() > 0);
-
- monitor.stop();
- Assert.assertEquals(monitor.getTickTimeMs(), 0);
- }
-
- @Test
- public void testDynamicBrokerPublisherThrottlingTickTimeMillis() throws Exception {
- cleanup();
- conf.setBrokerPublisherThrottlingMaxMessageRate(1000);
- conf.setBrokerPublisherThrottlingTickTimeMillis(100);
- setup();
-
- int prevTickMills = 100;
- BrokerService.PublishRateLimiterMonitor monitor = pulsar.getBrokerService().brokerPublishRateLimiterMonitor;
- Awaitility.await().until(() -> monitor.getTickTimeMs() == prevTickMills);
-
- int newTickMills = prevTickMills * 2;
- admin.brokers().updateDynamicConfiguration("brokerPublisherThrottlingTickTimeMillis",
- String.valueOf(newTickMills));
- Awaitility.await().until(() -> monitor.getTickTimeMs() == newTickMills);
-
- admin.brokers().updateDynamicConfiguration("brokerPublisherThrottlingTickTimeMillis",
- String.valueOf(0));
- Awaitility.await().until(() -> monitor.getTickTimeMs() == 0);
-
- admin.brokers().updateDynamicConfiguration("brokerPublisherThrottlingTickTimeMillis",
- String.valueOf(prevTickMills));
- Awaitility.await().until(() -> monitor.getTickTimeMs() == prevTickMills);
- }
-
- @Test
- public void testDynamicTopicPublisherThrottlingTickTimeMillis() throws Exception {
- cleanup();
- conf.setPreciseTopicPublishRateLimiterEnable(false);
- conf.setMaxPublishRatePerTopicInMessages(1000);
- conf.setTopicPublisherThrottlingTickTimeMillis(100);
- setup();
-
- @Cleanup
- Producer<byte[]> producer = pulsarClient.newProducer().topic("persistent://prop/ns-abc/test-topic").create();
-
- int prevTickMills = 100;
- BrokerService.PublishRateLimiterMonitor monitor = pulsar.getBrokerService().topicPublishRateLimiterMonitor;
- Awaitility.await().until(() -> monitor.getTickTimeMs() == prevTickMills);
-
- int newTickMills = prevTickMills * 2;
- admin.brokers().updateDynamicConfiguration("topicPublisherThrottlingTickTimeMillis",
- String.valueOf(newTickMills));
- Awaitility.await().until(() -> monitor.getTickTimeMs() == newTickMills);
-
- admin.brokers().updateDynamicConfiguration("topicPublisherThrottlingTickTimeMillis",
- String.valueOf(0));
- Awaitility.await().until(() -> monitor.getTickTimeMs() == 0);
-
- admin.brokers().updateDynamicConfiguration("topicPublisherThrottlingTickTimeMillis",
- String.valueOf(prevTickMills));
- Awaitility.await().until(() -> monitor.getTickTimeMs() == prevTickMills);
- }
-
@Test
public void shouldNotPreventCreatingTopicWhenNonexistingTopicIsCached() throws Exception {
// run multiple iterations to increase the chance of reproducing a race condition in the topic cache
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessagePublishBufferThrottleTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessagePublishBufferThrottleTest.java
index 173f772a731..27f72eac942 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessagePublishBufferThrottleTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessagePublishBufferThrottleTest.java
@@ -139,12 +139,5 @@ public class MessagePublishBufferThrottleTest extends BrokerTestBase {
Awaitility.await().untilAsserted(() ->
assertEquals(pulsar.getBrokerService().getPausedConnections(), 0));
-
- // Resume message publish.
- ((AbstractTopic)topicRef).producers.get("producer-name").getCnx().enableCnxAutoRead();
-
- flushFuture.get();
- Awaitility.await().untilAsserted(() ->
- assertEquals(pulsar.getBrokerService().getPausedConnections(), 0));
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisePublishLimiterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisePublishLimiterTest.java
deleted file mode 100644
index 9d5cfe5b5d2..00000000000
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisePublishLimiterTest.java
+++ /dev/null
@@ -1,60 +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.service;
-
-import static org.testng.Assert.assertFalse;
-import static org.testng.Assert.assertTrue;
-import lombok.Cleanup;
-import org.apache.pulsar.common.policies.data.PublishRate;
-import org.testng.annotations.Test;
-
-public class PrecisePublishLimiterTest {
-
- @Test
- void shouldResetMsgLimitAfterUpdate() {
- @Cleanup
- PrecisePublishLimiter precisePublishLimiter = new PrecisePublishLimiter(new PublishRate(), () -> {
- });
- precisePublishLimiter.update(new PublishRate(1, 1));
- assertFalse(precisePublishLimiter.tryAcquire(99, 99));
- precisePublishLimiter.update(new PublishRate(-1, 100));
- assertTrue(precisePublishLimiter.tryAcquire(99, 99));
- }
-
- @Test
- void shouldResetBytesLimitAfterUpdate() {
- @Cleanup
- PrecisePublishLimiter precisePublishLimiter = new PrecisePublishLimiter(new PublishRate(), () -> {
- });
- precisePublishLimiter.update(new PublishRate(1, 1));
- assertFalse(precisePublishLimiter.tryAcquire(99, 99));
- precisePublishLimiter.update(new PublishRate(100, -1));
- assertTrue(precisePublishLimiter.tryAcquire(99, 99));
- }
-
- @Test
- void shouldCloseResources() throws Exception {
- for (int i = 0; i < 20000; i++) {
- PrecisePublishLimiter precisePublishLimiter = new PrecisePublishLimiter(new PublishRate(100, 100), () -> {
- });
- precisePublishLimiter.tryAcquire(99, 99);
- precisePublishLimiter.close();
- }
- }
-}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java
index 8dfed77814b..ec952a7ca77 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java
@@ -18,7 +18,10 @@
*/
package org.apache.pulsar.broker.service;
-import static org.testng.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.testng.annotations.Test;
public class PublishRateLimiterDisableTest {
@@ -26,7 +29,9 @@ public class PublishRateLimiterDisableTest {
// GH issue #10603
@Test
void shouldAlwaysAllowAcquire() {
- PublishRateLimiterDisable publishRateLimiter = PublishRateLimiterDisable.DISABLED_RATE_LIMITER;
- assertTrue(publishRateLimiter.tryAcquire(Integer.MAX_VALUE, Long.MAX_VALUE));
+ PublishRateLimiter publishRateLimiter = new PublishRateLimiterImpl(AsyncTokenBucket.DEFAULT_SNAPSHOT_CLOCK);
+ Producer producer = mock(Producer.class);
+ publishRateLimiter.handlePublishThrottling(producer, Integer.MAX_VALUE, Long.MAX_VALUE);
+ verify(producer, never()).incrementThrottleCount();
}
}
\ No newline at end of file
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 f3cb25e789f..2c44ba7e230 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
@@ -16,154 +16,105 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.pulsar.broker.service;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import io.netty.channel.EventLoopGroup;
+import java.util.HashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.PublishRate;
-import org.apache.pulsar.common.util.RateLimiter;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import java.lang.reflect.Field;
-import java.lang.reflect.Method;
-import java.util.HashMap;
-import java.util.concurrent.ScheduledFuture;
-
-import static org.testng.Assert.assertFalse;
-import static org.testng.Assert.assertTrue;
-
@Test(groups = "broker")
public class PublishRateLimiterTest {
private final String CLUSTER_NAME = "clusterName";
private final Policies policies = new Policies();
private final PublishRate publishRate = new PublishRate(10, 100);
private final PublishRate newPublishRate = new PublishRate(20, 200);
+ private AtomicLong manualClockSource;
- private PrecisePublishLimiter precisePublishLimiter;
+ private Producer producer;
private PublishRateLimiterImpl publishRateLimiter;
+ private AtomicInteger throttleCount = new AtomicInteger(0);
+
@BeforeMethod
public void setup() throws Exception {
policies.publishMaxMessageRate = new HashMap<>();
policies.publishMaxMessageRate.put(CLUSTER_NAME, publishRate);
-
- precisePublishLimiter = new PrecisePublishLimiter(policies, CLUSTER_NAME, () -> System.out.print("Refresh permit"));
- publishRateLimiter = new PublishRateLimiterImpl(policies, CLUSTER_NAME);
+ manualClockSource = new AtomicLong(TimeUnit.SECONDS.toNanos(100));
+ publishRateLimiter = new PublishRateLimiterImpl(requestSnapshot -> manualClockSource.get());
+ publishRateLimiter.update(policies, CLUSTER_NAME);
+ producer = mock(Producer.class);
+ throttleCount.set(0);
+ doAnswer(a -> {
+ throttleCount.incrementAndGet();
+ return null;
+ }).when(producer).incrementThrottleCount();
+ doAnswer(a -> {
+ throttleCount.decrementAndGet();
+ return null;
+ }).when(producer).decrementThrottleCount();
+ TransportCnx transportCnx = mock(TransportCnx.class);
+ when(producer.getCnx()).thenReturn(transportCnx);
+ BrokerService brokerService = mock(BrokerService.class);
+ when(transportCnx.getBrokerService()).thenReturn(brokerService);
+ EventLoopGroup eventLoopGroup = mock(EventLoopGroup.class);
+ when(brokerService.executor()).thenReturn(eventLoopGroup);
+ doReturn(null).when(eventLoopGroup).schedule(any(Runnable.class), anyLong(), any());
+ incrementSeconds(1);
}
@AfterMethod
public void cleanup() throws Exception {
policies.publishMaxMessageRate.clear();
policies.publishMaxMessageRate = null;
- precisePublishLimiter.close();
- publishRateLimiter.close();
+ }
+
+ private void incrementSeconds(int seconds) {
+ manualClockSource.addAndGet(TimeUnit.SECONDS.toNanos(seconds));
}
@Test
public void testPublishRateLimiterImplExceed() throws Exception {
// increment not exceed
- publishRateLimiter.incrementPublishCount(5, 50);
- publishRateLimiter.checkPublishRate();
- assertFalse(publishRateLimiter.isPublishRateExceeded());
- publishRateLimiter.resetPublishCount();
+ publishRateLimiter.handlePublishThrottling(producer, 5, 50);
+ assertEquals(throttleCount.get(), 0);
+
+ incrementSeconds(1);
// numOfMessages increment exceeded
- publishRateLimiter.incrementPublishCount(11, 100);
- publishRateLimiter.checkPublishRate();
- assertTrue(publishRateLimiter.isPublishRateExceeded());
- publishRateLimiter.resetPublishCount();
+ publishRateLimiter.handlePublishThrottling(producer, 11, 100);
+ assertEquals(throttleCount.get(), 1);
- // msgSizeInBytes increment exceeded
- publishRateLimiter.incrementPublishCount(9, 110);
- publishRateLimiter.checkPublishRate();
- assertTrue(publishRateLimiter.isPublishRateExceeded());
+ incrementSeconds(1);
+ // msgSizeInBytes increment exceeded
+ publishRateLimiter.handlePublishThrottling(producer, 9, 110);
+ assertEquals(throttleCount.get(), 2);
}
@Test
public void testPublishRateLimiterImplUpdate() {
- publishRateLimiter.incrementPublishCount(11, 110);
- publishRateLimiter.checkPublishRate();
- assertTrue(publishRateLimiter.isPublishRateExceeded());
+ publishRateLimiter.handlePublishThrottling(producer, 11, 110);
+ assertEquals(throttleCount.get(), 1);
// update
+ throttleCount.set(0);
publishRateLimiter.update(newPublishRate);
- publishRateLimiter.incrementPublishCount(11, 110);
- publishRateLimiter.checkPublishRate();
- assertFalse(publishRateLimiter.isPublishRateExceeded());
-
- }
-
- @Test
- public void testPrecisePublishRateLimiterUpdate() {
- assertFalse(precisePublishLimiter.tryAcquire(15, 150));
-
- //update
- precisePublishLimiter.update(newPublishRate);
- assertTrue(precisePublishLimiter.tryAcquire(15, 150));
- }
-
- @Test
- public void testPrecisePublishRateLimiterAcquire() throws Exception {
- Class precisePublishLimiterClass = Class.forName("org.apache.pulsar.broker.service.PrecisePublishLimiter");
- Field topicPublishRateLimiterOnMessageField = precisePublishLimiterClass.getDeclaredField("topicPublishRateLimiterOnMessage");
- Field topicPublishRateLimiterOnByteField = precisePublishLimiterClass.getDeclaredField("topicPublishRateLimiterOnByte");
- topicPublishRateLimiterOnMessageField.setAccessible(true);
- topicPublishRateLimiterOnByteField.setAccessible(true);
-
- RateLimiter topicPublishRateLimiterOnMessage = (RateLimiter)topicPublishRateLimiterOnMessageField.get(
- precisePublishLimiter);
- RateLimiter topicPublishRateLimiterOnByte = (RateLimiter)topicPublishRateLimiterOnByteField.get(
- precisePublishLimiter);
-
- Method renewTopicPublishRateLimiterOnMessageMethod = topicPublishRateLimiterOnMessage.getClass().getDeclaredMethod("renew", null);
- Method renewTopicPublishRateLimiterOnByteMethod = topicPublishRateLimiterOnByte.getClass().getDeclaredMethod("renew", null);
- renewTopicPublishRateLimiterOnMessageMethod.setAccessible(true);
- renewTopicPublishRateLimiterOnByteMethod.setAccessible(true);
-
- // running tryAcquire in order to lazyInit the renewTask
- precisePublishLimiter.tryAcquire(1, 10);
-
- Field onMessageRenewTaskField = topicPublishRateLimiterOnMessage.getClass().getDeclaredField("renewTask");
- Field onByteRenewTaskField = topicPublishRateLimiterOnByte.getClass().getDeclaredField("renewTask");
- onMessageRenewTaskField.setAccessible(true);
- onByteRenewTaskField.setAccessible(true);
- ScheduledFuture<?> onMessageRenewTask = (ScheduledFuture<?>) onMessageRenewTaskField.get(topicPublishRateLimiterOnMessage);
- ScheduledFuture<?> onByteRenewTask = (ScheduledFuture<?>) onByteRenewTaskField.get(topicPublishRateLimiterOnByte);
-
- onMessageRenewTask.cancel(false);
- onByteRenewTask.cancel(false);
-
- // renewing the permits from previous tests
- renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
- renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
-
- // tryAcquire not exceeded
- assertTrue(precisePublishLimiter.tryAcquire(1, 10));
- renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
- renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
-
- // tryAcquire numOfMessages exceeded
- assertFalse(precisePublishLimiter.tryAcquire(11, 100));
- renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
- renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
-
- // tryAcquire msgSizeInBytes exceeded
- assertFalse(precisePublishLimiter.tryAcquire(10, 101));
- renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
- renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
- renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
- renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
-
- // tryAcquire exceeded exactly
- assertFalse(precisePublishLimiter.tryAcquire(10, 100));
- renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
- renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
- renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
- renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
-
- // tryAcquire not exceeded
- assertTrue(precisePublishLimiter.tryAcquire(9, 99));
+ publishRateLimiter.handlePublishThrottling(producer, 11, 110);
+ assertEquals(throttleCount.get(), 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 b97210b009a..747ef3b7f5c 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
@@ -33,6 +33,7 @@ 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;
@@ -59,6 +60,7 @@ public class ReplicatorRateLimiterTest extends ReplicatorTestBase {
@Override
@BeforeClass(timeOut = 300000)
public void setup() throws Exception {
+ AsyncTokenBucket.switchToConsistentTokensView();
super.setup();
}
@@ -66,6 +68,7 @@ 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/PrecisTopicPublishRateThrottleTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPublishRateThrottleTest.java
similarity index 72%
rename from pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisTopicPublishRateThrottleTest.java
rename to pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPublishRateThrottleTest.java
index c22ed41fc15..721d0493425 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisTopicPublishRateThrottleTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPublishRateThrottleTest.java
@@ -18,63 +18,31 @@
*/
package org.apache.pulsar.broker.service;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
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;
+import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
@Test(groups = "broker")
-public class PrecisTopicPublishRateThrottleTest extends BrokerTestBase{
+public class TopicPublishRateThrottleTest extends BrokerTestBase{
+ @BeforeMethod(alwaysRun = true)
@Override
protected void setup() throws Exception {
- //No-op
+ AsyncTokenBucket.switchToConsistentTokensView();
}
+ @AfterMethod(alwaysRun = true)
@Override
protected void cleanup() throws Exception {
- //No-op
- }
-
- @Test
- public void testPrecisTopicPublishRateLimitingDisabled() throws Exception {
- PublishRate publishRate = new PublishRate(1,10);
- // disable precis topic publish rate limiting
- conf.setPreciseTopicPublishRateLimiterEnable(false);
- conf.setMaxPendingPublishRequestsPerConnection(0);
- super.baseSetup();
- admin.namespaces().setPublishRate("prop/ns-abc", publishRate);
- final String topic = "persistent://prop/ns-abc/testPrecisTopicPublishRateLimiting";
- org.apache.pulsar.client.api.Producer<byte[]> producer = pulsarClient.newProducer()
- .topic(topic)
- .producerName("producer-name")
- .create();
-
- Topic topicRef = pulsar.getBrokerService().getTopicReference(topic).get();
- Assert.assertNotNull(topicRef);
- MessageId messageId = null;
- try {
- // first will be success
- messageId = producer.sendAsync(new byte[10]).get(500, TimeUnit.MILLISECONDS);
- Assert.assertNotNull(messageId);
- // second will be success
- messageId = producer.sendAsync(new byte[10]).get(500, TimeUnit.MILLISECONDS);
- Assert.assertNotNull(messageId);
- } catch (TimeoutException e) {
- // No-op
- }
- Thread.sleep(1000);
- try {
- messageId = producer.sendAsync(new byte[10]).get(1, TimeUnit.SECONDS);
- } catch (TimeoutException e) {
- // No-op
- }
- Assert.assertNotNull(messageId);
super.internalCleanup();
+ AsyncTokenBucket.resetToDefaultEventualConsistentTokensView();
}
@Test
@@ -103,7 +71,6 @@ public class PrecisTopicPublishRateThrottleTest extends BrokerTestBase{
} catch (TimeoutException e) {
// No-op
}
- super.internalCleanup();
}
@Test
@@ -139,7 +106,6 @@ public class PrecisTopicPublishRateThrottleTest extends BrokerTestBase{
// No-op
}
Assert.assertNotNull(messageId);
- super.internalCleanup();
}
@Test
@@ -164,9 +130,10 @@ public class PrecisTopicPublishRateThrottleTest extends BrokerTestBase{
"" + rateInMsg));
Topic topicRef = pulsar.getBrokerService().getTopicReference(topic).get();
Assert.assertNotNull(topicRef);
- PrecisePublishLimiter limiter = ((PrecisePublishLimiter) ((AbstractTopic) topicRef).topicPublishRateLimiter);
- Awaitility.await().untilAsserted(() -> Assert.assertEquals(limiter.publishMaxMessageRate, rateInMsg));
- Assert.assertEquals(limiter.publishMaxByteRate, 0);
+ PublishRateLimiterImpl limiter = ((PublishRateLimiterImpl) ((AbstractTopic) topicRef).topicPublishRateLimiter);
+ Awaitility.await()
+ .untilAsserted(() -> Assert.assertEquals(limiter.getTokenBucketOnMessage().getRate(), rateInMsg));
+ Assert.assertNull(limiter.getTokenBucketOnByte());
// maxPublishRatePerTopicInBytes
admin.brokers().updateDynamicConfiguration("maxPublishRatePerTopicInBytes", "" + rateInByte);
@@ -174,10 +141,10 @@ public class PrecisTopicPublishRateThrottleTest extends BrokerTestBase{
.untilAsserted(() ->
Assert.assertEquals(admin.brokers().getAllDynamicConfigurations().get("maxPublishRatePerTopicInBytes"),
"" + rateInByte));
- Awaitility.await().untilAsserted(() -> Assert.assertEquals(limiter.publishMaxByteRate, rateInByte));
- Assert.assertEquals(limiter.publishMaxMessageRate, rateInMsg);
+ Awaitility.await()
+ .untilAsserted(() -> Assert.assertEquals(limiter.getTokenBucketOnByte().getRate(), rateInByte));
+ Assert.assertEquals(limiter.getTokenBucketOnMessage().getRate(), rateInMsg);
producer.close();
- super.internalCleanup();
}
}
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 a6656815281..81a565f9d6f 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
@@ -50,6 +50,7 @@ import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.compaction.CompactionServiceFactory;
import org.testng.annotations.Test;
@@ -255,7 +256,9 @@ public class MessageDuplicationTest {
BrokerService brokerService = mock(BrokerService.class);
doReturn(eventLoopGroup).when(brokerService).executor();
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();
PersistentTopic persistentTopic = spyWithClassAndConstructorArgs(PersistentTopic.class, "topic-1", brokerService, managedLedger, messageDeduplication);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
index 80f9cd8913f..abd00d374f3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
@@ -120,25 +120,12 @@ public class PrometheusMetricsTest extends BrokerTestBase {
@Test
public void testPublishRateLimitedTimes() throws Exception {
- checkPublishRateLimitedTimes(true);
- checkPublishRateLimitedTimes(false);
- }
-
- private void checkPublishRateLimitedTimes(boolean preciseRateLimit) throws Exception {
cleanup();
- if (preciseRateLimit) {
- conf.setBrokerPublisherThrottlingTickTimeMillis(10000000);
- conf.setMaxPublishRatePerTopicInMessages(1);
- conf.setMaxPublishRatePerTopicInBytes(1);
- conf.setBrokerPublisherThrottlingMaxMessageRate(100000);
- conf.setBrokerPublisherThrottlingMaxByteRate(10000000);
- } else {
- conf.setBrokerPublisherThrottlingTickTimeMillis(1);
- conf.setBrokerPublisherThrottlingMaxMessageRate(1);
- conf.setBrokerPublisherThrottlingMaxByteRate(1);
- }
+ conf.setMaxPublishRatePerTopicInMessages(1);
+ conf.setMaxPublishRatePerTopicInBytes(1);
+ conf.setBrokerPublisherThrottlingMaxMessageRate(100000);
+ conf.setBrokerPublisherThrottlingMaxByteRate(10000000);
conf.setStatsUpdateFrequencyInSecs(100000000);
- conf.setPreciseTopicPublishRateLimiterEnable(preciseRateLimit);
setup();
String ns1 = "prop/ns-abc1" + UUID.randomUUID();
admin.namespaces().createNamespace(ns1, 1);
@@ -180,15 +167,9 @@ public class PrometheusMetricsTest extends BrokerTestBase {
assertEquals(item.value, 1);
return;
} else if (item.tags.get("topic").equals(topicName3)) {
- //When using precise rate limiting, we only trigger the rate limiting of the topic,
- // so if the topic is not using the same connection, the rate limiting times will be 0
- //When using asynchronous rate limiting, we will trigger the broker-level rate limiting,
- // and all connections will be limited at this time.
- if (preciseRateLimit) {
- assertEquals(item.value, 0);
- } else {
- assertEquals(item.value, 1);
- }
+ // We only trigger the rate limiting of the topic, so if the topic is not using
+ // the same connection, the rate limiting times will be 0
+ assertEquals(item.value, 0);
return;
}
fail("should not fail");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java
index 4f4affc39d3..360d27f6413 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java
@@ -49,6 +49,7 @@ import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.DispatchRate;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl;
+import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.awaitility.Awaitility;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -66,6 +67,7 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
@BeforeClass
@Override
protected void setup() throws Exception {
+ AsyncTokenBucket.switchToConsistentTokensView();
this.conf.setClusterName("test");
super.internalSetup();
super.producerBaseSetup();
@@ -75,6 +77,7 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
+ AsyncTokenBucket.resetToDefaultEventualConsistentTokensView();
}
@AfterMethod(alwaysRun = true)
@@ -246,20 +249,12 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
// create producer and topic
Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topicName).get();
- boolean isMessageRateUpdate = false;
- int retry = 5;
- for (int i = 0; i < retry; i++) {
- if (topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0
- || topic.getDispatchRateLimiter().get().getDispatchRateOnByte() > 0) {
- isMessageRateUpdate = true;
- break;
- } else {
- if (i != retry - 1) {
- Thread.sleep(100);
- }
- }
- }
- Assert.assertTrue(isMessageRateUpdate);
+
+ Awaitility.await()
+ .ignoreExceptions()
+ .until(() -> topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0
+ || topic.getDispatchRateLimiter().get().getDispatchRateOnByte() > 0);
+
Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate);
int numMessages = 500;
@@ -387,19 +382,11 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
// create producer and topic
Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topicName).get();
- boolean isMessageRateUpdate = false;
- int retry = 5;
- for (int i = 0; i < retry; i++) {
- if (topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0) {
- isMessageRateUpdate = true;
- break;
- } else {
- if (i != retry - 1) {
- Thread.sleep(100);
- }
- }
- }
- Assert.assertTrue(isMessageRateUpdate);
+
+ Awaitility.await()
+ .ignoreExceptions()
+ .until(() -> topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0);
+
Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate);
final int numProducedMessages = 20;
@@ -498,7 +485,7 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
*
* @throws Exception
*/
- @Test(timeOut = 5000)
+ @Test(timeOut = 10000)
public void testRateLimitingMultipleConsumers() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -516,19 +503,11 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
// create producer and topic
Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topicName).get();
- boolean isMessageRateUpdate = false;
- int retry = 5;
- for (int i = 0; i < retry; i++) {
- if (topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0) {
- isMessageRateUpdate = true;
- break;
- } else {
- if (i != retry - 1) {
- Thread.sleep(100);
- }
- }
- }
- Assert.assertTrue(isMessageRateUpdate);
+
+ Awaitility.await()
+ .ignoreExceptions()
+ .until(() -> topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0);
+
Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate);
final int numProducedMessages = 500;
@@ -536,11 +515,17 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
final AtomicInteger totalReceived = new AtomicInteger(0);
ConsumerBuilder<byte[]> consumerBuilder = pulsarClient.newConsumer().topic(topicName)
+ .receiverQueueSize(1)
.subscriptionName("my-subscriber-name").subscriptionType(SubscriptionType.Shared).messageListener((c1, msg) -> {
Assert.assertNotNull(msg, "Message cannot be null");
String receivedMessage = new String(msg.getData());
log.debug("Received message [{}] in the listener", receivedMessage);
totalReceived.incrementAndGet();
+ try {
+ c1.acknowledge(msg);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException(e);
+ }
});
Consumer<byte[]> consumer1 = consumerBuilder.subscribe();
Consumer<byte[]> consumer2 = consumerBuilder.subscribe();
@@ -558,10 +543,10 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
}
// it can make sure that consumer had enough time to consume message but couldn't consume due to throttling
- Thread.sleep(500);
+ Thread.sleep(1000);
- // consumer should not have received all published message due to message-rate throttling
- Assert.assertNotEquals(totalReceived.get(), numProducedMessages);
+ // rate limiter should have limited messages with at least 10% accuracy (or 2 messages if messageRate is low)
+ Assert.assertEquals(totalReceived.get(), messageRate, Math.max(messageRate / 10, 2));
consumer1.close();
consumer2.close();
@@ -593,19 +578,11 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).enableBatching(true)
.batchingMaxPublishDelay(1, TimeUnit.SECONDS).batchingMaxMessages(messagesPerBatch).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topicName).get();
- boolean isMessageRateUpdate = false;
- int retry = 5;
- for (int i = 0; i < retry; i++) {
- if (topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0) {
- isMessageRateUpdate = true;
- break;
- } else {
- if (i != retry - 1) {
- Thread.sleep(100);
- }
- }
- }
- Assert.assertTrue(isMessageRateUpdate);
+
+ Awaitility.await()
+ .ignoreExceptions()
+ .until(() -> topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0);
+
Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate);
final AtomicInteger totalReceived = new AtomicInteger(0);
@@ -732,20 +709,12 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
// create producer and topic
Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topicName).get();
- boolean isMessageRateUpdate = false;
- int retry = 5;
- for (int i = 0; i < retry; i++) {
- if (topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0
- && topic.getDispatchRateLimiter().get().getDispatchRateOnByte() > 0) {
- isMessageRateUpdate = true;
- break;
- } else {
- if (i != retry - 1) {
- Thread.sleep(100);
- }
- }
- }
- Assert.assertTrue(isMessageRateUpdate);
+
+ Awaitility.await()
+ .ignoreExceptions()
+ .until(() -> topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0
+ || topic.getDispatchRateLimiter().get().getDispatchRateOnByte() > 0);
+
Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate);
final int numProducedMessages = 200;
@@ -813,20 +782,12 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
// create producer and topic
Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topicName).get();
- boolean isMessageRateUpdate = false;
- int retry = 5;
- for (int i = 0; i < retry; i++) {
- if (topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0
- || topic.getDispatchRateLimiter().get().getDispatchRateOnByte() > 0) {
- isMessageRateUpdate = true;
- break;
- } else {
- if (i != retry - 1) {
- Thread.sleep(100);
- }
- }
- }
- Assert.assertTrue(isMessageRateUpdate);
+
+ Awaitility.await()
+ .ignoreExceptions()
+ .until(() -> topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0
+ || topic.getDispatchRateLimiter().get().getDispatchRateOnByte() > 0);
+
Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate);
int numMessages = 500;
@@ -1148,19 +1109,11 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
// create producer and topic
Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).enableBatching(false).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topicName).get();
- boolean isMessageRateUpdate = false;
- int retry = 10;
- for (int i = 0; i < retry; i++) {
- if (topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0) {
- isMessageRateUpdate = true;
- break;
- } else {
- if (i != retry - 1) {
- Thread.sleep(100);
- }
- }
- }
- Assert.assertTrue(isMessageRateUpdate);
+
+ Awaitility.await()
+ .ignoreExceptions()
+ .until(() -> topic.getDispatchRateLimiter().get().getDispatchRateOnMsg() > 0);
+
Assert.assertEquals(admin.namespaces().getDispatchRate(namespace), dispatchRate);
Thread.sleep(2000);
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 ad1955f08d7..1c0ae5547d5 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
@@ -18,9 +18,6 @@
*/
package org.apache.pulsar.client.impl;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertNotEquals;
-import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertTrue;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -34,14 +31,12 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.Cleanup;
import org.apache.pulsar.broker.service.Producer;
-import org.apache.pulsar.broker.service.PublishRateLimiter;
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.awaitility.Awaitility;
+import org.apache.pulsar.broker.qos.AsyncTokenBucket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -53,6 +48,7 @@ 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);
@@ -64,6 +60,7 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
+ AsyncTokenBucket.resetToDefaultEventualConsistentTokensView();
}
/**
@@ -86,16 +83,9 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer().topic(topicName)
.maxPendingMessages(30000).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topicName).get().get();
- // (1) verify message-rate is -1 initially
- Assert.assertEquals(topic.getTopicPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
// enable throttling
admin.namespaces().setPublishRate(namespace, publishMsgRate);
- retryStrategically((test) ->
- !topic.getTopicPublishRateLimiter().equals(PublishRateLimiter.DISABLED_RATE_LIMITER),
- 5,
- 200);
- Assert.assertNotEquals(topic.getTopicPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
Producer prod = topic.getProducers().values().iterator().next();
// reset counter
@@ -112,11 +102,6 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
// disable throttling
publishMsgRate.publishThrottlingRateInMsg = -1;
admin.namespaces().setPublishRate(namespace, publishMsgRate);
- retryStrategically((test) ->
- topic.getTopicPublishRateLimiter().equals(PublishRateLimiter.DISABLED_RATE_LIMITER),
- 5,
- 200);
- Assert.assertEquals(topic.getTopicPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
// reset counter
prod.updateRates();
@@ -150,16 +135,9 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
// create producer and topic
ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer().topic(topicName).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topicName).get();
- // (1) verify message-rate is -1 initially
- Assert.assertEquals(topic.getTopicPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
// enable throttling
admin.namespaces().setPublishRate(namespace, publishMsgRate);
- retryStrategically((test) ->
- !topic.getTopicPublishRateLimiter().equals(PublishRateLimiter.DISABLED_RATE_LIMITER),
- 5,
- 200);
- Assert.assertNotEquals(topic.getTopicPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
Producer prod = topic.getProducers().values().iterator().next();
// reset counter
@@ -176,9 +154,6 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
// disable throttling
publishMsgRate.publishThrottlingRateInByte = -1;
admin.namespaces().setPublishRate(namespace, publishMsgRate);
- retryStrategically((test) -> topic.getTopicPublishRateLimiter().equals(PublishRateLimiter.DISABLED_RATE_LIMITER), 5,
- 200);
- Assert.assertEquals(topic.getTopicPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
// reset counter
prod.updateRates();
@@ -214,8 +189,6 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
.enableBatching(false)
.maxPendingMessages(30000).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topicName).get().get();
- // (1) verify message-rate is -1 initially
- Assert.assertEquals(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
// enable throttling
admin.brokers().
@@ -223,19 +196,11 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
"brokerPublisherThrottlingMaxMessageRate",
Integer.toString(messageRate));
- retryStrategically(
- (test) ->
- (topic.getBrokerPublishRateLimiter() != PublishRateLimiter.DISABLED_RATE_LIMITER),
- 5,
- 200);
-
log.info("Get broker configuration: brokerTick {}, MaxMessageRate {}, MaxByteRate {}",
pulsar.getConfiguration().getBrokerPublisherThrottlingTickTimeMillis(),
pulsar.getConfiguration().getBrokerPublisherThrottlingMaxMessageRate(),
pulsar.getConfiguration().getBrokerPublisherThrottlingMaxByteRate());
- Assert.assertNotEquals(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
-
Producer prod = topic.getProducers().values().iterator().next();
// reset counter
prod.updateRates();
@@ -252,11 +217,6 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
// disable throttling
admin.brokers()
.updateDynamicConfiguration("brokerPublisherThrottlingMaxMessageRate", Integer.toString(0));
- retryStrategically((test) ->
- topic.getBrokerPublishRateLimiter().equals(PublishRateLimiter.DISABLED_RATE_LIMITER),
- 5,
- 200);
- Assert.assertEquals(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
// reset counter
prod.updateRates();
@@ -293,26 +253,16 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
.enableBatching(false)
.maxPendingMessages(30000).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topicName).get().get();
- // (1) verify byte-rate is -1 disabled
- Assert.assertEquals(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
// enable throttling
admin.brokers()
.updateDynamicConfiguration("brokerPublisherThrottlingMaxByteRate", Long.toString(byteRate));
- retryStrategically(
- (test) ->
- (topic.getBrokerPublishRateLimiter() != PublishRateLimiter.DISABLED_RATE_LIMITER),
- 5,
- 200);
-
log.info("Get broker configuration after enable: brokerTick {}, MaxMessageRate {}, MaxByteRate {}",
pulsar.getConfiguration().getBrokerPublisherThrottlingTickTimeMillis(),
pulsar.getConfiguration().getBrokerPublisherThrottlingMaxMessageRate(),
pulsar.getConfiguration().getBrokerPublisherThrottlingMaxByteRate());
- Assert.assertNotEquals(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
-
Producer prod = topic.getProducers().values().iterator().next();
// reset counter
prod.updateRates();
@@ -331,18 +281,12 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
// disable throttling
admin.brokers()
.updateDynamicConfiguration("brokerPublisherThrottlingMaxByteRate", Long.toString(0));
- retryStrategically((test) ->
- topic.getBrokerPublishRateLimiter().equals(PublishRateLimiter.DISABLED_RATE_LIMITER),
- 5,
- 200);
log.info("Get broker configuration after disable: brokerTick {}, MaxMessageRate {}, MaxByteRate {}",
pulsar.getConfiguration().getBrokerPublisherThrottlingTickTimeMillis(),
pulsar.getConfiguration().getBrokerPublisherThrottlingMaxMessageRate(),
pulsar.getConfiguration().getBrokerPublisherThrottlingMaxByteRate());
- Assert.assertEquals(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
-
// reset counter
prod.updateRates();
for (int i = 0; i < numMessage; i++) {
@@ -385,21 +329,12 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
.enableBatching(false)
.maxPendingMessages(30000).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topicName).get().get();
- // (1) verify both broker and topic limiter is disabled
- Assert.assertEquals(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
- Assert.assertEquals(topic.getTopicPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
// enable broker and topic throttling
admin.namespaces().setPublishRate(namespace, topicPublishMsgRate);
- Awaitility.await().untilAsserted(() -> {
- assertNotEquals(topic.getTopicPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
- });
admin.brokers().updateDynamicConfiguration("brokerPublisherThrottlingMaxByteRate",
Long.toString(brokerByteRate));
- Awaitility.await().untilAsserted(() -> {
- assertNotSame(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
- });
log.info("Get broker configuration after enable: brokerTick {}, MaxMessageRate {}, MaxByteRate {}",
pulsar.getConfiguration().getBrokerPublisherThrottlingTickTimeMillis(),
@@ -440,15 +375,7 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
producers.add(iProducer);
topics.add(iTopic);
- // verify both broker and topic limiter is enabled
- Assert.assertNotEquals(iTopic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
-
admin.namespaces().setPublishRate(namespace, topicPublishMsgRate);
- retryStrategically((test) ->
- !iTopic.getTopicPublishRateLimiter().equals(PublishRateLimiter.DISABLED_RATE_LIMITER),
- 5,
- 200);
- Assert.assertNotEquals(iTopic.getTopicPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
}
List<Callable<Void>> topicRatesCounter = Lists.newArrayListWithExpectedSize(3);
@@ -486,10 +413,6 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
topicPublishMsgRate.publishThrottlingRateInByte = -1;
admin.namespaces().setPublishRate(namespace, topicPublishMsgRate);
- Awaitility.await().untilAsserted(() ->
- assertEquals(topic.getTopicPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER)
- );
-
// reset counter
prod.updateRates();
for (int i = 0; i < numMessage; i++) {
@@ -505,18 +428,12 @@ public class MessagePublishThrottlingTest extends ProducerConsumerBase {
// disable broker throttling, expected no throttling.
admin.brokers()
.updateDynamicConfiguration("brokerPublisherThrottlingMaxByteRate", Long.toString(0));
- retryStrategically((test) ->
- topic.getBrokerPublishRateLimiter().equals(PublishRateLimiter.DISABLED_RATE_LIMITER),
- 5,
- 200);
log.info("Get broker configuration after disable: brokerTick {}, MaxMessageRate {}, MaxByteRate {}",
pulsar.getConfiguration().getBrokerPublisherThrottlingTickTimeMillis(),
pulsar.getConfiguration().getBrokerPublisherThrottlingMaxMessageRate(),
pulsar.getConfiguration().getBrokerPublisherThrottlingMaxByteRate());
- Assert.assertEquals(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
-
// reset counter
prod.updateRates();
for (int i = 0; i < numMessage; i++) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicPublishThrottlingInitTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicPublishThrottlingInitTest.java
index 1b8284c31e6..fe2e6ec9670 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicPublishThrottlingInitTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicPublishThrottlingInitTest.java
@@ -21,13 +21,10 @@ package org.apache.pulsar.client.impl;
import static org.testng.Assert.assertTrue;
import com.google.common.collect.Sets;
import org.apache.pulsar.broker.service.Producer;
-import org.apache.pulsar.broker.service.PublishRateLimiter;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.api.ProducerConsumerBase;
-import org.awaitility.Awaitility;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -74,16 +71,12 @@ public class TopicPublishThrottlingInitTest extends ProducerConsumerBase {
.enableBatching(false)
.maxPendingMessages(30000).create();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topicName).get().get();
- // (1) verify message-rate is initialized when value configured in broker
- Assert.assertNotEquals(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
log.info("Get broker configuration: brokerTick {}, MaxMessageRate {}, MaxByteRate {}",
pulsar.getConfiguration().getBrokerPublisherThrottlingTickTimeMillis(),
pulsar.getConfiguration().getBrokerPublisherThrottlingMaxMessageRate(),
pulsar.getConfiguration().getBrokerPublisherThrottlingMaxByteRate());
- Assert.assertNotEquals(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER);
-
Producer prod = topic.getProducers().values().iterator().next();
// reset counter
prod.updateRates();
@@ -100,8 +93,6 @@ public class TopicPublishThrottlingInitTest extends ProducerConsumerBase {
// disable throttling
admin.brokers()
.updateDynamicConfiguration("brokerPublisherThrottlingMaxMessageRate", Integer.toString(0));
- Awaitility.await().untilAsserted(() ->
- Assert.assertEquals(topic.getBrokerPublishRateLimiter(), PublishRateLimiter.DISABLED_RATE_LIMITER));
// reset counter
prod.updateRates();
diff --git a/pulsar-broker/src/test/resources/prometheus_metrics_sample.txt b/pulsar-broker/src/test/resources/prometheus_metrics_sample.txt
index 2022fbd8000..7cf8d3e7167 100644
--- a/pulsar-broker/src/test/resources/prometheus_metrics_sample.txt
+++ b/pulsar-broker/src/test/resources/prometheus_metrics_sample.txt
@@ -284,8 +284,6 @@ caffeine_cache_load_duration_seconds_count{cluster="use",cache="bookies-racks-da
caffeine_cache_load_duration_seconds_sum{cluster="use",cache="bookies-racks-data"} 0.0
caffeine_cache_load_duration_seconds_count{cluster="use",cache="global-zk-exists"} 0.0
caffeine_cache_load_duration_seconds_sum{cluster="use",cache="global-zk-exists"} 0.0
-# TYPE pulsar_broker_throttled_connections_global_limit gauge
-pulsar_broker_throttled_connections_global_limit{cluster="use"} 0.0
# TYPE process_cpu_seconds_total counter
process_cpu_seconds_total{cluster="use"} 101.68
# TYPE process_start_time_seconds gauge
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
deleted file mode 100644
index 4ecb29b2462..00000000000
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java
+++ /dev/null
@@ -1,286 +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.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;
-
-/**
- * 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>
- */
-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) {
- checkArgument(!isClosed(), "Rate limiter is already shutdown");
- // 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
deleted file mode 100644
index 3738027c635..00000000000
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/RateLimiterTest.java
+++ /dev/null
@@ -1,248 +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.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 lombok.Cleanup;
-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;
- @Cleanup
- 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;
- @Cleanup
- 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());
- }
-
-}
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/FunctionStatsManager.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/FunctionStatsManager.java
index 1bb46da9472..8737c8a4fa9 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/FunctionStatsManager.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/FunctionStatsManager.java
@@ -19,16 +19,15 @@
package org.apache.pulsar.functions.instance.stats;
import com.google.common.collect.EvictingQueue;
+import com.google.common.util.concurrent.RateLimiter;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import io.prometheus.client.Summary;
import java.util.Arrays;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.common.util.RateLimiter;
import org.apache.pulsar.functions.proto.InstanceCommunication;
/**
@@ -262,18 +261,8 @@ public class FunctionStatsManager extends ComponentStatsManager {
.help("Exception from sink.")
.create());
- userExceptionRateLimiter = RateLimiter.builder()
- .scheduledExecutorService(scheduledExecutorService)
- .permits(5)
- .rateTime(1)
- .timeUnit(TimeUnit.MINUTES)
- .build();
- sysExceptionRateLimiter = RateLimiter.builder()
- .scheduledExecutorService(scheduledExecutorService)
- .permits(5)
- .rateTime(1)
- .timeUnit(TimeUnit.MINUTES)
- .build();
+ userExceptionRateLimiter = RateLimiter.create(5.0d / 60.0d);
+ sysExceptionRateLimiter = RateLimiter.create(5.0d / 60.0d);
}
public void addUserException(Throwable ex) {
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SinkStatsManager.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SinkStatsManager.java
index 779a56cdf61..c515ce6bc87 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SinkStatsManager.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SinkStatsManager.java
@@ -19,13 +19,12 @@
package org.apache.pulsar.functions.instance.stats;
import com.google.common.collect.EvictingQueue;
+import com.google.common.util.concurrent.RateLimiter;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import java.util.Arrays;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
import lombok.Getter;
-import org.apache.pulsar.common.util.RateLimiter;
import org.apache.pulsar.functions.proto.InstanceCommunication;
public class SinkStatsManager extends ComponentStatsManager {
@@ -196,18 +195,8 @@ public class SinkStatsManager extends ComponentStatsManager {
.help("Exception from sink.")
.create());
- sysExceptionRateLimiter = RateLimiter.builder()
- .scheduledExecutorService(scheduledExecutorService)
- .permits(5)
- .rateTime(1)
- .timeUnit(TimeUnit.MINUTES)
- .build();
- sinkExceptionRateLimiter = RateLimiter.builder()
- .scheduledExecutorService(scheduledExecutorService)
- .permits(5)
- .rateTime(1)
- .timeUnit(TimeUnit.MINUTES)
- .build();
+ sysExceptionRateLimiter = RateLimiter.create(5.0d / 60.0d);
+ sinkExceptionRateLimiter = RateLimiter.create(5.0d / 60.0d);
}
@Override
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SourceStatsManager.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SourceStatsManager.java
index 4470310c6c3..1f7e159c4dc 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SourceStatsManager.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SourceStatsManager.java
@@ -19,13 +19,12 @@
package org.apache.pulsar.functions.instance.stats;
import com.google.common.collect.EvictingQueue;
+import com.google.common.util.concurrent.RateLimiter;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import java.util.Arrays;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
import lombok.Getter;
-import org.apache.pulsar.common.util.RateLimiter;
import org.apache.pulsar.functions.proto.InstanceCommunication;
public class SourceStatsManager extends ComponentStatsManager {
@@ -196,18 +195,8 @@ public class SourceStatsManager extends ComponentStatsManager {
.help("Exception from source.")
.create());
- sysExceptionRateLimiter = RateLimiter.builder()
- .scheduledExecutorService(scheduledExecutorService)
- .permits(5)
- .rateTime(1)
- .timeUnit(TimeUnit.MINUTES)
- .build();
- sourceExceptionRateLimiter = RateLimiter.builder()
- .scheduledExecutorService(scheduledExecutorService)
- .permits(5)
- .rateTime(1)
- .timeUnit(TimeUnit.MINUTES)
- .build();
+ sysExceptionRateLimiter = RateLimiter.create(5.0d / 60.0d);
+ sourceExceptionRateLimiter = RateLimiter.create(5.0d / 60.0d);
}
@Override