You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/07/20 09:36:17 UTC
(iotdb) branch fix/consensus-topic-config-validation created (now 83c81b94b21)
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a change to branch fix/consensus-topic-config-validation
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 83c81b94b21 [Subscription] Reject unsupported consensus topic attributes
This branch includes the following new commits:
new 83c81b94b21 [Subscription] Reject unsupported consensus topic attributes
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: [Subscription] Reject unsupported consensus topic attributes
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch fix/consensus-topic-config-validation
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 83c81b94b21c325900587abfc2446d8aeba5ed56
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Mon Jul 20 17:35:55 2026 +0800
[Subscription] Reject unsupported consensus topic attributes
---
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 3 ++
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 3 ++
.../persistence/subscription/SubscriptionInfo.java | 47 ++++++++++++++++++++++
.../SubscriptionInfoTopicValidationTest.java | 39 ++++++++++++++++++
4 files changed, 92 insertions(+)
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 4625d48a14c..494201b1e9f 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
@@ -662,4 +662,7 @@ public final class ConfigNodeMessages {
public static final String EXCEPTION_PROCEDURE_COMPLETED_EVICT_TTL_SHOULD_BE_GREATER_THAN_0_BUT_WAS_5A4D0CF6 =
"procedure_completed_evict_ttl should be greater than 0, but was ";
+ public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_CONSENSUS_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_3C2D0BDA =
+ "Failed to create or alter topic, mode=consensus does not support topic attributes %s";
}
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 6e45b04a001..0193e5a0485 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
@@ -707,4 +707,7 @@ public final class ConfigNodeMessages {
public static final String
EXCEPTION_PROCEDURE_COMPLETED_EVICT_TTL_SHOULD_BE_GREATER_THAN_0_BUT_WAS_5A4D0CF6 =
"procedure_completed_evict_ttl 应大于 0，但当前值为 ";
+ public static final String
+ EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_CONSENSUS_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_3C2D0BDA =
+ "创建或修改 topic 失败，mode=consensus 不支持 topic 属性 %s";
}
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 df6a0d51231..4d063e83152 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
@@ -110,6 +110,23 @@ public class SubscriptionInfo implements SnapshotProcessor {
TopicConstant.OWNER_EPOCH_KEY,
TopicConstant.MAX_OWNER_EPOCH_KEY,
TopicConstant.OWNER_LEASE_DURATION_MS_KEY);
+ private static final Set<String> CONSENSUS_TOPIC_SUPPORTED_ATTRIBUTE_KEYS =
+ Set.of(
+ SystemConstant.SQL_DIALECT_KEY,
+ TopicConstant.PATH_KEY,
+ TopicConstant.PATTERN_KEY,
+ TopicConstant.DATABASE_KEY,
+ TopicConstant.TABLE_KEY,
+ TopicConstant.COLUMN_FILTER_KEY,
+ TopicConstant.RETENTION_BYTES_KEY,
+ TopicConstant.RETENTION_MS_KEY,
+ TopicConstant.MODE_KEY,
+ TopicConstant.ORDER_MODE_KEY,
+ TopicConstant.FORMAT_KEY,
+ TopicConstant.OWNER_ID_KEY,
+ TopicConstant.OWNER_EPOCH_KEY,
+ TopicConstant.MAX_OWNER_EPOCH_KEY,
+ TopicConstant.OWNER_LEASE_DURATION_MS_KEY);
private final TopicMetaKeeper topicMetaKeeper;
private final ConsumerGroupMetaKeeper consumerGroupMetaKeeper;
@@ -329,6 +346,7 @@ public class SubscriptionInfo implements SnapshotProcessor {
throw new SubscriptionException(exceptionMessage);
}
+ validateConsensusTopicAttributes(topicConfig);
validateConsensusProtocolSupport(topicConfig);
if (topicConfig.isConsensusMode() && !topicConfig.isRecordFormat()) {
@@ -375,6 +393,35 @@ public class SubscriptionInfo implements SnapshotProcessor {
}
}
+ private void validateConsensusTopicAttributes(final TopicConfig topicConfig)
+ throws SubscriptionException {
+ if (!topicConfig.isConsensusMode()) {
+ return;
+ }
+
+ final List<String> unsupportedAttributes =
+ topicConfig.getAttribute().keySet().stream()
+ .filter(
+ key ->
+ Objects.isNull(key)
+ || !CONSENSUS_TOPIC_SUPPORTED_ATTRIBUTE_KEYS.contains(
+ key.trim().toLowerCase(Locale.ROOT)))
+ .map(String::valueOf)
+ .sorted()
+ .collect(Collectors.toList());
+ if (unsupportedAttributes.isEmpty()) {
+ return;
+ }
+
+ final String exceptionMessage =
+ String.format(
+ ConfigNodeMessages
+ .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_CONSENSUS_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_3C2D0BDA,
+ unsupportedAttributes);
+ LOGGER.warn(exceptionMessage);
+ throw new SubscriptionException(exceptionMessage);
+ }
+
private void validateConsensusProtocolSupport(final TopicConfig topicConfig)
throws SubscriptionException {
if (!topicConfig.isConsensusMode()) {
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 2fa1bf714d5..858bc41d758 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
@@ -115,6 +115,45 @@ public class SubscriptionInfoTopicValidationTest {
assertCreateRejected(subscriptionInfo, attributes, "mode=consensus only supports format");
}
+ @Test
+ public void testRejectUnsupportedAttributesOnConsensusTopic() {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes = newConsensusTableTopicAttributes();
+ attributes.put(TopicConstant.START_TIME_KEY, "0");
+ attributes.put(TopicConstant.STRICT_KEY, "false");
+ attributes.put("processor", "custom-processor");
+
+ assertCreateRejected(
+ subscriptionInfo,
+ attributes,
+ "mode=consensus does not support topic attributes [processor, start-time, strict]");
+ }
+
+ @Test
+ public void testRejectUnknownAttributeOnConsensusTopic() {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes = newConsensusTableTopicAttributes();
+ attributes.put("unknown-attribute", "value");
+
+ assertCreateRejected(
+ subscriptionInfo,
+ attributes,
+ "mode=consensus does not support topic attributes [unknown-attribute]");
+ }
+
+ @Test
+ public void testAllowPipeAttributesOnLiveTopic() throws Exception {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes = newLiveTableTopicAttributes();
+ attributes.put(TopicConstant.START_TIME_KEY, "0");
+ attributes.put(TopicConstant.STRICT_KEY, "false");
+ attributes.put("processor", "custom-processor");
+
+ Assert.assertTrue(
+ subscriptionInfo.validateBeforeCreatingTopic(
+ new TCreateTopicReq("table_topic").setTopicAttributes(attributes)));
+ }
+
@Test
public void testRejectEmptyColumnFilter() {
final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();