Updated the Errors in examples

Author: johnward

Cargo.toml

@@ -32,7 +32,8 @@ time = "0.3.7"
 rand = "0.8.5"
 lazy_static = "1.4.0"
 anyhow = "1.0.55"
-
+log = "0.4.16"
+error-chain = "0.12.4"
 
 [features]
 default = ["snappy", "gzip", "security"]

examples/console-producer.rs

@@ -6,6 +6,9 @@ use std::str::FromStr;
 use std::time::Duration;
 use std::{env, process};
 
+use anyhow::anyhow;
+use kafka::Error;
+
 use kafka::client::{
     Compression, KafkaClient, RequiredAcks, DEFAULT_CONNECTION_IDLE_TIMEOUT_MILLIS,
 };
@@ -159,7 +162,8 @@ fn send_batch(producer: &mut Producer, batch: &[Record<'_, (), Trimmed>]) -> Res
     for r in rs {
         for tpc in r.partition_confirms {
             if let Err(code) = tpc.offset {
-                return Err(ErrorKind::Kafka(kafka::error::ErrorKind::Kafka(code)));
+                //return Err(Error::Kafka(kafka::error::Error::Kafka(code)));
+                return Err(anyhow!("{:?}", code));
             }
         }
     }
@@ -199,11 +203,11 @@ impl Config {
 
         let m = match opts.parse(&args[1..]) {
             Ok(m) => m,
-            Err(e) => return Err(e),
+            Err(e) => return Err(anyhow!("Error {:?}", e)),
         };
         if m.opt_present("help") {
             let brief = format!("{} [options]", args[0]);
-            return Err(opts.usage(&brief));
+            return Err(anyhow!("opts usage: {:?}", opts.usage(&brief)));
         }
         Ok(Config {
             brokers: m
@@ -221,14 +225,21 @@ impl Config {
                 Some(ref s) if s.eq_ignore_ascii_case("gzip") => Compression::GZIP,
                 #[cfg(feature = "snappy")]
                 Some(ref s) if s.eq_ignore_ascii_case("snappy") => Compression::SNAPPY,
-                Some(s) => return Err(format!("Unsupported compression type: {}", s)),
+                Some(s) => {
+                    return Err(anyhow!(
+                        "Error {:?}",
+                        format!("Unsupported compression type: {}", s)
+                    ))
+                }
             },
             required_acks: match m.opt_str("required-acks") {
                 None => RequiredAcks::One,
                 Some(ref s) if s.eq_ignore_ascii_case("none") => RequiredAcks::None,
                 Some(ref s) if s.eq_ignore_ascii_case("one") => RequiredAcks::One,
                 Some(ref s) if s.eq_ignore_ascii_case("all") => RequiredAcks::All,
-                Some(s) => return Err(format!("Unknown --required-acks argument: {}", s)),
+                Some(s) => {
+                    return Err(anyhow!("{:?}", format!("Unknown --required-acks argument: {}", s)))
+                }
             },
             batch_size: to_number(m.opt_str("batch-size"), 1)?,
             conn_idle_timeout: Duration::from_millis(to_number(
@@ -248,7 +259,7 @@ fn to_number<N: FromStr>(s: Option<String>, _default: N) -> Result<N> {
         None => Ok(_default),
         Some(s) => match s.parse::<N>() {
             Ok(n) => Ok(n),
-            Err(_) => return Err(format!("Not a number: {}", s)),
+            Err(_) => return Err(anyhow!("{:?}", format!("Not a number: {}", s))),
         },
     }
 }

examples/example-fetch.rs

@@ -37,10 +37,10 @@ fn main() {
                 for t in resp.topics() {
                     for p in t.partitions() {
                         match p.data() {
-                            &Err(ref e) => {
+                            Err(ref e) => {
                                 println!("partition error: {}:{}: {}", t.topic(), p.partition(), e)
                             }
-                            &Ok(ref data) => {
+                            Ok(ref data) => {
                                 println!(
                                     "topic: {} / partition: {} / latest available message \
                                           offset: {}",

examples/offset-monitor.rs

@@ -7,6 +7,7 @@ use std::process;
 use std::thread;
 //use std::time as stdtime;
 use std::time::Duration;
+use std::time::SystemTime;
 
 use kafka::client::{FetchOffset, GroupOffsetStorage, KafkaClient};
 
@@ -45,7 +46,7 @@ fn run(cfg: Config) -> Result<()> {
         let ts = client.topics();
         let num_topics = ts.len();
         if num_topics == 0 {
-            return Err("no topics available");
+            return Err(Error::from("no topics available"));
         }
         let mut names: Vec<&str> = Vec::with_capacity(ts.len());
         names.extend(ts.names());
@@ -55,12 +56,12 @@ fn run(cfg: Config) -> Result<()> {
         for name in names {
             let _ = writeln!(buf, "topic: {}", name);
         }
-        return Err("choose a topic");
+        return Err(Error::from("choose a topic"));
     }
 
     // ~ otherwise let's loop over the topic partition offsets
     let num_partitions = match client.topics().partitions(&cfg.topic) {
-        None => return Err(format!("no such topic: {}", &cfg.topic)),
+        None => return Err(Error::from(format!("no such topic: {}", &cfg.topic))),
         Some(partitions) => partitions.len(),
     };
     let mut state = State::new(num_partitions, cfg.commited_not_consumed);
@@ -70,7 +71,7 @@ fn run(cfg: Config) -> Result<()> {
     // ~ initialize the state
     let mut first_time = true;
     loop {
-        let t = time::Time::now();
+        let t = SystemTime::now();
         state.update_partitions(&mut client, &cfg.topic, &cfg.group)?;
         if first_time {
             state.curr_to_prev();
@@ -225,14 +226,14 @@ impl<W: Write> Printer<W> {
         }
     }
 
-    fn print_offsets(&mut self, time: &time::Time, partitions: &[Partition]) -> Result<()> {
+    fn print_offsets(&mut self, time: &SystemTime, partitions: &[Partition]) -> Result<()> {
         self.out_buf.clear();
         {
             // ~ format
             use std::fmt::Write;
 
             self.fmt_buf.clear();
-            let _ = write!(self.fmt_buf, "{}", time.format(&self.timefmt));
+            let _ = write!(self.fmt_buf, "{}", time.elapsed().unwrap().as_secs());
             let _ = write!(self.out_buf, "{1:<0$}", self.time_width, self.fmt_buf);
             if self.print_summary {
                 let mut prev_latest = 0;
@@ -323,11 +324,11 @@ impl Config {
 
         let m = match opts.parse(&args[1..]) {
             Ok(m) => m,
-            Err(e) => return Err(e),
+            Err(e) => return Err(Error::from(e)),
         };
         if m.opt_present("help") {
             let brief = format!("{} [options]", args[0]);
-            return Err(opts.usage(&brief));
+            return Err(Error::from(opts.usage(&brief)));
         }
         let mut offset_storage = GroupOffsetStorage::Zookeeper;
         if let Some(s) = m.opt_str("storage") {
@@ -336,14 +337,14 @@ impl Config {
             } else if s.eq_ignore_ascii_case("kafka") {
                 offset_storage = GroupOffsetStorage::Kafka;
             } else {
-                return Err(format!("unknown offset store: {}", s));
+                return Err(Error::from(format!("unknown offset store: {}", s)));
             }
         }
         let mut period = Duration::from_secs(5);
         if let Some(s) = m.opt_str("sleep") {
             match s.parse::<u64>() {
                 Ok(n) if n != 0 => period = Duration::from_secs(n),
-                _ => return Err(format!("not a number greater than zero: {}", s)),
+                _ => return Err(Error::from(format!("not a number greater than zero: {}", s))),
             }
         }
         Ok(Config {

src/client/mod.rs

@@ -1006,10 +1006,10 @@ impl KafkaClient {
     ///   for t in resp.topics() {
     ///     for p in t.partitions() {
     ///       match p.data() {
-    ///         &Err(ref e) => {
+    ///         Err(ref e) => {
     ///           println!("partition error: {}:{}: {}", t.topic(), p.partition(), e)
     ///         }
-    ///         &Ok(ref data) => {
+    ///         Ok(ref data) => {
     ///           println!("topic: {} / partition: {} / latest available message offset: {}",
     ///                    t.topic(), p.partition(), data.highwatermark_offset());
     ///           for msg in data.messages() {

src/error.rs

@@ -1,10 +1,11 @@
 //! Error struct and methods
 
 use std::{io, result, sync::Arc};
+use thiserror::Error;
 
 pub type Result<T> = result::Result<T, Error>;
 
-#[derive(Debug, thiserror::Error)]
+#[derive(Debug, Error)]
 pub enum Error {
     #[error(transparent)]
     Io(#[from] io::Error),

tests/integration/consumer_producer/consumer.rs

@@ -155,7 +155,7 @@ fn test_consumer_non_existent_topic() {
         .unwrap_err();
 
     let error_code = match consumer_err {
-        error::Error(error::ErrorKind::Kafka(code), _) => code,
+        error::Error::Kafka(code) => code,
         _ => panic!("Should have received Kafka error"),
     };
 

tests/integration/consumer_producer/producer.rs

@@ -23,7 +23,7 @@ fn test_producer_send_non_existent_topic() {
         .send(&Record::from_value("non-topic", "foo".as_bytes()))
         .unwrap_err()
     {
-        error::Error(error::ErrorKind::Kafka(code), _) => code,
+        error::Error::Kafka(code) => code,
         _ => panic!("Should have received Kafka error"),
     };
 
@@ -45,10 +45,7 @@ fn test_producer_send_all() {
         assert_eq!(TEST_TOPIC_NAME.to_owned(), confirm.topic);
 
         for partition_confirm in confirm.partition_confirms {
-            assert!(
-                partition_confirm.offset.is_ok(),
-                format!("should have sent successfully. Got: {:?}", partition_confirm.offset)
-            );
+            assert!(partition_confirm.offset.is_ok(), "should have sent successfully.");
         }
     }
 }
@@ -63,7 +60,7 @@ fn test_producer_send_all_non_existent_topic() {
     ];
 
     let error_code = match producer.send_all(records).unwrap_err() {
-        error::Error(error::ErrorKind::Kafka(code), _) => code,
+        error::Error::Kafka(code) => code,
         _ => panic!("Should have received Kafka error"),
     };