Repository: kafka-rust/kafka-rust (language: rust)

Issue:
producer: Producer#send panics with no_acks configuration
If `Producer` is configured with `RequiredAcks::None`, sending a single message through `Producer::send` panics due to an invalid assertion in this method! This bug seems to be sitting there around since the `Producer's` birth.



Discussion / hints:

Fix patch:
diff --git a/examples/console-producer.rs b/examples/console-producer.rs
index 2b430a6ac..f2ac9ba8d 100644
--- a/examples/console-producer.rs
+++ b/examples/console-producer.rs
@@ -10,8 +10,6 @@ use std::ops::{Deref, DerefMut};
 use kafka::client::{KafkaClient, Compression, RequiredAcks};
 use kafka::producer::{AsBytes, Producer, Record};
 
-// how many brokers do we require to acknowledge the send messages
-const REQUIRED_ACKS: RequiredAcks = RequiredAcks::One;
 // how long do we allow wainting for the acknowledgement
 const ACK_TIMEOUT: i32 = 100;
 
@@ -62,7 +60,7 @@ fn produce(cfg: &Config) -> Result<(), Error> {
 fn produce_impl(src: &mut BufRead, client: KafkaClient, cfg: &Config) -> Result<(), Error> {
     let mut producer = try!(Producer::from_client(client)
                             .with_ack_timeout(ACK_TIMEOUT)
-                            .with_required_acks(REQUIRED_ACKS)
+                            .with_required_acks(cfg.required_acks)
                             .create());
     if cfg.batch_size < 2 {
         produce_impl_nobatch(&mut producer, src, cfg)
@@ -195,11 +193,14 @@ struct Config {
     topic: String,
     input_file: Option<String>,
     compression: Compression,
+    required_acks: RequiredAcks,
     batch_size: usize,
 }
 
 impl Config {
     fn from_cmdline() -> Result<Config, Error> {
+        use std::ascii::AsciiExt;
+
         let args: Vec<String> = env::args().collect();
         let mut opts = getopts::Options::new();
         opts.optflag("h", "help", "Print this help screen");
@@ -207,6 +208,8 @@ impl Config {
         opts.optopt("", "topic", "Specify target topic", "NAME");
         opts.optopt("", "input", "Specify input file", "FILE");
         opts.optopt("", "compression", "Compress messages [NONE, GZIP, SNAPPY]", "TYPE");
+        opts.optopt("", "required-acks", "Specify amount of required broker acknowledgments \
+                                          [NONE, ONE, ALL]", "TYPE");
         opts.optopt("", "batch-size", "Send N message in one batch.", "N");
 
         let m = match opts.parse(&args[1..]) {
@@ -226,17 +229,21 @@ impl Config {
             topic: m.opt_str("topic")
                 .unwrap_or_else(|| "my-topic".to_owned()),
             input_file: m.opt_str("input"),
-            compression: {
-                let s = m.opt_str("compression")
-                    .unwrap_or_else(|| "NONE".to_owned());
-                match s.trim() {
-                    "none" | "NONE" => Compression::NONE,
-                    #[cfg(feature = "gzip")]
-                    "gzip" | "GZIP" => Compression::GZIP,
-                    #[cfg(feature = "snappy")]
-                    "snappy" | "SNAPPY" => Compression::SNAPPY,
-                    _ => return Err(Error::Literal(format!("Unsupported compression type: {}", s))),
-                }
+            compression: match m.opt_str("compression") {
+                None => Compression::NONE,
+                Some(ref s) if s.eq_ignore_ascii_case("none") => Compression::NONE,
+                #[cfg(feature = "gzip")]
+                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(Error::Literal(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(Error::Literal(format!("Unknown --required-acks argument: {}", s))),
             },
             batch_size: {
                 match m.opt_str("batch-size") {
diff --git a/src/client/mod.rs b/src/client/mod.rs
index a68d832a1..780909a7a 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -1360,7 +1360,7 @@ fn __fetch_messages(conn_pool: &mut ConnectionPool,
     Ok(res)
 }
 
-/// ~ carries out the given produce requests and returns the reponse
+/// ~ carries out the given produce requests and returns the response
 fn __produce_messages(conn_pool: &mut ConnectionPool,
                       reqs: HashMap<&str, protocol::ProduceRequest>,
                       no_acks: bool)
@@ -1369,7 +1369,7 @@ fn __produce_messages(conn_pool: &mut ConnectionPool,
     // Call each broker with the request formed earlier
     if no_acks {
         for (host, req) in reqs {
-            try!(__send_noack::<protocol::ProduceRequest, protocol::ProduceResponse>(conn_pool, host, req));
+            try!(__send_noack::<_, protocol::ProduceResponse>(conn_pool, host, req));
         }
         Ok(vec!())
     } else {
diff --git a/src/producer.rs b/src/producer.rs
index d46a5de94..307e02099 100644
--- a/src/producer.rs
+++ b/src/producer.rs
@@ -249,8 +249,14 @@ impl<P: Partitioner> Producer<P> {
               V: AsBytes
     {
         let mut rs = try!(self.send_all(ref_slice(rec)));
-        assert_eq!(1, rs.len());
-        rs.pop().unwrap().offset.map(|_| ())
+        if let RequiredAcks::None = self.config.required_acks {
+            // ~ with no required_acks we get no response and
+            // consider the send-data request blindly as successful
+            Ok(())
+        } else {
+            assert_eq!(1, rs.len());
+            rs.pop().unwrap().offset.map(|_| ())
+        }
     }
 
     /// Synchronously send all of the specified messages to Kafka.


Test patch: