Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #5377 from influxdata/dom/bump-rskafka 

build: bump rskafka
pull/24376/head
Dom 2022-08-11 17:05:32 +01:00 committed by GitHub
parent 26f3849191 faa1db9a24
commit 6b8afef2ea
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 4 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@ -4211,7 +4211,7 @@ dependencies = [
[[package]]
name = "rskafka"
version = "0.3.0"
source = "git+https://github.com/influxdata/rskafka.git?rev=4f13b67b5e4b4ee8ad25c0c8370f1e63a06243e0#4f13b67b5e4b4ee8ad25c0c8370f1e63a06243e0"
source = "git+https://github.com/influxdata/rskafka.git?rev=cb9195a58fdf87be886deee961de08e119d4ca8f#cb9195a58fdf87be886deee961de08e119d4ca8f"
dependencies = [
"async-socks5",
"async-trait",

2
write_buffer/Cargo.toml
View File

@ -22,7 +22,7 @@ observability_deps = { path = "../observability_deps" }
parking_lot = "0.12"
pin-project = "1.0"
prost = "0.11"
rskafka = { git = "https://github.com/influxdata/rskafka.git", rev="4f13b67b5e4b4ee8ad25c0c8370f1e63a06243e0", default-features = false, features = ["compression-snappy", "transport-socks5"] }
rskafka = { git = "https://github.com/influxdata/rskafka.git", rev="cb9195a58fdf87be886deee961de08e119d4ca8f", default-features = false, features = ["compression-snappy", "transport-socks5"] }
schema = { path = "../schema" }
tokio = { version = "1.20", features = ["fs", "macros", "parking_lot", "rt", "sync", "time"] }
tokio-util = "0.7.3"

3
write_buffer/src/kafka/mod.rs
View File

@ -437,7 +437,7 @@ async fn setup_topic(
if let Some(topic) = topics.into_iter().find(|t| t.name == topic_name) {
let mut partition_clients = BTreeMap::new();
for partition in topic.partitions {
let c = client.partition_client(&topic_name, partition)?;
let c = client.partition_client(&topic_name, partition).await?;
let partition = u32::try_from(partition).map_err(WriteBufferError::invalid_data)?;
partition_clients.insert(partition, c);
}
@ -633,6 +633,7 @@ mod tests {
.await
.unwrap()
.partition_client(ctx.topic_name.clone(), sequencer_id as i32)
.await
.unwrap()
.produce(
vec![Record {