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 0f2ee2b4545 [Subscription] Reject unsupported consensus topic
attributes (#18257)
0f2ee2b4545 is described below
commit 0f2ee2b45450c144a7f21e46e5653e0511e150f3
Author: Caideyipi <[email protected]>
AuthorDate: Tue Jul 21 09:57:57 2026 +0800
[Subscription] Reject unsupported consensus topic attributes (#18257)
---
.../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();
