remove with prefix

config/config.md

@@ -279,11 +279,11 @@
 | `wal` | -- | -- | -- |
 | `wal.provider` | String | `raft_engine` | -- |
 | `wal.broker_endpoints` | Array | -- | The broker endpoints of the Kafka cluster. |
-| `wal.kafka_topic_num_topics` | Integer | `64` | Number of topics to be created upon start. |
-| `wal.kafka_topic_selector_type` | String | `round_robin` | Topic selector type.<br/>Available selector types:<br/>- `round_robin` (default) |
-| `wal.kafka_topic_name_prefix` | String | `greptimedb_wal_topic` | A Kafka topic is constructed by concatenating `name_prefix` and `topic_id`. |
-| `wal.kafka_topic_replication_factor` | Integer | `1` | Expected number of replicas of each partition. |
-| `wal.kafka_topic_create_topic_timeout` | String | `30s` | Above which a topic creation operation will be cancelled. |
+| `wal.num_topics` | Integer | `64` | Number of topics to be created upon start. |
+| `wal.selector_type` | String | `round_robin` | Topic selector type.<br/>Available selector types:<br/>- `round_robin` (default) |
+| `wal.topic_anme_prefix` | String | `greptimedb_wal_topic` | A Kafka topic is constructed by concatenating `name_prefix` and `topic_id`. |
+| `wal.replication_factor` | Integer | `1` | Expected number of replicas of each partition. |
+| `wal.create_topic_timeout` | String | `30s` | Above which a topic creation operation will be cancelled. |
 | `wal.backoff_init` | String | `500ms` | The initial backoff for kafka clients. |
 | `wal.backoff_max` | String | `10s` | The maximum backoff for kafka clients. |
 | `wal.backoff_base` | Integer | `2` | Exponential backoff rate, i.e. next backoff = base * current backoff. |

config/metasrv.example.toml

@@ -77,21 +77,21 @@ provider = "raft_engine"
 broker_endpoints = ["127.0.0.1:9092"]
 
 ## Number of topics to be created upon start.
-kafka_topic_num_topics = 64
+num_topics = 64
 
 ## Topic selector type.
 ## Available selector types:
 ## - `round_robin` (default)
-kafka_topic_selector_type = "round_robin"
+selector_type = "round_robin"
 
 ## A Kafka topic is constructed by concatenating `name_prefix` and `topic_id`.
-kafka_topic_name_prefix = "greptimedb_wal_topic"
+topic_name_prefix = "greptimedb_wal_topic"
 
 ## Expected number of replicas of each partition.
-kafka_topic_replication_factor = 1
+replication_factor = 1
 
 ## Above which a topic creation operation will be cancelled.
-kafka_topic_create_topic_timeout = "30s"
+create_topic_timeout = "30s"
 ## The initial backoff for kafka clients.
 backoff_init = "500ms"
 

src/common/meta/src/wal_options_allocator/kafka/topic_manager.rs

@@ -57,7 +57,7 @@ impl TopicManager {
     pub fn new(config: MetasrvKafkaConfig, kv_backend: KvBackendRef) -> Self {
         // Topics should be created.
         let topics = (0..config.kafka_topic.num_topics)
-            .map(|topic_id| format!("{}_{topic_id}", config.kafka_topic.name_prefix))
+            .map(|topic_id| format!("{}_{topic_id}", config.kafka_topic.topic_name_prefix))
             .collect::<Vec<_>>();
 
         let selector = match config.kafka_topic.selector_type {

src/common/wal/src/config.rs

@@ -139,12 +139,12 @@ mod tests {
             backoff_max = "10s"
             backoff_base = 2
             backoff_deadline = "5mins"
-            kafka_topic_num_topics = 32
-            kafka_topic_num_partitions = 1
-            kafka_topic_selector_type = "round_robin"
-            kafka_topic_replication_factor = 1
-            kafka_topic_create_topic_timeout = "30s"
-            kafka_topic_name_prefix = "greptimedb_wal_topic"
+            num_topics = 32
+            num_partitions = 1
+            selector_type = "round_robin"
+            replication_factor = 1
+            create_topic_timeout = "30s"
+            topic_name_prefix = "greptimedb_wal_topic"
         "#;
 
         // Deserialized to MetasrvWalConfig.
@@ -160,7 +160,7 @@ mod tests {
             kafka_topic: KafkaTopicConfig {
                 num_topics: 32,
                 selector_type: TopicSelectorType::RoundRobin,
-                name_prefix: "greptimedb_wal_topic".to_string(),
+                topic_name_prefix: "greptimedb_wal_topic".to_string(),
                 num_partitions: 1,
                 replication_factor: 1,
                 create_topic_timeout: Duration::from_secs(30),
@@ -183,7 +183,7 @@ mod tests {
             kafka_topic: KafkaTopicConfig {
                 num_topics: 32,
                 selector_type: TopicSelectorType::RoundRobin,
-                name_prefix: "greptimedb_wal_topic".to_string(),
+                topic_name_prefix: "greptimedb_wal_topic".to_string(),
                 num_partitions: 1,
                 replication_factor: 1,
                 create_topic_timeout: Duration::from_secs(30),

src/common/wal/src/config/kafka/common.rs

@@ -20,7 +20,6 @@ use serde_with::with_prefix;
 use crate::{TopicSelectorType, TOPIC_NAME_PREFIX};
 
 with_prefix!(pub backoff_prefix "backoff_");
-with_prefix!(pub kafka_topic_prefix "kafka_topic_");
 
 /// Backoff configurations for kafka clients.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -66,7 +65,7 @@ pub struct KafkaTopicConfig {
     #[serde(with = "humantime_serde")]
     pub create_topic_timeout: Duration,
     /// Topic name prefix.
-    pub name_prefix: String,
+    pub topic_name_prefix: String,
 }
 
 impl Default for KafkaTopicConfig {
@@ -77,7 +76,7 @@ impl Default for KafkaTopicConfig {
             selector_type: TopicSelectorType::RoundRobin,
             replication_factor: 1,
             create_topic_timeout: Duration::from_secs(30),
-            name_prefix: TOPIC_NAME_PREFIX.to_string(),
+            topic_name_prefix: TOPIC_NAME_PREFIX.to_string(),
         }
     }
 }

src/common/wal/src/config/kafka/datanode.rs

@@ -17,9 +17,7 @@ use std::time::Duration;
 use common_base::readable_size::ReadableSize;
 use serde::{Deserialize, Serialize};
 
-use crate::config::kafka::common::{
-    backoff_prefix, kafka_topic_prefix, BackoffConfig, KafkaTopicConfig,
-};
+use crate::config::kafka::common::{backoff_prefix, BackoffConfig, KafkaTopicConfig};
 use crate::BROKER_ENDPOINT;
 
 /// Kafka wal configurations for datanode.
@@ -39,7 +37,7 @@ pub struct DatanodeKafkaConfig {
     #[serde(flatten, with = "backoff_prefix")]
     pub backoff: BackoffConfig,
     /// The kafka topic config.
-    #[serde(flatten, with = "kafka_topic_prefix")]
+    #[serde(flatten)]
     pub kafka_topic: KafkaTopicConfig,
 }
 

src/common/wal/src/config/kafka/metasrv.rs

@@ -14,9 +14,7 @@
 
 use serde::{Deserialize, Serialize};
 
-use crate::config::kafka::common::{
-    backoff_prefix, kafka_topic_prefix, BackoffConfig, KafkaTopicConfig,
-};
+use crate::config::kafka::common::{backoff_prefix, BackoffConfig, KafkaTopicConfig};
 use crate::BROKER_ENDPOINT;
 
 /// Kafka wal configurations for metasrv.
@@ -29,7 +27,7 @@ pub struct MetasrvKafkaConfig {
     #[serde(flatten, with = "backoff_prefix")]
     pub backoff: BackoffConfig,
     /// The kafka config.
-    #[serde(flatten, with = "kafka_topic_prefix")]
+    #[serde(flatten)]
     pub kafka_topic: KafkaTopicConfig,
 }
 

tests/conf/metasrv-test.toml.template

@@ -4,7 +4,7 @@ provider = "raft_engine"
 {{ else }}
 provider = "kafka"
 broker_endpoints = {kafka_wal_broker_endpoints | unescaped}
-kafka_topic_num_topics = 64
-kafka_topic_selector_type = "round_robin"
-kafka_topic_name_prefix = "distributed_test_greptimedb_wal_topic"
+num_topics = 64
+selector_type = "round_robin"
+name_prefix = "distributed_test_greptimedb_wal_topic"
 {{ endif }}