feat(config): rename table topic type to schema type and update related configurations

Author: Gezi-lzq

Diff for: clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java

@@ -264,8 +264,8 @@ public class TopicConfig {
     public static final String TABLE_TOPIC_COMMIT_INTERVAL_DOC = "The table topic commit interval(ms)";
     public static final String TABLE_TOPIC_NAMESPACE_CONFIG = "automq.table.topic.namespace";
     public static final String TABLE_TOPIC_NAMESPACE_DOC = "The table topic table namespace";
-    public static final String TABLE_TOPIC_TYPE_CONFIG = "automq.table.topic.type";
-    public static final String TABLE_TOPIC_TYPE_DOC = "The table topic type, support schemaless, registry_schema";
+    public static final String TABLE_TOPIC_SCHEMA_TYPE_CONFIG = "automq.table.topic.schema.type";
+    public static final String TABLE_TOPIC_SCHEMA_TYPE_DOC = "The table topic schema type, support schemaless, schema";
     // AutoMQ inject end
 
 }

Diff for: server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java

@@ -88,7 +88,7 @@ public final class ServerTopicConfigSynonyms {
         sameName(TopicConfig.TABLE_TOPIC_ENABLE_CONFIG),
         sameName(TopicConfig.TABLE_TOPIC_COMMIT_INTERVAL_CONFIG),
         sameName(TopicConfig.TABLE_TOPIC_NAMESPACE_CONFIG),
-        sameName(TopicConfig.TABLE_TOPIC_TYPE_CONFIG),
+        sameName(TopicConfig.TABLE_TOPIC_SCHEMA_TYPE_CONFIG),
         // AutoMQ inject end
 
         sameNameWithLogPrefix(TopicConfig.PREALLOCATE_CONFIG),

Diff for: server-common/src/main/java/org/apache/kafka/server/record/TableTopicType.java renamed to server-common/src/main/java/org/apache/kafka/server/record/TableTopicSchemaType.java

@@ -17,22 +17,22 @@
 
 import static java.util.Arrays.asList;
 
-public enum TableTopicType {
+public enum TableTopicSchemaType {
     SCHEMALESS("schemaless"),
-    REGISTRY_SCHEMA("registry_schema");
+    SCHEMA("schema");
 
     public final String name;
-    private static final List<TableTopicType> VALUES = asList(values());
+    private static final List<TableTopicSchemaType> VALUES = asList(values());
 
-    TableTopicType(String name) {
+    TableTopicSchemaType(String name) {
         this.name = name;
     }
 
     public static List<String> names() {
         return VALUES.stream().map(v -> v.name).collect(Collectors.toList());
     }
 
-    public static TableTopicType forName(String n) {
+    public static TableTopicSchemaType forName(String n) {
         String name = n.toLowerCase(Locale.ROOT);
         return VALUES.stream().filter(v -> v.name.equals(name)).findFirst().orElseThrow(() ->
             new IllegalArgumentException("Unknown table topic type name: " + name)

Diff for: storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java

@@ -40,7 +40,7 @@
 import org.apache.kafka.server.config.ServerLogConfigs;
 import org.apache.kafka.server.config.ServerTopicConfigSynonyms;
 import org.apache.kafka.server.record.BrokerCompressionType;
-import org.apache.kafka.server.record.TableTopicType;
+import org.apache.kafka.server.record.TableTopicSchemaType;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -337,7 +337,7 @@ public Optional<String> serverConfigName(String configName) {
                 .define(TopicConfig.TABLE_TOPIC_ENABLE_CONFIG, BOOLEAN, false, null, MEDIUM, TopicConfig.TABLE_TOPIC_ENABLE_DOC)
                 .define(TopicConfig.TABLE_TOPIC_COMMIT_INTERVAL_CONFIG, LONG, TimeUnit.MINUTES.toMillis(5), between(1, TimeUnit.MINUTES.toMillis(15)), MEDIUM, TopicConfig.TABLE_TOPIC_COMMIT_INTERVAL_DOC)
                 .define(TopicConfig.TABLE_TOPIC_NAMESPACE_CONFIG, STRING, null, null, MEDIUM, TopicConfig.TABLE_TOPIC_NAMESPACE_DOC)
-                .define(TopicConfig.TABLE_TOPIC_TYPE_CONFIG, STRING, null, in(TableTopicType.names().toArray(new String[0])), MEDIUM, TopicConfig.TABLE_TOPIC_TYPE_DOC)
+                .define(TopicConfig.TABLE_TOPIC_SCHEMA_TYPE_CONFIG, STRING, TableTopicSchemaType.SCHEMALESS.name, in(TableTopicSchemaType.names().toArray(new String[0])), MEDIUM, TopicConfig.TABLE_TOPIC_SCHEMA_TYPE_DOC)
                 // AutoMQ inject end
                 .define(TopicConfig.REMOTE_LOG_DISABLE_POLICY_CONFIG, STRING, TopicConfig.REMOTE_LOG_DISABLE_POLICY_RETAIN,
                         in(TopicConfig.REMOTE_LOG_DISABLE_POLICY_RETAIN, TopicConfig.REMOTE_LOG_DISABLE_POLICY_DELETE),
@@ -392,7 +392,7 @@ public Optional<String> serverConfigName(String configName) {
     public final boolean tableTopicEnable;
     public final long tableTopicCommitInterval;
     public final String tableTopicNamespace;
-    public final TableTopicType tableTopicType;
+    public final TableTopicSchemaType tableTopicSchemaType;
     // AutoMQ inject end
 
     private final int maxMessageSize;
@@ -449,7 +449,7 @@ public LogConfig(Map<?, ?> props, Set<String> overriddenConfigs) {
         this.tableTopicEnable = getBoolean(TopicConfig.TABLE_TOPIC_ENABLE_CONFIG);
         this.tableTopicCommitInterval = getLong(TopicConfig.TABLE_TOPIC_COMMIT_INTERVAL_CONFIG);
         this.tableTopicNamespace = getString(TopicConfig.TABLE_TOPIC_NAMESPACE_CONFIG);
-        this.tableTopicType = TableTopicType.forName(getString(TopicConfig.TABLE_TOPIC_TYPE_CONFIG));
+        this.tableTopicSchemaType = TableTopicSchemaType.forName(getString(TopicConfig.TABLE_TOPIC_SCHEMA_TYPE_CONFIG));
         // AutoMQ inject end
 
         remoteLogConfig = new RemoteLogConfig(this);