diff --git a/Cargo.lock b/Cargo.lock
index 17d5882e5e..b59e53b371 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -12671,7 +12671,7 @@ dependencies = [
 [[package]]
 name = "rskafka"
 version = "0.6.0"
-source = "git+https://github.com/GreptimeTeam/rskafka.git?rev=f5688f83e7da591cda3f2674c2408b4c0ed4ed50#f5688f83e7da591cda3f2674c2408b4c0ed4ed50"
+source = "git+https://github.com/GreptimeTeam/rskafka.git?rev=5f9494104f7b43a619d6c42e2a2082e39a7b72aa#5f9494104f7b43a619d6c42e2a2082e39a7b72aa"
 dependencies = [
  "bytes",
  "chrono",
diff --git a/Cargo.toml b/Cargo.toml
index 642352674d..150880813c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -223,7 +223,7 @@ reqwest = { version = "0.13", default-features = false, features = [
 url = "2.3"
 # Branch: feat/request-timeout
 hostname = "0.4.0"
-rskafka = { git = "https://github.com/GreptimeTeam/rskafka.git", rev = "f5688f83e7da591cda3f2674c2408b4c0ed4ed50", features = [
+rskafka = { git = "https://github.com/GreptimeTeam/rskafka.git", rev = "5f9494104f7b43a619d6c42e2a2082e39a7b72aa", features = [
     "transport-tls",
 ] }
 rstest = "0.25"
diff --git a/config/config.md b/config/config.md
index 319110809d..8dcfacb1b8 100644
--- a/config/config.md
+++ b/config/config.md
@@ -106,7 +106,7 @@
 | `wal.recovery_parallelism` | Integer | `2` | Parallelism during WAL recovery. |
 | `wal.broker_endpoints` | Array | -- | The Kafka broker endpoints.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.connect_timeout` | String | `3s` | The connect timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
-| `wal.timeout` | String | `3s` | The timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.timeout` | String | `5s` | The total request timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.auto_create_topics` | Bool | `true` | Automatically create topics for WAL.<br/>Set to `true` to automatically create topics for WAL.<br/>Otherwise, use topics named `topic_name_prefix_[0..num_topics)` |
 | `wal.num_topics` | Integer | `64` | Number of topics.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.selector_type` | String | `round_robin` | Topic selector type.<br/>Available selector types:<br/>- `round_robin` (default)<br/>**It's only used when the provider is `kafka`**. |
@@ -460,6 +460,8 @@
 | `wal` | -- | -- | -- |
 | `wal.provider` | String | `raft_engine` | -- |
 | `wal.broker_endpoints` | Array | -- | The broker endpoints of the Kafka cluster.<br/><br/>**It's only used when the provider is `kafka`**. |
+| `wal.connect_timeout` | String | `3s` | The connect timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.timeout` | String | `5s` | The total request timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.auto_create_topics` | Bool | `true` | Automatically create topics for WAL.<br/>Set to `true` to automatically create topics for WAL.<br/>Otherwise, use topics named `topic_name_prefix_[0..num_topics)`<br/>**It's only used when the provider is `kafka`**. |
 | `wal.auto_prune_interval` | String | `30m` | Interval of automatically WAL pruning.<br/>Set to `0s` to disable automatically WAL pruning which delete unused remote WAL entries periodically.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.auto_prune_logical_delete` | Bool | `false` | Whether automatically WAL pruning only updates the metadata marker and skips Kafka DeleteRecords.<br/>Set to `true` for Kafka deployments that do not support DeleteRecords.<br/>**It's only used when the provider is `kafka`**. |
@@ -561,7 +563,7 @@
 | `wal.recovery_parallelism` | Integer | `2` | Parallelism during WAL recovery. |
 | `wal.broker_endpoints` | Array | -- | The Kafka broker endpoints.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.connect_timeout` | String | `3s` | The connect timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
-| `wal.timeout` | String | `3s` | The timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.timeout` | String | `5s` | The total request timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.max_batch_bytes` | String | `1MB` | The max size of a single producer batch.<br/>Warning: Kafka has a default limit of 1MB per message in a topic.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.consumer_wait_timeout` | String | `100ms` | The consumer wait timeout.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.create_index` | Bool | `false` | Whether to enable WAL index creation.<br/>**It's only used when the provider is `kafka`**. |
diff --git a/config/datanode.example.toml b/config/datanode.example.toml
index a99c817ae2..decc69dbf5 100644
--- a/config/datanode.example.toml
+++ b/config/datanode.example.toml
@@ -186,9 +186,9 @@ broker_endpoints = ["127.0.0.1:9092"]
 ## **It's only used when the provider is `kafka`**.
 #+ connect_timeout = "3s"
 
-## The timeout for kafka client.
+## The total request timeout for kafka client.
 ## **It's only used when the provider is `kafka`**.
-#+ timeout = "3s"
+#+ timeout = "5s"
 
 ## The max size of a single producer batch.
 ## Warning: Kafka has a default limit of 1MB per message in a topic.
diff --git a/config/metasrv.example.toml b/config/metasrv.example.toml
index ae373d0748..5ecbc924f7 100644
--- a/config/metasrv.example.toml
+++ b/config/metasrv.example.toml
@@ -229,6 +229,14 @@ provider = "raft_engine"
 ## **It's only used when the provider is `kafka`**.
 broker_endpoints = ["127.0.0.1:9092"]
 
+## The connect timeout for kafka client.
+## **It's only used when the provider is `kafka`**.
+#+ connect_timeout = "3s"
+
+## The total request timeout for kafka client.
+## **It's only used when the provider is `kafka`**.
+#+ timeout = "5s"
+
 ## Automatically create topics for WAL.
 ## Set to `true` to automatically create topics for WAL.
 ## Otherwise, use topics named `topic_name_prefix_[0..num_topics)`
diff --git a/config/standalone.example.toml b/config/standalone.example.toml
index 0444863727..696d739781 100644
--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -305,9 +305,9 @@ broker_endpoints = ["127.0.0.1:9092"]
 ## **It's only used when the provider is `kafka`**.
 #+ connect_timeout = "3s"
 
-## The timeout for kafka client.
+## The total request timeout for kafka client.
 ## **It's only used when the provider is `kafka`**.
-#+ timeout = "3s"
+#+ timeout = "5s"
 
 ## Automatically create topics for WAL.
 ## Set to `true` to automatically create topics for WAL.
diff --git a/src/common/wal/Cargo.toml b/src/common/wal/Cargo.toml
index b9ca39efca..62fdee79ae 100644
--- a/src/common/wal/Cargo.toml
+++ b/src/common/wal/Cargo.toml
@@ -28,5 +28,6 @@ snafu.workspace = true
 tokio.workspace = true
 
 [dev-dependencies]
+rskafka = { workspace = true, features = ["unstable-fuzzing"] }
 serde_json.workspace = true
 toml.workspace = true
diff --git a/src/common/wal/src/config.rs b/src/common/wal/src/config.rs
index 5a0d5db426..f00d615301 100644
--- a/src/common/wal/src/config.rs
+++ b/src/common/wal/src/config.rs
@@ -214,7 +214,7 @@ mod tests {
                     client_key_path: None,
                 }),
                 connect_timeout: Duration::from_secs(3),
-                timeout: Duration::from_secs(3),
+                timeout: Duration::from_secs(5),
             },
             kafka_topic: KafkaTopicConfig {
                 num_topics: 32,
@@ -252,7 +252,7 @@ mod tests {
                     client_key_path: None,
                 }),
                 connect_timeout: Duration::from_secs(3),
-                timeout: Duration::from_secs(3),
+                timeout: Duration::from_secs(5),
             },
             max_batch_bytes: ReadableSize::mb(1),
             consumer_wait_timeout: Duration::from_millis(100),
@@ -291,3 +291,33 @@ mod tests {
         assert!(!debug.contains("kafka-secret"));
     }
 }
+
+#[cfg(test)]
+mod rskafka_contract_tests {
+    use std::sync::Arc;
+    use std::time::Duration;
+
+    use rskafka::build_info::DEFAULT_CLIENT_ID;
+    use rskafka::messenger::{Messenger, RequestError, SyncVersionsError};
+
+    #[tokio::test]
+    async fn test_send_timeout_poisoned_messenger() {
+        let (_peer, stream) = tokio::io::duplex(1);
+        let mut messenger = Messenger::new(
+            stream,
+            1_000,
+            Arc::from(DEFAULT_CLIENT_ID),
+            Some(Duration::from_millis(50)),
+        );
+
+        let error = messenger.sync_versions().await.unwrap_err();
+        assert!(matches!(
+            error,
+            SyncVersionsError::RequestError(RequestError::Poisoned(error))
+                if matches!(
+                    error.as_ref(),
+                    RequestError::IO(error) if error.kind() == std::io::ErrorKind::TimedOut
+                )
+        ));
+    }
+}
diff --git a/src/common/wal/src/config/kafka/common.rs b/src/common/wal/src/config/kafka/common.rs
index 4266f30a85..cf632fbdce 100644
--- a/src/common/wal/src/config/kafka/common.rs
+++ b/src/common/wal/src/config/kafka/common.rs
@@ -198,7 +198,7 @@ pub struct KafkaConnectionConfig {
     /// The connect timeout for kafka client.
     #[serde(with = "humantime_serde")]
     pub connect_timeout: Duration,
-    /// The timeout for kafka client.
+    /// The total request timeout for kafka client.
     #[serde(with = "humantime_serde")]
     pub timeout: Duration,
 }
@@ -210,7 +210,7 @@ impl Default for KafkaConnectionConfig {
             sasl: None,
             tls: None,
             connect_timeout: Duration::from_secs(3),
-            timeout: Duration::from_secs(3),
+            timeout: Duration::from_secs(5),
         }
     }
 }
diff --git a/src/log-store/src/metrics.rs b/src/log-store/src/metrics.rs
index c82be98f66..f12ad3e815 100644
--- a/src/log-store/src/metrics.rs
+++ b/src/log-store/src/metrics.rs
@@ -54,6 +54,10 @@ lazy_static! {
         "greptime_logstore_op_elapsed",
         "logstore operation elapsed",
         &[LOGSTORE_LABEL, OPTYPE_LABEL],
+        vec![
+            0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 15.0, 20.0, 30.0,
+            45.0, 60.0,
+        ],
     )
     .unwrap();
     /// Timer of the append_batch operation on the kafka logstore.