Fix api in kafka tests

Signed-off-by: Heinz N. Gies <[email protected]>

Author: Licenser

src/connectors/tests.rs

@@ -332,7 +332,8 @@ impl ConnectorHarness {
         feature = "net-integration",
         feature = "ws-integration",
         feature = "s3-integration",
-        feature = "gcp-integration"
+        feature = "gcp-integration",
+        feature = "kafka-integration",
     ))]
     pub(crate) async fn send_to_sink(&self, event: Event, port: Port<'static>) -> Result<()> {
         self.addr.send_sink(SinkMsg::Event { event, port }).await

src/connectors/tests/kafka/consumer.rs

@@ -29,7 +29,7 @@ use rdkafka::{
     config::FromClientConfig,
     consumer::{BaseConsumer, Consumer},
     error::KafkaResult,
-    message::OwnedHeaders,
+    message::{Header, OwnedHeaders},
     producer::{FutureProducer, FutureRecord},
     ClientConfig, Offset,
 };
@@ -101,7 +101,10 @@ async fn transactional_retry() -> Result<()> {
         .key("foo")
         .partition(1)
         .timestamp(42)
-        .headers(OwnedHeaders::new().add("header", "snot"));
+        .headers(OwnedHeaders::new().insert(Header {
+            key: "header",
+            value: Some("snot"),
+        }));
     if producer.send(record, PRODUCE_TIMEOUT).await.is_err() {
         return Err("Unable to send record to kafka".into());
     }
@@ -328,7 +331,10 @@ async fn custom_no_retry() -> Result<()> {
         .key("foo")
         .partition(1)
         .timestamp(42)
-        .headers(OwnedHeaders::new().add("header", "snot"));
+        .headers(OwnedHeaders::new().insert(Header {
+            key: "header",
+            value: Some("snot"),
+        }));
     if producer.send(record, PRODUCE_TIMEOUT).await.is_err() {
         return Err("Unable to send record to kafka".into());
     }
@@ -530,7 +536,10 @@ async fn performance() -> Result<()> {
         .key("foo")
         .partition(1)
         .timestamp(42)
-        .headers(OwnedHeaders::new().add("header", "snot"));
+        .headers(OwnedHeaders::new().insert(Header {
+            key: "header",
+            value: Some("snot"),
+        }));
     if producer.send(record, PRODUCE_TIMEOUT).await.is_err() {
         return Err("Unable to send record to kafka".into());
     }

src/connectors/tests/kafka/producer.rs

@@ -174,7 +174,9 @@ async fn connector_kafka_producer() -> Result<()> {
             assert_eq!(Some(123), msg.timestamp().to_millis());
             let headers = msg.headers().expect("No headers found");
             assert_eq!(1, headers.count());
-            assert_eq!(Some(("foo", "baz".as_bytes())), headers.get(0));
+            let h = headers.get(0);
+            assert_eq!("foo", h.key);
+            assert_eq!("baz".as_bytes(), h.value.expect("no value"));
             consumer
                 .commit_message(&msg, CommitMode::Sync)
                 .expect("Commit failed");