Switch back to crate librdkafka

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

Author: Licenser

Cargo.lock

@@ -131,18 +131,16 @@ tokio-tungstenite = "0.18"
 # dns
 trust-dns-resolver = { version = "0.22" }
 
-# kafka. cmake is the encouraged way to build this and also the one that works on windows/with musl.
-# we stick with git until we have a release with https://github.com/fede1024/rust-rdkafka/pull/417
-#rdkafka = { version = "0.28", features = [
-rdkafka = { git = "https://github.com/fede1024/rust-rdkafka", rev = "7a2355f03c9aee8d4544e89bc4b0b0b6b96826e0", features = [
+rdkafka = { version = "0.29", features = [
   "cmake-build",
   "libz-static",
+  "tokio",
 ], default-features = false }
-# rdkafka-sys = { version = "4.2.0", features = [
-rdkafka-sys = { git = "https://github.com/fede1024/rust-rdkafka", rev = "7a2355f03c9aee8d4544e89bc4b0b0b6b96826e0", features = [
+# tracking the version rdkafka depends on
+rdkafka-sys = { version = "4.3.0", features = [
   "cmake-build",
   "libz-static",
-], default-features = false } # tracking the version rdkafka depends on
+], default-features = false }
 
 # crononome
 cron = "0.12.0"

Cargo.toml

@@ -450,12 +450,12 @@ impl Connector for KafkaConsumerConnector {
 fn kafka_meta(msg: &BorrowedMessage) -> Value<'static> {
     let headers = msg.headers().map(|headers| {
         let mut headers_meta = Value::object_with_capacity(headers.count());
-        for i in 0..headers.count() {
-            if let Some(header) = headers.get(i) {
-                let key = String::from(header.0);
-                let val = Value::Bytes(header.1.to_vec().into());
-                headers_meta.try_insert(key, val);
-            }
+        for header in headers.iter() {
+            let key = header.key.to_string();
+            let val = header
+                .value
+                .map_or(Value::const_null(), |v| Value::Bytes(v.to_vec().into()));
+            headers_meta.try_insert(key, val);
         }
         headers_meta
     });

src/connectors/impls/kafka/consumer.rs

@@ -24,8 +24,11 @@ use crate::{
     utils::task_id,
 };
 use halfbrown::HashMap;
-use rdkafka::config::{ClientConfig, FromClientConfigAndContext};
 use rdkafka::producer::{DeliveryFuture, Producer};
+use rdkafka::{
+    config::{ClientConfig, FromClientConfigAndContext},
+    message::Header,
+};
 use rdkafka::{
     message::OwnedHeaders,
     producer::{FutureProducer, FutureRecord},
@@ -68,7 +71,7 @@ impl ConfigImpl for Config {}
 #[derive(Default, Debug)]
 pub(crate) struct Builder {}
 
-#[async_trait::async_trait()]
+#[async_trait::async_trait]
 impl ConnectorBuilder for Builder {
     fn connector_type(&self) -> ConnectorType {
         "kafka_producer".into()
@@ -207,7 +210,10 @@ impl Sink for KafkaProducerSink {
                     for (k, v) in headers_obj.iter() {
                         // supporting string or bytes as headers value
                         if let Some(v_bytes) = v.as_bytes() {
-                            headers = headers.add(k, v_bytes);
+                            headers = headers.insert(Header {
+                                key: k,
+                                value: Some(v_bytes),
+                            });
                         }
                     }
                     record = record.headers(headers);
@@ -296,7 +302,7 @@ impl Sink for KafkaProducerSink {
             let wait_secs = Duration::from_secs(1);
             if producer.in_flight_count() > 0 {
                 info!("{ctx} Flushing messages. Waiting for {wait_secs:?} seconds.");
-                producer.flush(wait_secs);
+                producer.flush(wait_secs)?;
             }
         }
         Ok(())