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 41c0e2d4ee8 [Subscription] Isolate tree and table consensus topics
(#18469)
41c0e2d4ee8 is described below
commit 41c0e2d4ee8a58396d31e2519e23abd6b1d87f70
Author: Caideyipi <[email protected]>
AuthorDate: Mon Aug 24 10:58:21 2026 +0800
[Subscription] Isolate tree and table consensus topics (#18469)
* [Subscription] Isolate tree and table consensus topics
* [Subscription] Bind consensus queues by DataRegion model
---
.../ConsensusSubscriptionSetupHandler.java | 47 +++++++++++++++-------
.../ConsensusSubscriptionSetupHandlerTest.java | 35 ++++++++++++++++
2 files changed, 67 insertions(+), 15 deletions(-)
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 3fea7020e72..01e9b3d8232 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
@@ -59,6 +59,7 @@ import java.util.function.Consumer;
import java.util.function.Predicate;
import static org.apache.iotdb.commons.schema.table.Audit.isAuditDatabase;
+import static org.apache.iotdb.commons.utils.PathUtils.isTableModelDatabase;
/**
* Handles setup and teardown of consensus-based subscription queues on
DataNode.
@@ -184,8 +185,7 @@ public class ConsensusSubscriptionSetupHandler {
final String dbRaw = dataRegion.getDatabaseName();
final String dbTableModel = dbRaw.startsWith("root.") ?
dbRaw.substring(5) : dbRaw;
- // For table topics, skip if this region's database doesn't match
the topic filter.
- if (!matchesTopicDatabase(topicConfig, dbTableModel)) {
+ if (!matchesTopicDataRegion(dbRaw, topicConfig, isTableModel)) {
continue;
}
@@ -464,10 +464,11 @@ public class ConsensusSubscriptionSetupHandler {
* <p>This method discovers local DataRegion consensus groups that match the
topic filter and
* binds one consensus subscription queue to each matching region.
*
- * <p>For table-model topics, only regions whose database matches the
topic's {@code DATABASE_KEY}
- * filter are bound. For tree-model topics, all local data regions are
candidates. Additionally,
- * the {@link #onNewRegionCreated} callback ensures that regions created
after this method runs
- * are also automatically bound.
+ * <p>Only regions whose database names identify the same data model as the
consumer group are
+ * candidates. A database name with the {@code root.} prefix identifies a
tree-model region. For
+ * table-model topics, the candidate region's database must also match the
topic's {@code
+ * DATABASE_KEY} filter. Additionally, the {@link #onNewRegionCreated}
callback ensures that
+ * regions created after this method runs are also automatically bound.
*/
private static void setupConsensusQueueForTopic(
final String consumerGroupId,
@@ -527,16 +528,19 @@ public class ConsensusSubscriptionSetupHandler {
}
final String dbRaw = dataRegion.getDatabaseName();
final String dbTableModel = dbRaw.startsWith("root.") ?
dbRaw.substring(5) : dbRaw;
+ final boolean dataRegionIsTableModel = isTableModelDatabase(dbRaw);
- if (!matchesTopicDatabase(topicConfig, dbTableModel)) {
- LOGGER.info(
- DataNodePipeMessages
-
.PIPE_LOG_SKIPPING_REGION_DATABASE_FOR_TABLE_TOPIC_DATABASE_KEY_2DA27A84,
- groupId,
- dbTableModel,
- topicName,
- topicConfig.getStringOrDefault(
- TopicConstant.DATABASE_KEY,
TopicConstant.DATABASE_DEFAULT_VALUE));
+ if (!matchesTopicDataRegion(dbRaw, topicConfig, isTableModel)) {
+ if (isTableModel && dataRegionIsTableModel &&
topicConfig.isTableTopic()) {
+ LOGGER.info(
+ DataNodePipeMessages
+
.PIPE_LOG_SKIPPING_REGION_DATABASE_FOR_TABLE_TOPIC_DATABASE_KEY_2DA27A84,
+ groupId,
+ dbTableModel,
+ topicName,
+ topicConfig.getStringOrDefault(
+ TopicConstant.DATABASE_KEY,
TopicConstant.DATABASE_DEFAULT_VALUE));
+ }
continue;
}
@@ -654,6 +658,19 @@ public class ConsensusSubscriptionSetupHandler {
||
buildTablePattern(topicConfig).matchesDatabase(actualDatabaseName));
}
+ static boolean matchesTopicDataRegion(
+ final String databaseName, final TopicConfig topicConfig, final boolean
isTableModel) {
+ if (databaseName == null
+ || topicConfig == null
+ || isTableModelDatabase(databaseName) != isTableModel
+ || topicConfig.isTableTopic() != isTableModel) {
+ return false;
+ }
+ final String actualDatabaseName =
+ databaseName.startsWith("root.") ? databaseName.substring(5) :
databaseName;
+ return matchesTopicDatabase(topicConfig, actualDatabaseName);
+ }
+
private static TablePattern buildTablePattern(final TopicConfig topicConfig)
{
return new TablePattern(
true,
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 97ee9df9d3c..2bec37f65d2 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
@@ -156,6 +156,41 @@ public class ConsensusSubscriptionSetupHandlerTest {
assertTrue(ConsensusSubscriptionSetupHandler.matchesTopicDatabase(treeTopicConfig,
"user_db"));
}
+ @Test
+ public void testTopicDataRegionModelIsolation() {
+ final TopicConfig treeTopicConfig = new
TopicConfig(Collections.emptyMap());
+ final Map<String, String> tableTopicAttributes = new HashMap<>();
+ tableTopicAttributes.put(
+ SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TABLE_VALUE);
+ tableTopicAttributes.put(TopicConstant.DATABASE_KEY, "table_db");
+ final TopicConfig tableTopicConfig = new TopicConfig(tableTopicAttributes);
+
+ assertTrue(
+ ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+ "root.tree_db", treeTopicConfig, false));
+ assertFalse(
+ ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+ "table_db", treeTopicConfig, false));
+ assertFalse(
+ ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+ "table_db", treeTopicConfig, true));
+ assertFalse(
+ ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+ "root.table_db", tableTopicConfig, true));
+ assertTrue(
+ ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+ "table_db", tableTopicConfig, true));
+ assertFalse(
+ ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+ "other_table_db", tableTopicConfig, true));
+ assertFalse(
+ ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+ "table_db", tableTopicConfig, false));
+ assertFalse(
+ ConsensusSubscriptionSetupHandler.matchesTopicDataRegion(
+ "root.table_db", tableTopicConfig, false));
+ }
+
private static void failOnSecondTopic(
final String topicName, final Set<String> attemptedTopicNames) {
attemptedTopicNames.add(topicName);
