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

Issue:
Stronger validity on produce_message#required_acks
It appears that as of Kafka 0.10 the protocol specifies the following regarding the `RequiredAcks` field in a "produce message request":

> This field indicates how many acknowledgements the servers should receive before responding to the request. If it is 0 the server will not send any response (this is the only case where the server will not reply to a request). If it is 1, the server will wait the data is written to the local log before sending a response. If it is -1 the server will block until the message is committed by all in sync replicas before sending a response.

If anything else is specified by the client, the broker should be responding with the `InvalidRequiredAcksCode(21)` error code. 1) We should verify this. 2) We should prepare for this API already now even for Kafka 0.8 and 0.9 servers. Changes to `kafka-rust`'s API are desired to make this type-safe and prevent clients from making an error accidentally.



Discussion / hints:

Fix patch:
diff --git a/examples/console-producer.rs b/examples/console-producer.rs
index 61e28ffd4..2b430a6ac 100644
--- a/examples/console-producer.rs
+++ b/examples/console-producer.rs
@@ -7,11 +7,11 @@ use std::fs::File;
 use std::io::{self, stdin, stderr, Write, BufRead, BufReader};
 use std::ops::{Deref, DerefMut};
 
-use kafka::client::{KafkaClient, Compression};
+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: i16 = 1;
+const REQUIRED_ACKS: RequiredAcks = RequiredAcks::One;
 // how long do we allow wainting for the acknowledgement
 const ACK_TIMEOUT: i32 = 100;
 
diff --git a/examples/produce.rs b/examples/produce.rs
index 460babcd1..c33a10f96 100644
--- a/examples/produce.rs
+++ b/examples/produce.rs
@@ -1,7 +1,7 @@
 extern crate kafka;
 extern crate env_logger;
 
-use kafka::producer::{Producer, Record};
+use kafka::producer::{Producer, Record, RequiredAcks};
 use kafka::error::Error as KafkaError;
 
 /// This program demonstrates sending single message through a
@@ -33,7 +33,7 @@ fn produce_message<'a, 'b>(data: &'a [u8], topic: &'b str, brokers: Vec<String>)
              // ~ give the brokers one second time to ack the message
              .with_ack_timeout(1000)
              // ~ require only one broker to ack the message
-             .with_required_acks(1)
+             .with_required_acks(RequiredAcks::One)
              // ~ build the producer with the above settings
              .create());
 
diff --git a/src/client/mod.rs b/src/client/mod.rs
index 4627fec71..ad2c20ec5 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -278,6 +278,27 @@ impl<'a> AsRef<CommitOffset<'a>> for CommitOffset<'a> {
 
 // --------------------------------------------------------------------
 
+/// Possible choices on acknowledgement requirements when
+/// producing/sending messages to Kafka. See
+/// `KafkaClient::produce_messages`.
+#[derive(Debug, Copy, Clone)]
+pub enum RequiredAcks {
+    /// Indicates to the receiving Kafka broker not to acknowlegde
+    /// messages sent to it at all. Sending messages with this
+    /// acknowledgement requirement translates into a fire-and-forget
+    /// scenario which - of course - is very fast but not reliable.
+    None = 0,
+    /// Requires the receiving Kafka broker to wait until the sent
+    /// messages are written to local disk.  Such messages can be
+    /// regarded as acknowledged by one broker in the cluster.
+    One = 1,
+    /// Requires the sent messages to be acknowledged by all in-sync
+    /// replicas of the targeted topic partitions.
+    All = -1,
+}
+
+// --------------------------------------------------------------------
+
 /// Message data to be sent/produced to a particular topic partition.
 /// See `KafkaClient::produce_messages` and `Producer::send`.
 #[derive(Debug)]
@@ -954,22 +975,13 @@ impl KafkaClient {
     /// Send a message to Kafka
     ///
     /// `required_acks` - indicates how many acknowledgements the
-    /// servers should receive before responding to the request. If it
-    /// is 0 the server will not send any response (this is the only
-    /// case where the server will not reply to a request).  If it is
-    /// 1, the server will wait the data is written to the local log
-    /// before sending a response. If it is -1 the server will block
-    /// until the message is committed by all in sync replicas before
-    /// sending a response. For any number > 1 the server will block
-    /// waiting for this number of acknowledgements to occur (but the
-    /// server will never wait for more acknowledgements than there
-    /// are in-sync replicas).
-    ///
-    /// `ack_timeout` - This provides a maximum time in milliseconds
-    /// the server can await the receipt of the number of
-    /// acknowledgements in `required_acks`
-    ///
-    /// `input` - A set of `ProduceMessage`s
+    /// servers should receive before responding to the request
+    ///
+    /// `ack_timeout` - provides a maximum time in milliseconds the
+    /// server can await the receipt of the number of acknowledgements
+    /// in `required_acks`
+    ///
+    /// `input` - the set of `ProduceMessage`s to send
     ///
     /// Note: Unlike the higher-level `Producer` API, this method will
     /// *not* automatically determine the partition to deliver the
@@ -982,13 +994,13 @@ impl KafkaClient {
     /// # Example
     ///
     /// ```no_run
-    /// use kafka::client::{KafkaClient, ProduceMessage};
+    /// use kafka::client::{KafkaClient, ProduceMessage, RequiredAcks};
     ///
     /// let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
     /// client.load_metadata_all().unwrap();
     /// let req = vec![ProduceMessage::new("my-topic", 0, None, Some("a".as_bytes())),
     ///                ProduceMessage::new("my-topic-2", 0, None, Some("b".as_bytes()))];
-    /// println!("{:?}", client.produce_messages(1, 100, req));
+    /// println!("{:?}", client.produce_messages(RequiredAcks::One, 100, req));
     /// ```
     ///
     /// The return value will contain a vector of topic, partition,
@@ -997,10 +1009,12 @@ impl KafkaClient {
     // XXX rework signaling an error; note that we need to either return the
     // messages which kafka failed to accept or otherwise tell the client about them
 
-    pub fn produce_messages<'a, 'b, I, J>(&mut self, required_acks: i16, ack_timeout: i32, messages: I)
-                                       -> Result<Vec<TopicPartitionOffset>>
+    pub fn produce_messages<'a, 'b, I, J>(&mut self, ack: RequiredAcks, ack_timeout: i32, messages: I)
+                                          -> Result<Vec<TopicPartitionOffset>>
         where J: AsRef<ProduceMessage<'a, 'b>>, I: IntoIterator<Item=J>
     {
+        let required_acks = ack as i16;
+
         let state = &mut self.state;
         let correlation = state.next_correlation_id();
 
diff --git a/src/producer.rs b/src/producer.rs
index 49a4812e8..d46a5de94 100644
--- a/src/producer.rs
+++ b/src/producer.rs
@@ -14,12 +14,12 @@
 //! # Example
 //! ```no_run
 //! use std::fmt::Write;
-//! use kafka::producer::{Producer, Record};
+//! use kafka::producer::{Producer, Record, RequiredAcks};
 //!
 //! let mut producer =
 //!     Producer::from_hosts(vec!("localhost:9092".to_owned()))
 //!         .with_ack_timeout(1000)
-//!         .with_required_acks(1)
+//!         .with_required_acks(RequiredAcks::One)
 //!         .create()
 //!         .unwrap();
 //!
@@ -65,6 +65,7 @@ use std::hash::{Hasher, SipHasher, BuildHasher, BuildHasherDefault};
 use client::{self, KafkaClient};
 // public re-exports
 pub use client::Compression;
+pub use client::RequiredAcks;
 use error::Result;
 use utils::TopicPartitionOffset;
 
@@ -76,11 +77,11 @@ use client::SecurityConfig;
 #[cfg(not(feature = "security"))]
 type SecurityConfig = ();
 
-/// The default value for `Builder::with_ack_timeout`.
+/// The default value for `Builder::with_acks_timeout`.
 pub const DEFAULT_ACK_TIMEOUT: i32 = 30 * 1000;
 
 /// The default value for `Builder::with_required_acks`.
-pub const DEFAULT_REQUIRED_ACKS: i16 = 1;
+pub const DEFAULT_REQUIRED_ACKS: RequiredAcks = RequiredAcks::One;
 
 // --------------------------------------------------------------------
 
@@ -219,7 +220,7 @@ struct Config {
     ack_timeout: i32,
     /// The number of acks to request. See
     /// `KafkaClient::produce_messages`.
-    required_acks: i16,
+    required_acks: RequiredAcks,
 }
 
 impl Producer {
@@ -317,7 +318,7 @@ pub struct Builder<P = DefaultPartitioner> {
     hosts: Vec<String>,
     compression: Compression,
     ack_timeout: i32,
-    required_acks: i16,
+    required_acks: RequiredAcks,
     partitioner: P,
     security_config: Option<SecurityConfig>,
 }
@@ -365,17 +366,11 @@ impl Builder {
     }
 
     /// Sets how many acknowledgements the kafka brokers should
-    /// receive before responding to sent messages.  If it is 0 the
-    /// servers will not send any response.  If it is 1, the server
-    /// will wait the data is written to the local server log before
-    /// sending a replying.  If it is -1 the servers will block until
-    /// the messages are committed by all in sync replicas before
-    /// replaying.  For any number `> 1` the servers will block
-    /// waiting for this number of acknowledgements to occur (but the
-    /// servers will never wait for more acknowledgements than there
-    /// are in-sync replicas).
-    pub fn with_required_acks(mut self, required_acks: i16) -> Self {
-        self.required_acks = required_acks;
+    /// receive before responding to sent messages.
+    ///
+    /// See `RequiredAcks`.
+    pub fn with_required_acks(mut self, acks: RequiredAcks) -> Self {
+        self.required_acks = acks;
         self
     }
 }


Test patch: