Merge pull request #2 from fede1024/master

Update to latest master from upstream

Author: rhishikeshj

README.md

@@ -139,6 +139,7 @@ Here are some of the projects using rust-rdkafka:
 - [kafka-benchmark]: a high performance benchmarking tool for Kafka.
 - [callysto]: Stream processing framework in Rust.
 - [bytewax]: Python stream processing framework using Timely Dataflow.
+- [kafka-mock-gen] easy to use mock data producer allowing stress broker
 
 *If you are using rust-rdkafka, please let us know!*
 
@@ -272,6 +273,7 @@ logging framework.
 [rdkafka-sys-known-issues]: https://github.com/fede1024/rust-rdkafka/tree/master/rdkafka-sys/README.md#known-issues
 [smol]: https://docs.rs/smol
 [Tokio]: https://tokio.rs/
+[kafka-mock-gen]: https://github.com/tomaszkubacki/kafka-mock-gen
 
 ## rdkafka-sys
 

src/admin.rs

@@ -6,6 +6,7 @@
 
 use std::collections::HashMap;
 use std::ffi::{c_void, CStr, CString};
+use std::fmt;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -43,6 +44,15 @@ pub struct AdminClient<C: ClientContext> {
     handle: Option<JoinHandle<()>>,
 }
 
+impl<C: ClientContext> fmt::Debug for AdminClient<C> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let mut debug = f.debug_struct("AdminClient");
+        debug.field("has_handle", &self.handle.is_some());
+        debug.field("stop_requested", &self.should_stop.load(Ordering::Relaxed));
+        debug.finish()
+    }
+}
+
 impl<C: ClientContext> AdminClient<C> {
     /// Creates new topics according to the provided `NewTopic` specifications.
     ///

src/config.rs

@@ -22,8 +22,9 @@
 //!
 //! [librdkafka-config]: https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 use std::ffi::CString;
+use std::fmt::Debug;
 use std::iter::FromIterator;
 use std::os::raw::c_char;
 use std::ptr;
@@ -36,6 +37,16 @@ use crate::error::{IsError, KafkaError, KafkaResult};
 use crate::log::{log_enabled, DEBUG, INFO, WARN};
 use crate::util::{ErrBuf, KafkaDrop, NativePtr};
 
+const SENSITIVE_CONFIG_KEYS: &[&str] = &[
+    "sasl.password",
+    "ssl.key.password",
+    "ssl.keystore.password",
+    "ssl.truststore.password",
+    "sasl.oauthbearer.client.secret",
+];
+
+const SANITIZED_VALUE_PLACEHOLDER: &str = "[sanitized for safety]";
+
 /// The log levels supported by librdkafka.
 #[derive(Copy, Clone, Debug)]
 pub enum RDKafkaLogLevel {
@@ -181,14 +192,35 @@ impl NativeClientConfig {
 }
 
 /// Client configuration.
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct ClientConfig {
     conf_map: HashMap<String, String>,
     /// The librdkafka logging level. Refer to [`RDKafkaLogLevel`] for the list
     /// of available levels.
     pub log_level: RDKafkaLogLevel,
 }
 
+impl Debug for ClientConfig {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let sanitized: BTreeMap<&str, &str> = self
+            .conf_map
+            .iter()
+            .filter_map(|(key, value)| {
+                if SENSITIVE_CONFIG_KEYS.contains(&key.as_str()) {
+                    None
+                } else {
+                    Some((key.as_str(), value.as_str()))
+                }
+            })
+            .collect();
+
+        let mut debug_struct = f.debug_struct("ClientConfig");
+        debug_struct.field("log_level", &self.log_level);
+        debug_struct.field("conf_map", &sanitized);
+        debug_struct.finish()
+    }
+}
+
 impl Default for ClientConfig {
     fn default() -> Self {
         Self::new()
@@ -204,9 +236,21 @@ impl ClientConfig {
         }
     }
 
-    /// Gets a reference to the underlying config map
-    pub fn config_map(&self) -> &HashMap<String, String> {
-        &self.conf_map
+    /// Returns a sanitized view of the underlying config map.
+    ///
+    /// Sensitive keys have their values replaced with a placeholder string so they never appear in
+    /// clear text when inspected.
+    pub fn config_map(&self) -> BTreeMap<&str, &str> {
+        self.conf_map
+            .iter()
+            .map(|(key, value)| {
+                if SENSITIVE_CONFIG_KEYS.contains(&key.as_str()) {
+                    (key.as_str(), SANITIZED_VALUE_PLACEHOLDER)
+                } else {
+                    (key.as_str(), value.as_str())
+                }
+            })
+            .collect()
     }
 
     /// Gets the value of a parameter in the configuration.

src/topic_partition_list.rs

@@ -5,6 +5,7 @@
 use std::collections::HashMap;
 use std::ffi::{CStr, CString};
 use std::fmt;
+use std::iter::FromIterator;
 use std::slice;
 use std::str;
 
@@ -398,6 +399,40 @@ impl Default for TopicPartitionList {
     }
 }
 
+impl FromIterator<(String, i32, Offset)> for TopicPartitionList {
+    fn from_iter<I>(iter: I) -> Self
+    where
+        I: IntoIterator<Item = (String, i32, Offset)>,
+    {
+        let iter = iter.into_iter();
+        let (lower_bound, _) = iter.size_hint();
+        let mut tpl = TopicPartitionList::with_capacity(lower_bound);
+
+        for (topic, partition, offset) in iter {
+            let mut elem = tpl.add_partition(topic.as_str(), partition);
+            elem.set_offset(offset).unwrap_or_else(|err| {
+                panic!(
+                    "failed to set offset via collect() for {}:{} (offset: {:?}): {:?}",
+                    topic, partition, offset, err
+                );
+            });
+        }
+
+        tpl
+    }
+}
+
+impl FromIterator<((String, i32), Offset)> for TopicPartitionList {
+    fn from_iter<I>(iter: I) -> Self
+    where
+        I: IntoIterator<Item = ((String, i32), Offset)>,
+    {
+        iter.into_iter()
+            .map(|((topic, partition), offset)| (topic, partition, offset))
+            .collect()
+    }
+}
+
 impl fmt::Debug for TopicPartitionList {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_list().entries(self.elements()).finish()
@@ -541,4 +576,51 @@ mod tests {
         assert_eq!(topic_map, topic_map2);
         assert_eq!(tpl, tpl2);
     }
+
+    #[test]
+    fn collect_topic_partition_list() {
+        let tpl: TopicPartitionList = vec![
+            ("t1".to_string(), 0, Offset::Beginning),
+            ("t1".to_string(), 1, Offset::Offset(42)),
+            ("t2".to_string(), 0, Offset::End),
+        ]
+        .into_iter()
+        .collect();
+
+        assert_eq!(tpl.count(), 3);
+
+        let t1_0 = tpl.find_partition("t1", 0).unwrap();
+        assert_eq!(t1_0.offset(), Offset::Beginning);
+
+        let t1_1 = tpl.find_partition("t1", 1).unwrap();
+        assert_eq!(t1_1.offset(), Offset::Offset(42));
+
+        let t2_0 = tpl.find_partition("t2", 0).unwrap();
+        assert_eq!(t2_0.offset(), Offset::End);
+    }
+
+    #[test]
+    fn collect_topic_partition_list_pairs() {
+        let tpl: TopicPartitionList = vec![
+            (("t1".to_string(), 0), Offset::Beginning),
+            (("t1".to_string(), 1), Offset::Offset(7)),
+            (("t2".to_string(), 0), Offset::Stored),
+        ]
+        .into_iter()
+        .collect();
+
+        assert_eq!(tpl.count(), 3);
+        assert_eq!(
+            tpl.find_partition("t1", 0).unwrap().offset(),
+            Offset::Beginning
+        );
+        assert_eq!(
+            tpl.find_partition("t1", 1).unwrap().offset(),
+            Offset::Offset(7)
+        );
+        assert_eq!(
+            tpl.find_partition("t2", 0).unwrap().offset(),
+            Offset::Stored
+        );
+    }
 }