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 2021/12/11 07:18:09 UTC
[pulsar] branch branch-2.7 updated: Remove RateLimiter constructors and replace with builder usage (#11599)
This is an automated email from the ASF dual-hosted git repository.
mmarshall pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.7 by this push:
new 0c48d23 Remove RateLimiter constructors and replace with builder usage (#11599)
0c48d23 is described below
commit 0c48d2351006cac3ff99ec1a2f0aa774daed9d0f
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Aug 9 19:32:09 2021 +0300
Remove RateLimiter constructors and replace with builder usage (#11599)
(cherry picked from commit fcd3336c083f359e9f5a5d624edffb21df62d9d1)
---
.../service/persistent/DispatchRateLimiter.java | 22 +++++++++---
.../service/persistent/SubscribeRateLimiter.java | 8 +++--
.../org/apache/pulsar/common/util/RateLimiter.java | 36 -------------------
.../apache/pulsar/common/util/RateLimiterTest.java | 41 ++++++++++++++--------
.../instance/stats/FunctionStatsManager.java | 14 ++++++--
.../functions/instance/stats/SinkStatsManager.java | 14 ++++++--
.../instance/stats/SourceStatsManager.java | 14 ++++++--
7 files changed, 87 insertions(+), 62 deletions(-)
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 2899966..97048df 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
@@ -333,8 +333,15 @@ public class DispatchRateLimiter {
// update msg-rateLimiter
if (msgRate > 0) {
if (this.dispatchRateLimiterOnMessage == null) {
- this.dispatchRateLimiterOnMessage = new RateLimiter(brokerService.pulsar().getExecutor(), msgRate,
- ratePeriod, TimeUnit.SECONDS, permitUpdaterMsg, true);
+ this.dispatchRateLimiterOnMessage =
+ RateLimiter.builder()
+ .scheduledExecutorService(brokerService.pulsar().getExecutor())
+ .permits(msgRate)
+ .rateTime(ratePeriod)
+ .timeUnit(TimeUnit.SECONDS)
+ .permitUpdater(permitUpdaterMsg)
+ .isDispatchOrPrecisePublishRateLimiter(true)
+ .build();
} else {
this.dispatchRateLimiterOnMessage.setRate(msgRate, dispatchRate.ratePeriodInSecond,
TimeUnit.SECONDS, permitUpdaterMsg);
@@ -353,8 +360,15 @@ public class DispatchRateLimiter {
// update byte-rateLimiter
if (byteRate > 0) {
if (this.dispatchRateLimiterOnByte == null) {
- this.dispatchRateLimiterOnByte = new RateLimiter(brokerService.pulsar().getExecutor(), byteRate,
- ratePeriod, TimeUnit.SECONDS, permitUpdaterByte, true);
+ this.dispatchRateLimiterOnByte =
+ RateLimiter.builder()
+ .scheduledExecutorService(brokerService.pulsar().getExecutor())
+ .permits(byteRate)
+ .rateTime(ratePeriod)
+ .timeUnit(TimeUnit.SECONDS)
+ .permitUpdater(permitUpdaterByte)
+ .isDispatchOrPrecisePublishRateLimiter(true)
+ .build();
} else {
this.dispatchRateLimiterOnByte.setRate(byteRate, dispatchRate.ratePeriodInSecond,
TimeUnit.SECONDS, permitUpdaterByte);
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 32fe834..51862b1 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
@@ -135,8 +135,12 @@ public class SubscribeRateLimiter {
if (ratePerConsumer > 0) {
if (this.subscribeRateLimiter.get(consumerIdentifier) == null) {
this.subscribeRateLimiter.put(consumerIdentifier,
- new RateLimiter(brokerService.pulsar().getExecutor(), ratePerConsumer,
- ratePeriod, TimeUnit.SECONDS));
+ 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);
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
index 0c22edf..48196bb 100644
--- 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
@@ -64,42 +64,6 @@ public class RateLimiter implements AutoCloseable{
private RateLimitFunction rateLimitFunction;
private boolean isDispatchOrPrecisePublishRateLimiter;
- public RateLimiter(final long permits, final long rateTime, final TimeUnit timeUnit) {
- this(null, permits, rateTime, timeUnit);
- }
-
- public RateLimiter(final long permits, final long rateTime, final TimeUnit timeUnit, boolean isDispatchOrPrecisePublishRateLimiter) {
- this(null, permits, rateTime, timeUnit, null, isDispatchOrPrecisePublishRateLimiter);
- }
-
- public RateLimiter(final long permits, final long rateTime, final TimeUnit timeUnit,
- RateLimitFunction autoReadResetFunction) {
- this(null, permits, rateTime, timeUnit, null, false);
- this.rateLimitFunction = autoReadResetFunction;
- }
-
- public RateLimiter(final long permits, final long rateTime, final TimeUnit timeUnit,
- RateLimitFunction autoReadResetFunction, boolean isDispatchOrPrecisePublishRateLimiter) {
- this(null, permits, rateTime, timeUnit, null, isDispatchOrPrecisePublishRateLimiter);
- this.rateLimitFunction = autoReadResetFunction;
- }
-
- public RateLimiter(final ScheduledExecutorService service, final long permits, final long rateTime,
- final TimeUnit timeUnit) {
- this(service, permits, rateTime, timeUnit, (Supplier<Long>) null);
- }
-
- public RateLimiter(final ScheduledExecutorService service, final long permits, final long rateTime,
- final TimeUnit timeUnit, Supplier<Long> permitUpdater) {
- this(service, permits, rateTime, timeUnit, permitUpdater, false);
- }
-
- public RateLimiter(final ScheduledExecutorService service, final long permits, final long rateTime,
- final TimeUnit timeUnit, Supplier<Long> permitUpdater, boolean isDispatchOrPrecisePublishRateLimiter) {
- this(service, permits, rateTime, timeUnit, permitUpdater, isDispatchOrPrecisePublishRateLimiter,
- null);
- }
-
@Builder
RateLimiter(final ScheduledExecutorService scheduledExecutorService, final long permits, final long rateTime,
final TimeUnit timeUnit, Supplier<Long> permitUpdater, boolean isDispatchOrPrecisePublishRateLimiter,
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
index 0d37bee..968abed 100644
--- 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
@@ -33,14 +33,14 @@ public class RateLimiterTest {
@Test
public void testInvalidRenewTime() {
try {
- new RateLimiter(0, 100, TimeUnit.SECONDS);
+ 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 {
- new RateLimiter(10, 0, TimeUnit.SECONDS);
+ RateLimiter.builder().permits(10).rateTime(0).timeUnit(TimeUnit.SECONDS).build();
fail("should have thrown exception: invalid rateTime, must be > 0");
} catch (IllegalArgumentException ie) {
// Ok
@@ -49,7 +49,7 @@ public class RateLimiterTest {
@Test
public void testClose() throws Exception {
- RateLimiter rate = new RateLimiter(1, 1000, TimeUnit.MILLISECONDS);
+ RateLimiter rate = RateLimiter.builder().permits(1).rateTime(1000).timeUnit(TimeUnit.MILLISECONDS).build();
assertFalse(rate.isClosed());
rate.close();
assertTrue(rate.isClosed());
@@ -64,7 +64,8 @@ public class RateLimiterTest {
@Test
public void testAcquireBlock() throws Exception {
final long rateTimeMSec = 1000;
- RateLimiter rate = new RateLimiter(1, rateTimeMSec, TimeUnit.MILLISECONDS);
+ RateLimiter rate = RateLimiter.builder().permits(1).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
rate.acquire();
assertEquals(rate.getAvailablePermits(), 0);
long start = System.currentTimeMillis();
@@ -79,7 +80,8 @@ public class RateLimiterTest {
public void testAcquire() throws Exception {
final long rateTimeMSec = 1000;
final int permits = 100;
- RateLimiter rate = new RateLimiter(permits, rateTimeMSec, TimeUnit.MILLISECONDS);
+ 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();
@@ -95,7 +97,8 @@ public class RateLimiterTest {
final long rateTimeMSec = 1000;
final int permits = 100;
final int acquirePermits = 50;
- RateLimiter rate = new RateLimiter(permits, rateTimeMSec, TimeUnit.MILLISECONDS);
+ 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);
@@ -109,7 +112,8 @@ public class RateLimiterTest {
@Test
public void testTryAcquireNoPermits() {
final long rateTimeMSec = 1000;
- RateLimiter rate = new RateLimiter(1, rateTimeMSec, TimeUnit.MILLISECONDS);
+ RateLimiter rate = RateLimiter.builder().permits(1).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
assertTrue(rate.tryAcquire());
assertFalse(rate.tryAcquire());
assertEquals(rate.getAvailablePermits(), 0);
@@ -120,7 +124,8 @@ public class RateLimiterTest {
public void testTryAcquire() {
final long rateTimeMSec = 1000;
final int permits = 100;
- RateLimiter rate = new RateLimiter(permits, rateTimeMSec, TimeUnit.MILLISECONDS);
+ RateLimiter rate = RateLimiter.builder().permits(permits).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
for (int i = 0; i < permits; i++) {
rate.tryAcquire();
}
@@ -133,7 +138,8 @@ public class RateLimiterTest {
final long rateTimeMSec = 1000;
final int permits = 100;
final int acquirePermits = 50;
- RateLimiter rate = new RateLimiter(permits, rateTimeMSec, TimeUnit.MILLISECONDS);
+ RateLimiter rate = RateLimiter.builder().permits(permits).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .build();
for (int i = 0; i < permits / acquirePermits; i++) {
rate.tryAcquire(acquirePermits);
}
@@ -145,7 +151,8 @@ public class RateLimiterTest {
public void testResetRate() throws Exception {
final long rateTimeMSec = 1000;
final int permits = 100;
- RateLimiter rate = new RateLimiter(permits, rateTimeMSec, TimeUnit.MILLISECONDS);
+ 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
@@ -168,7 +175,9 @@ public class RateLimiterTest {
public void testDispatchRate() throws Exception {
final long rateTimeMSec = 1000;
final int permits = 100;
- RateLimiter rate = new RateLimiter(null, permits, rateTimeMSec, TimeUnit.MILLISECONDS, null, true);
+ RateLimiter rate = RateLimiter.builder().permits(permits).rateTime(rateTimeMSec).timeUnit(TimeUnit.MILLISECONDS)
+ .isDispatchOrPrecisePublishRateLimiter(true)
+ .build();
rate.tryAcquire(100);
rate.tryAcquire(100);
rate.tryAcquire(100);
@@ -191,7 +200,9 @@ public class RateLimiterTest {
long rateTime = 1;
long newUpdatedRateLimit = 100L;
Supplier<Long> permitUpdater = () -> newUpdatedRateLimit;
- RateLimiter limiter = new RateLimiter(null, permits, 1, TimeUnit.SECONDS, permitUpdater);
+ 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);
@@ -204,8 +215,10 @@ public class RateLimiterTest {
long rateTime = 1;
int reNewTime = 3;
RateLimitFunction rateLimitFunction = atomicInteger::incrementAndGet;
- RateLimiter rateLimiter = new RateLimiter(permits, rateTime, TimeUnit.SECONDS, rateLimitFunction);
- for (int i = 0 ; i < reNewTime; i++) {
+ 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 7e6e379c..f5a9586 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
@@ -232,8 +232,18 @@ public class FunctionStatsManager extends ComponentStatsManager{
.help("Exception from sink.")
.register(collectorRegistry);
- userExceptionRateLimiter = new RateLimiter(scheduledExecutorService, 5, 1, TimeUnit.MINUTES);
- sysExceptionRateLimiter = new RateLimiter(scheduledExecutorService, 5, 1, TimeUnit.MINUTES);
+ 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();
}
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 5c28d43..98b8897 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
@@ -174,8 +174,18 @@ public class SinkStatsManager extends ComponentStatsManager {
.help("Exception from sink.")
.register(collectorRegistry);
- sysExceptionRateLimiter = new RateLimiter(scheduledExecutorService, 5, 1, TimeUnit.MINUTES);
- sinkExceptionRateLimiter = new RateLimiter(scheduledExecutorService, 5, 1, TimeUnit.MINUTES);
+ 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();
}
@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 0baaa2e..74f4999 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
@@ -173,8 +173,18 @@ public class SourceStatsManager extends ComponentStatsManager {
.help("Exception from source.")
.register(collectorRegistry);
- sysExceptionRateLimiter = new RateLimiter(scheduledExecutorService, 5, 1, TimeUnit.MINUTES);
- sourceExceptionRateLimiter = new RateLimiter(scheduledExecutorService, 5, 1, TimeUnit.MINUTES);
+ 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();
}
@Override