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 <[email protected]> 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();
