This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 8b0abccd89a [Subscription] Exclude audit database from subscriptions
(#18451)
8b0abccd89a is described below
commit 8b0abccd89adef633004e5d86e1920352cb4d9c2
Author: Caideyipi <[email protected]>
AuthorDate: Thu Aug 13 15:15:19 2026 +0800
[Subscription] Exclude audit database from subscriptions (#18451)
* [Subscription] Exclude audit database from subscriptions
* [Subscription] Reject audit-only topic filters
---
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 3 +
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 3 +
.../persistence/subscription/SubscriptionInfo.java | 79 ++++++++++++++++
.../SubscriptionInfoTopicValidationTest.java | 100 +++++++++++++++++++++
.../dataregion/DataRegionListeningFilter.java | 9 ++
.../ConsensusSubscriptionSetupHandler.java | 9 +-
.../dataregion/DataRegionListeningFilterTest.java | 50 +++++++++++
.../ConsensusSubscriptionSetupHandlerTest.java | 25 ++++++
.../apache/iotdb/commons/schema/table/Audit.java | 13 +++
.../iotdb/commons/schema/table/AuditTest.java | 41 +++++++++
10 files changed, 329 insertions(+), 3 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
index 5e79dffbffc..5ac20660610 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
@@ -672,4 +672,7 @@ public final class ConfigNodeMessages {
public static final String
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_AND_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_D86CEA8E
=
"Failed to create or alter topic, %s and %s are only supported for
incremental topics";
+ public static final String
+
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_SUBSCRIBING_ONLY_TO_THE_AUDIT_DATABASE_OR_PATHS_UNDER_IT_IS_NOT_ALLOWED_3E96A6BA
=
+ "Failed to create or alter topic, subscribing only to the __audit
database or paths under it is not allowed";
}
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
index 5f9a0d8d9ab..9519406be3c 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
@@ -715,4 +715,7 @@ public final class ConfigNodeMessages {
public static final String
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_AND_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_D86CEA8E
=
"创建或修改 topic 失败，%s 和 %s 仅支持 incremental 模式的 topic";
+ public static final String
+
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_SUBSCRIBING_ONLY_TO_THE_AUDIT_DATABASE_OR_PATHS_UNDER_IT_IS_NOT_ALLOWED_3E96A6BA
=
+ "创建或修改 topic 失败，不允许仅订阅 __audit 数据库或其下的路径";
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
index 6e26f20c0d8..36b6f35c31d 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
@@ -20,7 +20,11 @@
package org.apache.iotdb.confignode.persistence.subscription;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
import
org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper;
@@ -51,6 +55,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.mpp.rpc.thrift.TTopicOwnerLeaseEntry;
+import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
@@ -67,6 +72,7 @@ import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
@@ -80,6 +86,9 @@ import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
+import static
org.apache.iotdb.commons.schema.table.Audit.TABLE_MODEL_AUDIT_DATABASE;
+import static org.apache.iotdb.commons.schema.table.Audit.includeByAuditTreeDB;
+
public class SubscriptionInfo implements SnapshotProcessor {
private static final Logger LOGGER =
LoggerFactory.getLogger(SubscriptionInfo.class);
@@ -341,6 +350,7 @@ public class SubscriptionInfo implements SnapshotProcessor {
private void validateTopicConfig(final TopicConfig topicConfig) throws
SubscriptionException {
validateDuplicateTopicAttributes(topicConfig);
+ validateTopicDoesNotOnlySelectAuditDatabase(topicConfig);
final String mode = topicConfig.getMode();
if (!TopicConfig.isValidMode(mode)) {
@@ -403,6 +413,75 @@ public class SubscriptionInfo implements SnapshotProcessor
{
}
}
+ private void validateTopicDoesNotOnlySelectAuditDatabase(final TopicConfig
topicConfig)
+ throws SubscriptionException {
+ final boolean onlySelectsAuditDatabase =
+ topicConfig.isTableTopic()
+ ? onlySelectsAuditTableDatabase(topicConfig)
+ : onlySelectsAuditTreePaths(topicConfig);
+
+ if (onlySelectsAuditDatabase) {
+ final String exceptionMessage =
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_SUBSCRIBING_ONLY_TO_THE_AUDIT_DATABASE_OR_PATHS_UNDER_IT_IS_NOT_ALLOWED_3E96A6BA;
+ LOGGER.warn(exceptionMessage);
+ throw new SubscriptionException(exceptionMessage);
+ }
+ }
+
+ private static boolean onlySelectsAuditTableDatabase(final TopicConfig
topicConfig) {
+ final Map<String, String> sourceAttributes =
+ new
HashMap<>(topicConfig.getAttributesWithSourceDatabaseAndTableName());
+ sourceAttributes.putAll(topicConfig.getAttributesWithSourcePrefix());
+
+ try {
+ return TABLE_MODEL_AUDIT_DATABASE.equalsIgnoreCase(
+ TablePattern.parsePipePatternFromSourceParameters(new
TopicConfig(sourceAttributes))
+ .getDatabasePattern());
+ } catch (final PipeException ignored) {
+ // Invalid patterns are reported by the existing Pipe source validation
path.
+ return false;
+ }
+ }
+
+ private static boolean isAuditTreePath(final PartialPath path) {
+ return includeByAuditTreeDB(path);
+ }
+
+ private static boolean onlySelectsAuditTreePaths(final TopicConfig
topicConfig) {
+ final Map<String, String> sourceAttributes =
+ new HashMap<>(topicConfig.getAttributesWithSourcePathOrPattern());
+ sourceAttributes.putAll(topicConfig.getAttributesWithSourcePrefix());
+
+ try {
+ final TreePattern treePattern =
+ TreePattern.parsePipePatternFromSourceParameters(new
TopicConfig(sourceAttributes));
+ try {
+ return onlySelectsAuditTreePaths(treePattern);
+ } catch (final UnsupportedOperationException ignored) {
+ // Exclusions can only narrow the selected paths. Inspect the
inclusion paths when the
+ // effective pattern cannot expose a finite base path list.
+
sourceAttributes.keySet().removeIf(SubscriptionInfo::isTreeExclusionAttribute);
+ return onlySelectsAuditTreePaths(
+ TreePattern.parsePipePatternFromSourceParameters(new
TopicConfig(sourceAttributes)));
+ }
+ } catch (final PipeException ignored) {
+ // Invalid patterns are reported by the existing Pipe source validation
path.
+ return false;
+ }
+ }
+
+ private static boolean onlySelectsAuditTreePaths(final TreePattern
treePattern) {
+ final List<PartialPath> inclusionPaths =
treePattern.getBaseInclusionPaths();
+ return !inclusionPaths.isEmpty()
+ && inclusionPaths.stream().allMatch(SubscriptionInfo::isAuditTreePath);
+ }
+
+ private static boolean isTreeExclusionAttribute(final String key) {
+ return
PipeSourceConstant.SOURCE_PATTERN_EXCLUSION_KEY.equalsIgnoreCase(key)
+ || PipeSourceConstant.SOURCE_PATH_EXCLUSION_KEY.equalsIgnoreCase(key);
+ }
+
private void validateIncrementalTopicAttributes(final TopicConfig
topicConfig)
throws SubscriptionException {
if (!topicConfig.isIncrementalMode()) {
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java
index bda9cf81d7f..48523171345 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.confignode.persistence.subscription;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
@@ -85,6 +86,92 @@ public class SubscriptionInfoTopicValidationTest {
assertCreateRejected(subscriptionInfo, attributes, "only supported for
table topics");
}
+ @Test
+ public void testRejectTopicThatOnlySelectsAuditDatabase() {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+
+ final Map<String, String> tableAttributes =
newInitialTableTopicAttributes();
+ tableAttributes.put(TopicConstant.DATABASE_KEY, "__audit");
+ assertCreateRejected(subscriptionInfo, tableAttributes, "only to the
__audit database");
+
+ tableAttributes.put(TopicConstant.DATABASE_KEY, "__AUDIT");
+ assertCreateRejected(subscriptionInfo, tableAttributes, "only to the
__audit database");
+
+ tableAttributes.put(TopicConstant.DATABASE_KEY, "__audit_data");
+ assertCreateAccepted(subscriptionInfo, "table_topic", tableAttributes);
+
+ tableAttributes.put(TopicConstant.DATABASE_KEY, "__audit|user_db");
+ assertCreateAccepted(subscriptionInfo, "table_regex_topic",
tableAttributes);
+
+ tableAttributes.put(TopicConstant.DATABASE_KEY, "__audit");
+ tableAttributes.put(PipeSourceConstant.SOURCE_DATABASE_NAME_KEY,
"user_db");
+ assertCreateAccepted(subscriptionInfo, "table_source_override_topic",
tableAttributes);
+
+ tableAttributes.put(TopicConstant.DATABASE_KEY, "user_db");
+ tableAttributes.put(PipeSourceConstant.SOURCE_DATABASE_NAME_KEY,
"__audit");
+ assertCreateRejected(subscriptionInfo, tableAttributes, "only to the
__audit database");
+
+ final Map<String, String> treeAttributes = new HashMap<>();
+ treeAttributes.put(TopicConstant.PATH_KEY, "root.__audit.**");
+ assertCreateRejected(subscriptionInfo, treeAttributes, "only to the
__audit database");
+
+ treeAttributes.put(TopicConstant.PATH_KEY, "root.__audit.log.device");
+ assertCreateRejected(subscriptionInfo, treeAttributes, "only to the
__audit database");
+
+ treeAttributes.put(TopicConstant.PATH_KEY,
"root.__audit.log.**,root.__audit.user.**");
+ assertCreateRejected(subscriptionInfo, treeAttributes, "only to the
__audit database");
+
+ treeAttributes.put(TopicConstant.PATH_KEY, "root.__audit.**,root.user.**");
+ assertCreateAccepted(subscriptionInfo, "tree_topic", treeAttributes);
+
+ treeAttributes.put(TopicConstant.PATH_KEY, "root.__audit_data.**");
+ assertCreateAccepted(subscriptionInfo, "tree_similar_name_topic",
treeAttributes);
+
+ treeAttributes.remove(TopicConstant.PATH_KEY);
+ treeAttributes.put(TopicConstant.PATTERN_KEY, "root.__audit.log");
+ assertCreateRejected(subscriptionInfo, treeAttributes, "only to the
__audit database");
+
+ treeAttributes.put(TopicConstant.PATTERN_KEY, "root.__audit");
+ assertCreateAccepted(subscriptionInfo, "tree_prefix_topic",
treeAttributes);
+ }
+
+ @Test
+ public void testAuditOnlyValidationHandlesSourceInclusionAndExclusion() {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(TopicConstant.PATH_KEY, "root.__audit.**");
+ attributes.put("source.path.exclusion", "root.__audit.internal.**");
+ assertCreateRejected(subscriptionInfo, attributes, "only to the __audit
database");
+
+ attributes.put("source.path.inclusion", "root.user.**");
+ assertCreateAccepted(subscriptionInfo, "tree_source_inclusion_topic",
attributes);
+
+ final Map<String, String> attributesWithFullyExcludedUserPath = new
HashMap<>();
+ attributesWithFullyExcludedUserPath.put(TopicConstant.PATH_KEY,
"root.__audit.**,root.user.**");
+ attributesWithFullyExcludedUserPath.put("source.path.exclusion",
"root.user.**");
+ assertCreateRejected(
+ subscriptionInfo, attributesWithFullyExcludedUserPath, "only to the
__audit database");
+ }
+
+ @Test
+ public void testRejectAlteringTopicToOnlySelectAuditDatabase() throws
Exception {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> originalAttributes = new HashMap<>();
+ originalAttributes.put(TopicConstant.PATH_KEY, "root.user.**");
+ subscriptionInfo.createTopic(
+ new CreateTopicPlan(new TopicMeta("tree_topic", 1L,
originalAttributes)));
+
+ final Map<String, String> updatedAttributes = new HashMap<>();
+ updatedAttributes.put(TopicConstant.PATH_KEY, "root.__audit.log.**");
+ try {
+ subscriptionInfo.validateBeforeAlteringTopic(
+ new TopicMeta("tree_topic", 2L, updatedAttributes));
+ Assert.fail("Expected audit-only topic validation to fail");
+ } catch (final SubscriptionException e) {
+ Assert.assertTrue(e.getMessage().contains("only to the __audit
database"));
+ }
+ }
+
@Test
public void testRejectDuplicateTopicConfigKeys() {
final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
@@ -353,4 +440,17 @@ public class SubscriptionInfoTopicValidationTest {
Assert.assertTrue(e.getMessage().contains(expectedMessagePart));
}
}
+
+ private static void assertCreateAccepted(
+ final SubscriptionInfo subscriptionInfo,
+ final String topicName,
+ final Map<String, String> attributes) {
+ try {
+ Assert.assertTrue(
+ subscriptionInfo.validateBeforeCreatingTopic(
+ new TCreateTopicReq(topicName).setTopicAttributes(attributes)));
+ } catch (final SubscriptionException e) {
+ Assert.fail(e.getMessage());
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/DataRegionListeningFilter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/DataRegionListeningFilter.java
index e2a15a64480..ad4c1fddff3 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/DataRegionListeningFilter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/DataRegionListeningFilter.java
@@ -38,6 +38,7 @@ import java.util.stream.Collectors;
import static
org.apache.iotdb.commons.pipe.datastructure.options.PipeInclusionOptions.getExclusionString;
import static
org.apache.iotdb.commons.pipe.datastructure.options.PipeInclusionOptions.getInclusionString;
import static
org.apache.iotdb.commons.pipe.datastructure.options.PipeInclusionOptions.parseOptions;
+import static org.apache.iotdb.commons.schema.table.Audit.isAuditDatabase;
/**
* {@link DataRegionListeningFilter} is to tell the insertion and deletion for
{@link PipeTask} on
@@ -59,6 +60,10 @@ public class DataRegionListeningFilter {
public static boolean shouldDatabaseBeListened(
final PipeParameters parameters, final boolean isTableModel, final
String databaseRawName)
throws IllegalPathException {
+ if (isAuditDatabase(databaseRawName)) {
+ return false;
+ }
+
final Pair<Boolean, Boolean> insertionDeletionListeningOptionPair =
parseInsertionDeletionListeningOptionPair(parameters);
final boolean hasSpecificListeningOption =
@@ -101,6 +106,10 @@ public class DataRegionListeningFilter {
}
final String databaseRawName = dataRegion.getDatabaseName();
+ if (isAuditDatabase(databaseRawName)) {
+ return false;
+ }
+
final String databaseTreeModel =
databaseRawName.startsWith("root.") ? databaseRawName : "root." +
databaseRawName;
final String databaseTableModel =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
index bb512922248..3fea7020e72 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
@@ -58,6 +58,8 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import java.util.function.Predicate;
+import static org.apache.iotdb.commons.schema.table.Audit.isAuditDatabase;
+
/**
* Handles setup and teardown of consensus-based subscription queues on
DataNode.
*
@@ -645,10 +647,11 @@ public class ConsensusSubscriptionSetupHandler {
return new ConsensusLogToTabletConverter(treePattern, tablePattern, null,
actualDatabaseName);
}
- private static boolean matchesTopicDatabase(
+ static boolean matchesTopicDatabase(
final TopicConfig topicConfig, final String actualDatabaseName) {
- return !topicConfig.isTableTopic()
- || buildTablePattern(topicConfig).matchesDatabase(actualDatabaseName);
+ return !isAuditDatabase(actualDatabaseName)
+ && (!topicConfig.isTableTopic()
+ ||
buildTablePattern(topicConfig).matchesDatabase(actualDatabaseName));
}
private static TablePattern buildTablePattern(final TopicConfig topicConfig)
{
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/DataRegionListeningFilterTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/DataRegionListeningFilterTest.java
new file mode 100644
index 00000000000..4901afcf912
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/DataRegionListeningFilterTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.iotdb.db.pipe.source.dataregion;
+
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
+import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class DataRegionListeningFilterTest {
+
+ @Test
+ public void testAuditDatabaseIsNeverListened() throws Exception {
+ final Map<String, String> topicAttributes = new HashMap<>();
+ topicAttributes.put(SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TABLE_VALUE);
+ final PipeParameters parameters =
+ new PipeParameters(
+ new TopicMeta("topic", 1,
topicAttributes).generateExtractorAttributes("root"));
+
+ assertFalse(DataRegionListeningFilter.shouldDatabaseBeListened(parameters,
true, "__audit"));
+ assertFalse(
+ DataRegionListeningFilter.shouldDatabaseBeListened(parameters, false,
"root.__audit"));
+ assertFalse(DataRegionListeningFilter.shouldDatabaseBeListened(parameters,
true, "__AUDIT"));
+ assertTrue(DataRegionListeningFilter.shouldDatabaseBeListened(parameters,
true, "user_db"));
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
index bd31d67a790..97ee9df9d3c 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
@@ -20,7 +20,10 @@
package org.apache.iotdb.db.subscription.broker.consensus;
import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
import org.junit.Rule;
@@ -30,11 +33,14 @@ import org.junit.rules.TemporaryFolder;
import java.io.File;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.LinkedHashSet;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -131,6 +137,25 @@ public class ConsensusSubscriptionSetupHandlerTest {
}
}
+ @Test
+ public void testAuditDatabaseNeverMatchesTopic() {
+ final Map<String, String> tableTopicAttributes = new HashMap<>();
+ tableTopicAttributes.put(
+ SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TABLE_VALUE);
+ tableTopicAttributes.put(TopicConstant.DATABASE_KEY, ".*");
+ final TopicConfig tableTopicConfig = new TopicConfig(tableTopicAttributes);
+
+ assertFalse(
+
ConsensusSubscriptionSetupHandler.matchesTopicDatabase(tableTopicConfig,
"__audit"));
+ assertFalse(
+
ConsensusSubscriptionSetupHandler.matchesTopicDatabase(tableTopicConfig,
"root.__audit"));
+
assertTrue(ConsensusSubscriptionSetupHandler.matchesTopicDatabase(tableTopicConfig,
"user_db"));
+
+ final TopicConfig treeTopicConfig = new
TopicConfig(Collections.emptyMap());
+
assertFalse(ConsensusSubscriptionSetupHandler.matchesTopicDatabase(treeTopicConfig,
"__audit"));
+
assertTrue(ConsensusSubscriptionSetupHandler.matchesTopicDatabase(treeTopicConfig,
"user_db"));
+ }
+
private static void failOnSecondTopic(
final String topicName, final Set<String> attemptedTopicNames) {
attemptedTopicNames.add(topicName);
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/Audit.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/Audit.java
index e9d8f7fe1f4..a33cda1b5dc 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/Audit.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/Audit.java
@@ -52,6 +52,19 @@ public class Audit {
&& TABLE_MODEL_AUDIT_DATABASE.equalsIgnoreCase(nodes[1]);
}
+ public static boolean isAuditDatabase(final String databaseName) {
+ return isDatabaseOrDescendant(databaseName, TABLE_MODEL_AUDIT_DATABASE)
+ || isDatabaseOrDescendant(databaseName, TREE_MODEL_AUDIT_DATABASE);
+ }
+
+ private static boolean isDatabaseOrDescendant(
+ final String databaseName, final String auditDatabaseName) {
+ return databaseName != null
+ && databaseName.regionMatches(true, 0, auditDatabaseName, 0,
auditDatabaseName.length())
+ && (databaseName.length() == auditDatabaseName.length()
+ || databaseName.charAt(auditDatabaseName.length()) == '.');
+ }
+
public static String getReservedDatabaseNameErrorMsg(String databaseName) {
return String.format(RESERVED_DATABASE_NAME_ERROR_MSG, databaseName);
}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/AuditTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/AuditTest.java
new file mode 100644
index 00000000000..ad66cd0f284
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/AuditTest.java
@@ -0,0 +1,41 @@
+/*
+ * 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.iotdb.commons.schema.table;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class AuditTest {
+
+ @Test
+ public void testIsAuditDatabase() {
+ assertTrue(Audit.isAuditDatabase("__audit"));
+ assertTrue(Audit.isAuditDatabase("root.__audit"));
+ assertTrue(Audit.isAuditDatabase("__AUDIT"));
+ assertTrue(Audit.isAuditDatabase("ROOT.__AUDIT"));
+ assertTrue(Audit.isAuditDatabase("__audit.data"));
+ assertTrue(Audit.isAuditDatabase("root.__audit.data"));
+ assertFalse(Audit.isAuditDatabase("__audit_data"));
+ assertFalse(Audit.isAuditDatabase("root.__audit_data"));
+ assertFalse(Audit.isAuditDatabase(null));
+ }
+}
