Make PublishRateLimiter extend AutoCloseable

- addresses review comment
apache · May 17, 2021 · ac41481 · ac41481
1 parent bfa4914
commit ac41481
Show file tree

Hide file tree

Showing 6 changed files with 15 additions and 6 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PrecisPublishLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PrecisPublishLimiter.java
@@ -25,7 +25,7 @@
 import org.apache.pulsar.common.util.RateLimitFunction;
 import org.apache.pulsar.common.util.RateLimiter;
 
-public class PrecisPublishLimiter implements PublishRateLimiter, AutoCloseable {
+public class PrecisPublishLimiter implements PublishRateLimiter {
     protected volatile int publishMaxMessageRate = 0;
     protected volatile long publishMaxByteRate = 0;
     protected volatile boolean publishThrottlingEnabled = false;

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
@@ -21,7 +21,7 @@
 import org.apache.pulsar.common.policies.data.Policies;
 import org.apache.pulsar.common.policies.data.PublishRate;
 
-public interface PublishRateLimiter {
+public interface PublishRateLimiter extends AutoCloseable {
 
     PublishRateLimiter DISABLED_RATE_LIMITER = PublishRateLimiterDisable.DISABLED_RATE_LIMITER;
 

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
@@ -62,4 +62,8 @@ public boolean tryAcquire(int numbers, long bytes) {
         return false;
     }
 
+    @Override
+    public void close() throws Exception {
+        // 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
@@ -108,4 +108,9 @@ public void update(PublishRate maxPublishRate) {
     public boolean tryAcquire(int numbers, long bytes) {
         return false;
     }
+
+    @Override
+    public void close() throws Exception {
+        // no-op
+    }
 }
diff --git a/...oker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/...oker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -438,9 +438,9 @@ public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect
 
         replicators.forEach((cluster, replicator) -> futures.add(replicator.disconnect()));
         producers.values().forEach(producer -> futures.add(producer.disconnect()));
-        if (topicPublishRateLimiter instanceof AutoCloseable) {
+        if (topicPublishRateLimiter != null) {
             try {
-                ((AutoCloseable) topicPublishRateLimiter).close();
+                topicPublishRateLimiter.close();
             } catch (Exception e) {
                 log.warn("Error closing topicPublishRateLimiter for topic {}", topic, e);
             }

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
@@ -1155,9 +1155,9 @@ public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect
         futures.add(transactionBuffer.closeAsync());
         replicators.forEach((cluster, replicator) -> futures.add(replicator.disconnect()));
         producers.values().forEach(producer -> futures.add(producer.disconnect()));
-        if (topicPublishRateLimiter instanceof AutoCloseable) {
+        if (topicPublishRateLimiter != null) {
             try {
-                ((AutoCloseable) topicPublishRateLimiter).close();
+                topicPublishRateLimiter.close();
             } catch (Exception e) {
                 log.warn("Error closing topicPublishRateLimiter for topic {}", topic, e);
             }