This method discovers local DataRegion consensus groups that match the topic filter and * binds one consensus subscription queue to each matching region. * - *
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. + *
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 @@ private static void setupConsensusQueueForTopic(
}
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 @@ static boolean matchesTopicDatabase(
|| 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 97ee9df9d3c5..2bec37f65d23 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 void testAuditDatabaseNeverMatchesTopic() {
assertTrue(ConsensusSubscriptionSetupHandler.matchesTopicDatabase(treeTopicConfig, "user_db"));
}
+ @Test
+ public void testTopicDataRegionModelIsolation() {
+ final TopicConfig treeTopicConfig = new TopicConfig(Collections.emptyMap());
+ final Map