remove isTimeBacklogExceeded

Author: Jason918

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -1715,14 +1715,22 @@ public synchronized void monitorBacklogQuota() {
                 if (persistentTopic.isSizeBacklogExceeded()) {
                     getBacklogQuotaManager().handleExceededBacklogQuota(persistentTopic,
                             BacklogQuota.BacklogQuotaType.destination_storage, false);
-                } else if (persistentTopic.isTimeBacklogExceeded()) {
-                    getBacklogQuotaManager().handleExceededBacklogQuota(persistentTopic,
-                            BacklogQuota.BacklogQuotaType.message_age,
-                            pulsar.getConfiguration().isPreciseTimeBasedBacklogQuotaCheck());
                 } else {
-                    if (log.isDebugEnabled()) {
-                        log.debug("quota not exceeded for [{}]", topic.getName());
-                    }
+                    persistentTopic.checkTimeBacklogExceeded().thenAccept(isExceeded -> {
+                        if (isExceeded) {
+                            getBacklogQuotaManager().handleExceededBacklogQuota(persistentTopic,
+                                    BacklogQuota.BacklogQuotaType.message_age,
+                                    pulsar.getConfiguration().isPreciseTimeBasedBacklogQuotaCheck());
+                        } else {
+                            if (log.isDebugEnabled()) {
+                                log.debug("quota not exceeded for [{}]", topic.getName());
+                            }
+                        }
+                    }).exceptionally(throwable -> {
+                        log.error("Error when checkTimeBacklogExceeded({}) in monitorBacklogQuota",
+                                persistentTopic.getName(), throwable);
+                        return null;
+                    });
                 }
             }
         });

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -2561,15 +2561,6 @@ public boolean isSizeBacklogExceeded() {
     /**
      * @return determine if backlog quota enforcement needs to be done for topic based on time limit
      */
-    public boolean isTimeBacklogExceeded() {
-        try {
-            return checkTimeBacklogExceeded().get();
-        } catch (Throwable e) {
-            log.error("[{}] checkTimeBacklogExceeded failed.", topic, e);
-            return false;
-        }
-    }
-
     public CompletableFuture<Boolean> checkTimeBacklogExceeded() {
         TopicName topicName = TopicName.get(getName());
         int backlogQuotaLimitInSecond = getBacklogQuota(BacklogQuotaType.message_age).getLimitTime();

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SystemTopic.java

@@ -36,8 +36,8 @@ public boolean isSizeBacklogExceeded() {
     }
 
     @Override
-    public boolean isTimeBacklogExceeded() {
-        return false;
+    public CompletableFuture<Boolean> checkTimeBacklogExceeded() {
+        return CompletableFuture.completedFuture(false);
     }
 
     @Override