[Subscription] Reject unsupported consensus topic attributes

iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java

@@ -662,4 +662,7 @@ private 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";
 }

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java

@@ -707,4 +707,7 @@ private 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";
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java

@@ -110,6 +110,23 @@
           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 @@
       throw new SubscriptionException(exceptionMessage);
     }
 
+    validateConsensusTopicAttributes(topicConfig);
     validateConsensusProtocolSupport(topicConfig);
 
     if (topicConfig.isConsensusMode() && !topicConfig.isRecordFormat()) {
@@ -375,6 +393,35 @@
     }
   }
 
+  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()) {

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java

@@ -115,6 +115,45 @@ public void testRejectLegacyTsFileAliasOnConsensusTopic() {
     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();