mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-09-07 22:18:57 +00:00
fix(wal): bound Kafka requests and extend latency buckets (#9026)
* fix(wal): bound Kafka requests and extend latency buckets Signed-off-by: WenyXu <wenymedia@gmail.com> * chore(wal): update rskafka request timeout revision Signed-off-by: WenyXu <wenymedia@gmail.com> * docs(config): document Kafka WAL timeouts in MetaSrv Signed-off-by: WenyXu <wenymedia@gmail.com> * style: sort common-wal dev dependencies Signed-off-by: WenyXu <wenymedia@gmail.com> --------- Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
Weny Xu
Generated
+1
-1
@@ -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",
|
||||
|
||||
+1
-1
@@ -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"
|
||||
|
||||
+4
-2
@@ -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`**. |
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)`
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
)
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user