[ISSUE #7308] Adding topic blacklist and filter in tiered storage module (#7310)

2026-08-30 18:10:44 +08:00 · 2023-09-06 20:42:24 +08:00
parent 37017dbaec
commit e11e29419f
6 changed files with 136 additions and 9 deletions
@@ -48,6 +48,8 @@ import org.apache.rocketmq.tieredstore.file.CompositeQueueFlatFile;
 import org.apache.rocketmq.tieredstore.file.TieredFlatFileManager;
 import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsConstant;
 import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsManager;
+import org.apache.rocketmq.tieredstore.provider.TieredStoreTopicBlackListFilter;
+import org.apache.rocketmq.tieredstore.provider.TieredStoreTopicFilter;
 import org.apache.rocketmq.tieredstore.util.CQItemBufferUtil;
 import org.apache.rocketmq.tieredstore.util.MessageBufferUtil;
 import org.apache.rocketmq.tieredstore.util.TieredStoreUtil;
@@ -56,6 +58,7 @@ public class TieredDispatcher extends ServiceThread implements CommitLogDispatch

    private static final Logger logger = LoggerFactory.getLogger(TieredStoreUtil.TIERED_STORE_LOGGER_NAME);

+    private TieredStoreTopicFilter topicFilter;
    private final String brokerName;
    private final MessageStore defaultStore;
    private final TieredMessageStoreConfig storeConfig;
@@ -70,15 +73,15 @@ public class TieredDispatcher extends ServiceThread implements CommitLogDispatch
        this.defaultStore = defaultStore;
        this.storeConfig = storeConfig;
        this.brokerName = storeConfig.getBrokerName();
+        this.topicFilter = new TieredStoreTopicBlackListFilter();
        this.tieredFlatFileManager = TieredFlatFileManager.getInstance(storeConfig);
        this.dispatchRequestReadMap = new ConcurrentHashMap<>();
        this.dispatchRequestWriteMap = new ConcurrentHashMap<>();
        this.dispatchTaskLock = new ReentrantLock();
        this.dispatchWriteLock = new ReentrantLock();
-        this.initScheduleTask();
    }

-    private void initScheduleTask() {
+    protected void initScheduleTask() {
        TieredStoreExecutor.commonScheduledExecutor.scheduleWithFixedDelay(() ->
            tieredFlatFileManager.deepCopyFlatFileToList().forEach(flatFile -> {
                if (!flatFile.getCompositeFlatFileLock().isLocked()) {
@@ -87,6 +90,14 @@ public class TieredDispatcher extends ServiceThread implements CommitLogDispatch
            }), 30, 10, TimeUnit.SECONDS);
    }

+    public TieredStoreTopicFilter getTopicFilter() {
+        return topicFilter;
+    }
+
+    public void setTopicFilter(TieredStoreTopicFilter topicFilter) {
+        this.topicFilter = topicFilter;
+    }
+
    @Override
    public void dispatch(DispatchRequest request) {
        if (stopped) {
@@ -94,7 +105,7 @@ public class TieredDispatcher extends ServiceThread implements CommitLogDispatch
        }

        String topic = request.getTopic();
-        if (TieredStoreUtil.isSystemTopic(topic)) {
+        if (topicFilter != null && topicFilter.filterTopic(topic)) {
            return;
        }

@@ -219,6 +230,10 @@ public class TieredDispatcher extends ServiceThread implements CommitLogDispatch
            return;
        }

+        if (topicFilter != null && topicFilter.filterTopic(flatFile.getMessageQueue().getTopic())) {
+            return;
+        }
+
        if (flatFile.getDispatchOffset() == -1L) {
            return;
        }
@@ -90,6 +90,7 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
        boolean loadNextStore = next.load();
        boolean result = loadFlatFile && loadNextStore;
        if (result) {
+            dispatcher.initScheduleTask();
            dispatcher.start();
        }
        return result;
@@ -134,21 +134,21 @@ public class TieredFlatFileManager {
    public void doCleanExpiredFile() {
        long expiredTimeStamp = System.currentTimeMillis() -
            TimeUnit.HOURS.toMillis(storeConfig.getTieredStoreFileReservedTime());
-        Random random = new Random();
        for (CompositeQueueFlatFile flatFile : deepCopyFlatFileToList()) {
-            int delay = random.nextInt(storeConfig.getMaxCommitJitter());
-            TieredStoreExecutor.cleanExpiredFileExecutor.schedule(() -> {
+            TieredStoreExecutor.cleanExpiredFileExecutor.submit(() -> {
                flatFile.getCompositeFlatFileLock().lock();
                try {
                    flatFile.cleanExpiredFile(expiredTimeStamp);
                    flatFile.destroyExpiredFile();
                    if (flatFile.getConsumeQueueBaseOffset() == -1) {
+                        logger.info("Clean flatFile because file not initialized, topic={}, queueId={}",
+                            flatFile.getMessageQueue().getTopic(), flatFile.getMessageQueue().getQueueId());
                        destroyCompositeFile(flatFile.getMessageQueue());
                    }
                } finally {
                    flatFile.getCompositeFlatFileLock().unlock();
                }
-            }, delay, TimeUnit.MILLISECONDS);
+            });
        }
        if (indexFile != null) {
            indexFile.cleanExpiredFile(expiredTimeStamp);
@@ -218,8 +218,13 @@ public class TieredFlatFileManager {
                                storeConfig.getBrokerName(), queueMetadata.getQueue().getQueueId()));
                            queueCount.incrementAndGet();
                        });
-                        logger.info("Recover TopicFlatFile, topic: {}, queueCount: {}, cost: {}ms",
-                            topicMetadata.getTopic(), queueCount.get(), subWatch.elapsed(TimeUnit.MILLISECONDS));
+
+                        if (queueCount.get() == 0L) {
+                            metadataStore.deleteTopic(topicMetadata.getTopic());
+                        } else {
+                            logger.info("Recover TopicFlatFile, topic: {}, queueCount: {}, cost: {}ms",
+                                topicMetadata.getTopic(), queueCount.get(), subWatch.elapsed(TimeUnit.MILLISECONDS));
+                        }
                    } catch (Exception e) {
                        logger.error("Recover TopicFlatFile error, topic: {}", topicMetadata.getTopic(), e);
                    } finally {
@@ -0,0 +1,45 @@
+/*
+ * 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.rocketmq.tieredstore.provider;
+
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.tieredstore.util.TieredStoreUtil;
+
+public class TieredStoreTopicBlackListFilter implements TieredStoreTopicFilter {
+
+    private final Set<String> topicBlackSet;
+
+    public TieredStoreTopicBlackListFilter() {
+        this.topicBlackSet = new HashSet<>();
+    }
+
+    @Override
+    public boolean filterTopic(String topicName) {
+        if (StringUtils.isBlank(topicName)) {
+            return true;
+        }
+        return TieredStoreUtil.isSystemTopic(topicName) || topicBlackSet.contains(topicName);
+    }
+
+    @Override
+    public void addTopicToWhiteList(String topicName) {
+        this.topicBlackSet.add(topicName);
+    }
+}
@@ -0,0 +1,25 @@
+/*
+ * 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.rocketmq.tieredstore.provider;
+
+public interface TieredStoreTopicFilter {
+
+    boolean filterTopic(String topicName);
+
+    void addTopicToWhiteList(String topicName);
+}
@@ -0,0 +1,36 @@
+/*
+ * 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.rocketmq.tieredstore.provider;
+
+import org.apache.rocketmq.common.topic.TopicValidator;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TieredStoreTopicBlackListFilterTest {
+
+    @Test
+    public void filterTopicTest() {
+        TieredStoreTopicFilter topicFilter = new TieredStoreTopicBlackListFilter();
+        Assert.assertTrue(topicFilter.filterTopic(""));
+        Assert.assertTrue(topicFilter.filterTopic(TopicValidator.SYSTEM_TOPIC_PREFIX + "_Topic"));
+
+        String topicName = "WhiteTopic";
+        Assert.assertFalse(topicFilter.filterTopic(topicName));
+        topicFilter.addTopicToWhiteList(topicName);
+        Assert.assertTrue(topicFilter.filterTopic(topicName));
+    }
+}