Merge pull request #189 from influxdata/dependabot/cargo/rdkafka-0.29

chore(deps): update rdkafka requirement from 0.28 to 0.29

Author: kodiakhq[bot]

Cargo.toml

@@ -36,7 +36,7 @@ thiserror = "1.0"
 tokio = { version = "1.19", default-features = false, features = ["io-util", "net", "rt", "sync", "time", "macros"] }
 tokio-rustls = { version = "0.23", optional = true }
 tracing = "0.1"
-zstd = { version = "0.10", optional = true }
+zstd = { version = "0.11", optional = true }
 
 [dev-dependencies]
 assert_matches = "1.5"
@@ -49,7 +49,7 @@ procspawn = "0.10"
 proptest = "1"
 proptest-derive = "0.3"
 rustls-pemfile = "1.0"
-rdkafka = { version = "0.28", default-features = false, features = ["libz", "tokio", "zstd"] }
+rdkafka = { version = "0.29", default-features = false, features = ["libz", "tokio", "zstd"] }
 tokio = { version = "1.14", features = ["macros", "rt-multi-thread"] }
 tracing-log = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }

tests/rdkafka_helper.rs

@@ -4,7 +4,7 @@ use chrono::{TimeZone, Utc};
 use futures::{StreamExt, TryStreamExt};
 use rdkafka::{
     consumer::{Consumer, StreamConsumer},
-    message::{Headers, OwnedHeaders},
+    message::{Header, Headers, OwnedHeaders},
     producer::{FutureProducer, FutureRecord},
     util::Timeout,
     ClientConfig, Message, TopicPartitionList,
@@ -49,7 +49,10 @@ pub async fn produce(
     for (topic_name, partition_index, record) in records {
         let mut headers = OwnedHeaders::new();
         for (k, v) in record.headers {
-            headers = headers.add(&k, &v);
+            headers = headers.insert(Header {
+                key: &k,
+                value: Some(&v),
+            });
         }
 
         let mut f_record = FutureRecord::to(&topic_name)
@@ -104,8 +107,8 @@ pub async fn consume(
                             .map(|headers| {
                                 (0..headers.count())
                                     .map(|i| {
-                                        let (k, v) = headers.get(i).unwrap();
-                                        (k.to_owned(), v.to_vec())
+                                        let header = headers.get(i);
+                                        (header.key.to_owned(), header.value.unwrap().to_vec())
                                     })
                                     .collect()
                             })