feat: implement the OrderedBatchProducer (#4134)

* feat: implement the `OrderedBatchProducer`

* test: add test of cancel safety

* chore: apply suggestions from CR

* chore: apply suggestions from CR

* refactor: simplify the `BackgroundProducerWorker`

* feat: implement the OrderedBatchProducer v2

* refactor: switch to `OrderedBatchProducer`

* chore: rename to `MAX_FLUSH_QUEUE_SIZE`

* refactor: switch to `OrderedBatchProducerV2`

* refactor: remove `OrderedBatchProducerV1`

* test: add tests

* refactor: make config configurable

* refactor: minor refactor

* chore: remove unused code

* chore: remove `benchmarks` crate

* chore: update config doc

* chore: remove unused comment

* refactor: refactor client registry

* refactor: rename `max_batch_size` to `max_batch_bytes`

* chore: use constant value

* chore: ensure serialized meta < ESTIMATED_META_SIZE

* chore: apply suggestions from CR

* chore: remove the `CHANNEL_SIZE`

* chore: apply suggestions from CR

* fix: ensure serialized meta < ESTIMATED_META_SIZE

* chore: apply suggestions from CR

* chore: apply suggestions from CR

src/common/wal/src/config.rs

@@ -101,8 +101,7 @@ impl From<StandaloneWalConfig> for DatanodeWalConfig {
             StandaloneWalConfig::Kafka(config) => Self::Kafka(DatanodeKafkaConfig {
                 broker_endpoints: config.broker_endpoints,
                 compression: config.compression,
-                max_batch_size: config.max_batch_size,
-                linger: config.linger,
+                max_batch_bytes: config.max_batch_bytes,
                 consumer_wait_timeout: config.consumer_wait_timeout,
                 backoff: config.backoff,
             }),
@@ -176,7 +175,7 @@ mod tests {
             topic_name_prefix = "greptimedb_wal_topic"
             replication_factor = 1
             create_topic_timeout = "30s"
-            max_batch_size = "1MB"
+            max_batch_bytes = "1MB"
             linger = "200ms"
             consumer_wait_timeout = "100ms"
             backoff_init = "500ms"
@@ -209,8 +208,7 @@ mod tests {
         let expected = DatanodeKafkaConfig {
             broker_endpoints: vec!["127.0.0.1:9092".to_string()],
             compression: Compression::default(),
-            max_batch_size: ReadableSize::mb(1),
-            linger: Duration::from_millis(200),
+            max_batch_bytes: ReadableSize::mb(1),
             consumer_wait_timeout: Duration::from_millis(100),
             backoff: BackoffConfig {
                 init: Duration::from_millis(500),
@@ -232,8 +230,7 @@ mod tests {
             replication_factor: 1,
             create_topic_timeout: Duration::from_secs(30),
             compression: Compression::default(),
-            max_batch_size: ReadableSize::mb(1),
-            linger: Duration::from_millis(200),
+            max_batch_bytes: ReadableSize::mb(1),
             consumer_wait_timeout: Duration::from_millis(100),
             backoff: BackoffConfig {
                 init: Duration::from_millis(500),

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

@@ -30,11 +30,10 @@ pub struct DatanodeKafkaConfig {
     /// The compression algorithm used to compress kafka records.
     #[serde(skip)]
     pub compression: Compression,
+    /// TODO(weny): Remove the alias once we release v0.9.
     /// The max size of a single producer batch.
-    pub max_batch_size: ReadableSize,
-    /// The linger duration of a kafka batch producer.
-    #[serde(with = "humantime_serde")]
-    pub linger: Duration,
+    #[serde(alias = "max_batch_size")]
+    pub max_batch_bytes: ReadableSize,
     /// The consumer wait timeout.
     #[serde(with = "humantime_serde")]
     pub consumer_wait_timeout: Duration,
@@ -49,8 +48,7 @@ impl Default for DatanodeKafkaConfig {
             broker_endpoints: vec![BROKER_ENDPOINT.to_string()],
             compression: Compression::NoCompression,
             // Warning: Kafka has a default limit of 1MB per message in a topic.
-            max_batch_size: ReadableSize::mb(1),
-            linger: Duration::from_millis(200),
+            max_batch_bytes: ReadableSize::mb(1),
             consumer_wait_timeout: Duration::from_millis(100),
             backoff: BackoffConfig::default(),
         }

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

@@ -43,11 +43,10 @@ pub struct StandaloneKafkaConfig {
     /// The compression algorithm used to compress kafka records.
     #[serde(skip)]
     pub compression: Compression,
+    /// TODO(weny): Remove the alias once we release v0.9.
     /// The max size of a single producer batch.
-    pub max_batch_size: ReadableSize,
-    /// The linger duration of a kafka batch producer.
-    #[serde(with = "humantime_serde")]
-    pub linger: Duration,
+    #[serde(alias = "max_batch_size")]
+    pub max_batch_bytes: ReadableSize,
     /// The consumer wait timeout.
     #[serde(with = "humantime_serde")]
     pub consumer_wait_timeout: Duration,
@@ -70,8 +69,7 @@ impl Default for StandaloneKafkaConfig {
             create_topic_timeout: Duration::from_secs(30),
             compression: Compression::NoCompression,
             // Warning: Kafka has a default limit of 1MB per message in a topic.
-            max_batch_size: ReadableSize::mb(1),
-            linger: Duration::from_millis(200),
+            max_batch_bytes: ReadableSize::mb(1),
             consumer_wait_timeout: Duration::from_millis(100),
             backoff: BackoffConfig::default(),
         }