[ISSUE #9313] Add scheduled clean task. (#9314)

* add scheduled clean task.

* make code compatible to lmq.

* make code compatible to lmq.

* make code compatible to lmq.

* make code compatible to lmq.

Author: GenerousMan

broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java

@@ -691,6 +691,18 @@ public void run() {
             }
         }, 10, 1, TimeUnit.SECONDS);
 
+        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    BrokerController.this.messageStore.getTimerMessageStore().getTimerMetrics()
+                            .cleanMetrics(BrokerController.this.topicConfigManager.getTopicConfigTable().keySet());
+                } catch (Throwable e) {
+                    LOG.error("BrokerController: failed to clean unused timer metrics.", e);
+                }
+            }
+        }, 3, 3, TimeUnit.MINUTES);
+
         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
 
             @Override

broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java

@@ -825,6 +825,7 @@ private void deleteTopicInBroker(String topic) {
         this.brokerController.getConsumerOffsetManager().cleanOffsetByTopic(topic);
         this.brokerController.getPopInflightMessageCounter().clearInFlightMessageNumByTopicName(topic);
         this.brokerController.getMessageStore().deleteTopics(Sets.newHashSet(topic));
+        this.brokerController.getMessageStore().getTimerMessageStore().getTimerMetrics().removeTimingCount(topic);
     }
 
     private synchronized RemotingCommand updateAndCreateAccessConfig(ChannelHandlerContext ctx, RemotingCommand request) {

broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java

@@ -259,6 +259,8 @@ public void init() throws Exception {
 
         brokerController.getTopicConfigManager().getTopicConfigTable().put(topic, new TopicConfig(topic));
         brokerController.getMessageStoreConfig().setTimerWheelEnable(false);
+        when(this.brokerController.getMessageStore().getTimerMessageStore()).thenReturn(timerMessageStore);
+        when(this.timerMessageStore.getTimerMetrics()).thenReturn(timerMetrics);
     }
 
     @After

store/src/main/java/org/apache/rocketmq/store/timer/TimerMetrics.java

@@ -37,6 +37,7 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import org.apache.rocketmq.common.ConfigManager;
+import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.message.MessageExt;
@@ -184,7 +185,8 @@ public void cleanMetrics(Set<String> topics) {
         while (iterator.hasNext()) {
             Map.Entry<String, Metric> entry = iterator.next();
             final String topic = entry.getKey();
-            if (topic.startsWith(TopicValidator.SYSTEM_TOPIC_PREFIX)) {
+            if (topic.startsWith(TopicValidator.SYSTEM_TOPIC_PREFIX)
+                    || topic.startsWith(MixAll.LMQ_PREFIX)) {
                 continue;
             }
             if (topics.contains(topic)) {
@@ -196,6 +198,16 @@ public void cleanMetrics(Set<String> topics) {
         }
     }
 
+    public boolean removeTimingCount(String topic) {
+        try {
+            timingCount.remove(topic);
+        } catch (Exception e) {
+            log.error("removeTimingCount error", e);
+            return false;
+        }
+        return true;
+    }
+
     public static class TimerMetricsSerializeWrapper extends RemotingSerializable {
         private ConcurrentMap<String, Metric> timingCount =
                 new ConcurrentHashMap<>(1024);