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

Issue:
client/consumer/producer: time api
Rework the public API such that it accepts `std::time::Duration` in places where it currently expects the user to provide a duration in milliseconds. This will be less prone to unit mismatches and errors by mistake. So far the places identified are:
- `kafka::producer::Builder::with_ack_timeout`
- `kafka::client::KafkaClient::produce_messages` (ack_timeout parameter)
- `kafka::client::KafkaClient::set_retry_backoff_time` and `kafka::client::KafkaClient::retry_backoff_time`
- `kafka::client::KafkaClient::fetch_max_wait_time` and `kafka::client::KafkaClient::set_fetch_max_wait_time`
- `kafka::consumer::Builder::with_fetch_max_wait_time`

Since `const_fn` is not stable yet (and `Duration` constructor functions are not `const`), we'll keep the type of the following but rename them for sake of clarity:
- `kafka::producer::DEFAULT_ACK_TIMEOUT` -> `DEFAULT_ACK_TIMEOUT_MILLIS`
- `kafka::client::DEFAULT_RETRY_BACKOFF_TIME` -> `DEFAULT_RETRY_BACKOFF_TIME_MILLIS`
- `kafka::client::DEFAULT_FETCH_MAX_WAIT_TIME` -> `DEFAULT_FETCH_MAX_WAIT_TIME_MILLIS`

Internally, we might persist the values as milliseconds if necessary to avoid repeated unnecessary conversions from the duration representation. However, the conversion of a `Duration` to an `i32` (as often required by the kafka protocol) might fail; we shall introduce an `InvalidDuration` error  (ideally with some information about which duration was found to be invalid) and fail if the conversion is not safely possible: For `Producer` and `Consumer` we would fail upon the call to their `Builder::create()` method. In `KafkaClient` we'll be failing lazily in `fetch_message` and `produce_messages`.

This will be a breaking change of course.

All new, future public API aspects around a duration shall be represented with the information being exposed as a `std::time::Duration`.



Discussion / hints:
@xitep Could:

`kafka::producer::DEFAULT_ACK_TIMEOUT` -> `DEFAULT_ACK_TIMEOUT_MILLIS`
`kafka::client::DEFAULT_RETRY_BACKOFF_TIME` -> `DEFAULT_RETRY_BACKOFF_TIME_MILLIS`
`kafka::client::DEFAULT_FETCH_MAX_WAIT_TIME` -> `DEFAULT_FETCH_MAX_WAIT_TIME_MILLIS`

Be handled with lazy statics?

@Hoverbear that's a good idea! thanks for the hint, i'll try that out.

it looks like #102 will be merged sooner than this ticket, therefore, the newly added elements from #102 shall be changed as outlined in this ticket; these new newly added elements are (or will be):
- `kafka::client::DEFAULT_CONNECTION_IDLE_TIMEOUT`
- `kafka::client::KafkaClient::set_connection_idle_timeout` and `kafka::client::KafkaClient::connection_idle_timeout`
- `kafka::producer::Builder::with_connection_idle_timeout`
- `kafka::consumer::Builder::with_connection_idle_timeout`

@Hoverbear just some feedback on the `lazy_static` approach; it works, but i'm not really satisfied with the resulting api. first, the "constants" need to be de-referenced explicitly. secondly, in the generated documentation they show up in a different section than the rest of the constants.

i'll go with the originally proposed re-naming then. anyway, many thanks for the idea!


Fix patch:
diff --git a/examples/console-consumer.rs b/examples/console-consumer.rs
index 939e1a61c..af5d6342b 100644
--- a/examples/console-consumer.rs
+++ b/examples/console-consumer.rs
@@ -3,6 +3,7 @@ extern crate getopts;
 extern crate env_logger;
 
 use std::{env, fmt, process};
+use std::time::Duration;
 use std::io::{self, Write};
 use std::ascii::AsciiExt;
 
@@ -31,7 +32,7 @@ fn process(cfg: Config) -> Result<(), Error> {
         let mut cb = Consumer::from_hosts(cfg.brokers)
             .with_group(cfg.group)
             .with_fallback_offset(cfg.fallback_offset)
-            .with_fetch_max_wait_time(1000)
+            .with_fetch_max_wait_time(Duration::from_secs(1))
             .with_fetch_min_bytes(1_000)
             .with_fetch_max_bytes_per_partition(100_000)
             .with_retry_max_bytes_limit(1_000_000)
diff --git a/examples/console-producer.rs b/examples/console-producer.rs
index 0c3f692cc..dbaa96649 100644
--- a/examples/console-producer.rs
+++ b/examples/console-producer.rs
@@ -4,14 +4,14 @@ extern crate env_logger;
 
 use std::{env, fmt, process};
 use std::fs::File;
+use std::str::FromStr;
 use std::io::{self, stdin, stderr, Write, BufRead, BufReader};
 use std::ops::{Deref, DerefMut};
+use std::time::Duration;
 
-use kafka::client::{KafkaClient, Compression, RequiredAcks, DEFAULT_CONNECTION_IDLE_TIMEOUT};
-use kafka::producer::{AsBytes, Producer, Record};
-
-// how long do we allow waiting for the acknowledgement
-const ACK_TIMEOUT: i32 = 100;
+use kafka::client::{KafkaClient, Compression, RequiredAcks,
+                    DEFAULT_CONNECTION_IDLE_TIMEOUT_MILLIS};
+use kafka::producer::{AsBytes, Producer, Record, DEFAULT_ACK_TIMEOUT_MILLIS};
 
 /// This is a very simple command line application sending every
 /// non-empty line of standard input to a specified kafka topic; one
@@ -58,7 +58,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_ack_timeout(cfg.ack_timeout)
                             .with_required_acks(cfg.required_acks)
                             .with_compression(cfg.compression)
                             .with_connection_idle_timeout(cfg.conn_idle_timeout)
@@ -196,7 +196,8 @@ struct Config {
     compression: Compression,
     required_acks: RequiredAcks,
     batch_size: usize,
-    conn_idle_timeout: u32,
+    conn_idle_timeout: Duration,
+    ack_timeout: Duration,
 }
 
 impl Config {
@@ -212,8 +213,9 @@ impl Config {
         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("", "ack-timeout", "Specify time to wait for acks", "MILLIS");
         opts.optopt("", "batch-size", "Send N message in one batch.", "N");
-        opts.optopt("", "idle-timeout", "Specify timeout for idle connections", "SECS");
+        opts.optopt("", "idle-timeout", "Specify timeout for idle connections", "MILLIS");
 
         let m = match opts.parse(&args[1..]) {
             Ok(m) => m,
@@ -248,24 +250,21 @@ impl Config {
                 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") {
-                    None => 1,
-                    Some(s) => {
-                        match s.parse::<usize>() {
-                            Ok(n) => n,
-                            Err(_) => return Err(Error::Literal(format!("Not a number: {}", s))),
-                        }
-                    }
-                }
-            },
-            conn_idle_timeout: match m.opt_str("idle-timeout") {
-                None => DEFAULT_CONNECTION_IDLE_TIMEOUT,
-                Some(s) => match s.parse::<u32>() {
-                    Err(_) => return Err(Error::Literal(format!("Not a number: {}", s))),
-                    Ok(n) => n * 1_000,
-                },
-            }
+            batch_size: try!(to_number(m.opt_str("batch-size"), 1)),
+            conn_idle_timeout: Duration::from_millis(try!(to_number(m.opt_str("idle-timeout"),
+                                                                    DEFAULT_CONNECTION_IDLE_TIMEOUT_MILLIS))),
+            ack_timeout: Duration::from_millis(try!(to_number(m.opt_str("ack-timeout"),
+                                                              DEFAULT_ACK_TIMEOUT_MILLIS))),
         })
     }
 }
+
+fn to_number<N: FromStr>(s: Option<String>, _default: N) -> Result<N, Error> {
+    match s {
+        None => Ok(_default),
+        Some(s) => match s.parse::<N>() {
+            Ok(n) => Ok(n),
+            Err(_) => return Err(Error::Literal(format!("Not a number: {}", s))),
+        }
+    }
+}
diff --git a/examples/example-produce.rs b/examples/example-produce.rs
index 5f0c6dca4..508e6a917 100644
--- a/examples/example-produce.rs
+++ b/examples/example-produce.rs
@@ -1,6 +1,8 @@
 extern crate kafka;
 extern crate env_logger;
 
+use std::time::Duration;
+
 use kafka::producer::{Producer, Record, RequiredAcks};
 use kafka::error::Error as KafkaError;
 
@@ -31,7 +33,7 @@ fn produce_message<'a, 'b>(data: &'a [u8], topic: &'b str, brokers: Vec<String>)
     let mut producer =
         try!(Producer::from_hosts(brokers)
              // ~ give the brokers one second time to ack the message
-             .with_ack_timeout(1000)
+             .with_ack_timeout(Duration::from_secs(1))
              // ~ require only one broker to ack the message
              .with_required_acks(RequiredAcks::One)
              // ~ build the producer with the above settings
diff --git a/src/client/mod.rs b/src/client/mod.rs
index a2a16927c..481077c97 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -23,6 +23,8 @@ use codecs::{ToByte, FromByte};
 use error::{Result, Error, KafkaCode};
 use protocol::{self, ResponseParser};
 
+use client_internals::KafkaClientInternals;
+
 pub mod metadata;
 mod state;
 mod network;
@@ -49,7 +51,7 @@ fn default_conn_rw_timeout() -> Option<Duration> {
 pub const DEFAULT_COMPRESSION: Compression = Compression::NONE;
 
 /// The default value for `KafkaClient::set_fetch_max_wait_time(..)`
-pub const DEFAULT_FETCH_MAX_WAIT_TIME: i32 = 100; // milliseconds
+pub const DEFAULT_FETCH_MAX_WAIT_TIME_MILLIS: u64 = 100;
 
 /// The default value for `KafkaClient::set_fetch_min_bytes(..)`
 pub const DEFAULT_FETCH_MIN_BYTES: i32 = 4096;
@@ -64,19 +66,15 @@ pub const DEFAULT_FETCH_CRC_VALIDATION: bool = true;
 pub const DEFAULT_GROUP_OFFSET_STORAGE: GroupOffsetStorage = GroupOffsetStorage::Zookeeper;
 
 /// The default value for `KafkaClient::set_retry_backoff_time(..)`
-pub const DEFAULT_RETRY_BACKOFF_TIME: u32 = 100;
+pub const DEFAULT_RETRY_BACKOFF_TIME_MILLIS: u64 = 100;
 
 /// The default value for `KafkaClient::set_retry_max_attempts(..)`
 // the default value: re-attempt a repeatable operation for
 // approximetaly up to two minutes
-pub const DEFAULT_RETRY_MAX_ATTEMPTS: u32 = 120_000 / DEFAULT_RETRY_BACKOFF_TIME;
+pub const DEFAULT_RETRY_MAX_ATTEMPTS: u32 = 120_000 / DEFAULT_RETRY_BACKOFF_TIME_MILLIS as u32;
 
 /// The default value for `KafkaClient::set_connection_idle_timeout(..)`
-pub const DEFAULT_CONNECTION_IDLE_TIMEOUT: u32 = 540_000;
-
-fn default_conn_idle_timeout() -> Duration {
-    Duration::from_millis(DEFAULT_CONNECTION_IDLE_TIMEOUT as u64)
-}
+pub const DEFAULT_CONNECTION_IDLE_TIMEOUT_MILLIS: u64 = 540_000;
 
 /// Client struct keeping track of brokers and topic metadata.
 ///
@@ -114,10 +112,10 @@ struct ClientConfig {
     // storage type.
     offset_fetch_version: protocol::OffsetFetchVersion,
     offset_commit_version: protocol::OffsetCommitVersion,
-    // ~ the number of millis to wait before retrying a failed
+    // ~ the duration to wait before retrying a failed
     // operation like refreshing group coordinators; this avoids
     // operation retries in a tight loop.
-    retry_backoff_time: u32,
+    retry_backoff_time: Duration,
     // ~ the number of repeated retry attempts; prevents endless
     // repetition of a retry attempt
     retry_max_attempts: u32,
@@ -356,17 +354,20 @@ impl KafkaClient {
                 client_id: CLIENTID.to_owned(),
                 hosts: hosts,
                 compression: DEFAULT_COMPRESSION,
-                fetch_max_wait_time: DEFAULT_FETCH_MAX_WAIT_TIME,
+                fetch_max_wait_time: protocol::to_millis_i32(
+                    Duration::from_millis(DEFAULT_FETCH_MAX_WAIT_TIME_MILLIS))
+                    .expect("invalid default-fetch-max-time-millis"),
                 fetch_min_bytes: DEFAULT_FETCH_MIN_BYTES,
                 fetch_max_bytes_per_partition: DEFAULT_FETCH_MAX_BYTES_PER_PARTITION,
                 fetch_crc_validation: DEFAULT_FETCH_CRC_VALIDATION,
                 offset_fetch_version: DEFAULT_GROUP_OFFSET_STORAGE.offset_fetch_version(),
                 offset_commit_version: DEFAULT_GROUP_OFFSET_STORAGE.offset_commit_version(),
-                retry_backoff_time: DEFAULT_RETRY_BACKOFF_TIME,
+                retry_backoff_time: Duration::from_millis(DEFAULT_RETRY_BACKOFF_TIME_MILLIS),
                 retry_max_attempts: DEFAULT_RETRY_MAX_ATTEMPTS,
             },
             conn_pool: network::Connections::new(
-                default_conn_rw_timeout(), default_conn_idle_timeout()),
+                default_conn_rw_timeout(),
+                Duration::from_millis(DEFAULT_CONNECTION_IDLE_TIMEOUT_MILLIS)),
             state: state::ClientState::new(),
         }
     }
@@ -413,17 +414,20 @@ impl KafkaClient {
                 client_id: CLIENTID.to_owned(),
                 hosts: hosts,
                 compression: DEFAULT_COMPRESSION,
-                fetch_max_wait_time: DEFAULT_FETCH_MAX_WAIT_TIME,
+                fetch_max_wait_time: protocol::to_millis_i32(
+                    Duration::from_millis(DEFAULT_FETCH_MAX_WAIT_TIME_MILLIS))
+                    .expect("invalid default-fetch-max-time-millis"),
                 fetch_min_bytes: DEFAULT_FETCH_MIN_BYTES,
                 fetch_max_bytes_per_partition: DEFAULT_FETCH_MAX_BYTES_PER_PARTITION,
                 fetch_crc_validation: DEFAULT_FETCH_CRC_VALIDATION,
                 offset_fetch_version: DEFAULT_GROUP_OFFSET_STORAGE.offset_fetch_version(),
                 offset_commit_version: DEFAULT_GROUP_OFFSET_STORAGE.offset_commit_version(),
-                retry_backoff_time: DEFAULT_RETRY_BACKOFF_TIME,
+                retry_backoff_time: Duration::from_millis(DEFAULT_RETRY_BACKOFF_TIME_MILLIS),
                 retry_max_attempts: DEFAULT_RETRY_MAX_ATTEMPTS,
             },
             conn_pool: network::Connections::new_with_security(
-                default_conn_rw_timeout(), default_conn_idle_timeout(),
+                default_conn_rw_timeout(),
+                Duration::from_millis(DEFAULT_CONNECTION_IDLE_TIMEOUT_MILLIS),
                 Some(security)),
             state: state::ClientState::new(),
         }
@@ -465,15 +469,16 @@ impl KafkaClient {
     /// See also `KafkaClient::set_fetch_min_bytes(..)` and
     /// `KafkaClient::set_fetch_max_bytes_per_partition(..)`.
     #[inline]
-    pub fn set_fetch_max_wait_time(&mut self, max_wait_time: i32) {
-        self.config.fetch_max_wait_time = max_wait_time;
+    pub fn set_fetch_max_wait_time(&mut self, max_wait_time: Duration) -> Result<()> {
+        self.config.fetch_max_wait_time = try!(protocol::to_millis_i32(max_wait_time));
+        Ok(())
     }
 
     /// Retrieves the current `KafkaClient::set_fetch_max_wait_time`
     /// setting.
     #[inline]
-    pub fn fetch_max_wait_time(&self) -> i32 {
-        self.config.fetch_max_wait_time
+    pub fn fetch_max_wait_time(&self) -> Duration {
+        Duration::from_millis(self.config.fetch_max_wait_time as u64)
     }
 
     /// Sets the minimum number of bytes of available data to wait for
@@ -490,11 +495,12 @@ impl KafkaClient {
     /// # Example
     ///
     /// ```no_run
+    /// use std::time::Duration;
     /// use kafka::client::{KafkaClient, FetchPartition};
     ///
     /// let mut client = KafkaClient::new(vec!["localhost:9092".to_owned()]);
     /// client.load_metadata_all().unwrap();
-    /// client.set_fetch_max_wait_time(100);
+    /// client.set_fetch_max_wait_time(Duration::from_millis(100));
     /// client.set_fetch_min_bytes(64 * 1024);
     /// let r = client.fetch_messages(&[FetchPartition::new("my-topic", 0, 0)]);
     /// ```
@@ -596,17 +602,17 @@ impl KafkaClient {
         }
     }
 
-    /// Specifies the number of milliseconds to wait before retrying a
-    /// failed, repeatable operation against Kafka.  This avoids
-    /// retrying such operations in a tight loop.
+    /// Specifies the time to wait before retrying a failed,
+    /// repeatable operation against Kafka.  This avoids retrying such
+    /// operations in a tight loop.
     #[inline]
-    pub fn set_retry_backoff_time(&mut self, millis: u32) {
-        self.config.retry_backoff_time = millis;
+    pub fn set_retry_backoff_time(&mut self, time: Duration) {
+        self.config.retry_backoff_time = time;
     }
 
     /// Retrieves the current `KafkaClient::set_retry_backoff_time`
     /// setting.
-    pub fn retry_backoff_time(&self) -> u32 {
+    pub fn retry_backoff_time(&self) -> Duration {
         self.config.retry_backoff_time
     }
 
@@ -625,24 +631,22 @@ impl KafkaClient {
         self.config.retry_max_attempts
     }
 
-    /// Specifies the timeout (in number of milliseconds) after which
-    /// idle connections will transparently be closed/re-established
-    /// by `KafkaClient`.
+    /// Specifies the timeout after which idle connections will
+    /// transparently be closed/re-established by `KafkaClient`.
     ///
     /// To be effective this value must be smaller than the [remote
     /// broker's `connections.max.idle.ms`
     /// setting](https://kafka.apache.org/documentation.html#brokerconfigs).
     #[inline]
-    pub fn set_connection_idle_timeout(&mut self, millis: u32) {
-        self.conn_pool.set_idle_timeout(Duration::from_millis(millis as u64))
+    pub fn set_connection_idle_timeout(&mut self, timeout: Duration) {
+        self.conn_pool.set_idle_timeout(timeout);
     }
 
     /// Retrieves the current
     /// `KafkaClient::set_connection_idle_timeout` setting.
     #[inline]
-    pub fn connection_idle_timeout(&self) -> u32 {
-        let d = self.conn_pool.idle_timeout();
-        d.as_secs() as u32 * 1_000 + d.subsec_nanos() / 1_000_000
+    pub fn connection_idle_timeout(&self) -> Duration {
+        self.conn_pool.idle_timeout()
     }
 
     /// Provides a view onto the currently loaded metadata of known .
@@ -958,13 +962,14 @@ impl KafkaClient {
     /// # Example
     ///
     /// ```no_run
+    /// use std::time::Duration;
     /// 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(RequiredAcks::One, 100, req));
+    /// println!("{:?}", client.produce_messages(RequiredAcks::One, Duration::from_millis(100), req));
     /// ```
     ///
     /// The return value will contain a vector of topic, partition,
@@ -973,31 +978,13 @@ 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, ack: RequiredAcks, ack_timeout: i32, messages: I)
-                                          -> Result<Vec<TopicPartitionOffset>>
+    pub fn produce_messages<'a, 'b, I, J>(
+        &mut self, acks: RequiredAcks, ack_timeout: Duration, 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();
-
-        // ~ map topic and partition to the corresponding brokers
-        let config = &self.config;
-        let mut reqs: HashMap<&str, protocol::ProduceRequest> = HashMap::new();
-        for msg in messages {
-            let msg = msg.as_ref();
-            match state.find_broker(msg.topic, msg.partition) {
-                None => return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)),
-                Some(broker) => reqs.entry(broker)
-                    .or_insert_with(
-                        || protocol::ProduceRequest::new(
-                            required_acks, ack_timeout, correlation,
-                            &config.client_id, config.compression))
-                    .add(msg.topic, msg.partition, msg.key, msg.value),
-            }
-        }
-        __produce_messages(&mut self.conn_pool, reqs, required_acks == 0)
+        self.internal_produce_messages(
+            acks as i16, try!(protocol::to_millis_i32(ack_timeout)), messages)
     }
 
     /// Commit offset for a topic partitions on behalf of a consumer group.
@@ -1132,6 +1119,35 @@ impl KafkaClient {
     }
 }
 
+impl KafkaClientInternals for KafkaClient {
+    fn internal_produce_messages<'a, 'b, I, J>(
+        &mut self, required_acks: i16, ack_timeout: i32, messages: I)
+        -> Result<Vec<TopicPartitionOffset>>
+        where J: AsRef<ProduceMessage<'a, 'b>>, I: IntoIterator<Item=J>
+    {
+        let state = &mut self.state;
+        let correlation = state.next_correlation_id();
+
+        // ~ map topic and partition to the corresponding brokers
+        let config = &self.config;
+        let mut reqs: HashMap<&str, protocol::ProduceRequest> = HashMap::new();
+        for msg in messages {
+            let msg = msg.as_ref();
+            match state.find_broker(msg.topic, msg.partition) {
+                None => return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)),
+                Some(broker) => reqs.entry(broker)
+                    .or_insert_with(
+                        || protocol::ProduceRequest::new(
+                            required_acks, ack_timeout, correlation,
+                            &config.client_id, config.compression))
+                    .add(msg.topic, msg.partition, msg.key, msg.value),
+            }
+        }
+        __produce_messages(&mut self.conn_pool, reqs, required_acks == 0)
+    }
+
+}
+
 fn __get_group_coordinator<'a>(group: &str,
                                state: &'a mut state::ClientState,
                                conn_pool: &mut network::Connections,
@@ -1450,7 +1466,5 @@ fn __get_response_size(conn: &mut network::KafkaConnection) -> Result<i32> {
 /// Suspends the calling thread for the configured "retry" time. This
 /// method should be called _only_ as part of a retry attempt.
 fn __retry_sleep(cfg: &ClientConfig) {
-    if cfg.retry_backoff_time > 0 {
-        thread::sleep(Duration::from_millis(cfg.retry_backoff_time as u64))
-    }
+    thread::sleep(cfg.retry_backoff_time)
 }
diff --git a/src/client_internals.rs b/src/client_internals.rs
new file mode 100644
index 000000000..0ced10a89
--- /dev/null
+++ b/src/client_internals.rs
@@ -0,0 +1,12 @@
+//! A crate private module to expose `KafkaClient` internals for use
+//! within this crate but not outside of it.
+
+use client;
+use error::Result;
+
+pub trait KafkaClientInternals {
+    fn internal_produce_messages<'a, 'b, I, J>(
+        &mut self, required_acks: i16, ack_timeout: i32, messages: I)
+        -> Result<Vec<client::TopicPartitionOffset>>
+        where J: AsRef<client::ProduceMessage<'a, 'b>>, I: IntoIterator<Item=J>;
+}
diff --git a/src/consumer/builder.rs b/src/consumer/builder.rs
index a2e8689b4..01e51b296 100644
--- a/src/consumer/builder.rs
+++ b/src/consumer/builder.rs
@@ -1,4 +1,5 @@
 use std::collections::HashMap;
+use std::time::Duration;
 
 use client::{self, KafkaClient, FetchOffset, GroupOffsetStorage};
 use error::{Error, Result};
@@ -23,14 +24,14 @@ pub struct Builder {
     group: String,
     assignments: HashMap<String, Vec<i32>>,
     fallback_offset: FetchOffset,
-    fetch_max_wait_time: i32,
+    fetch_max_wait_time: Duration,
     fetch_min_bytes: i32,
     fetch_max_bytes_per_partition: i32,
     retry_max_bytes_limit: i32,
     fetch_crc_validation: bool,
     security_config: Option<SecurityConfig>,
     group_offset_storage: GroupOffsetStorage,
-    conn_idle_timeout: u32,
+    conn_idle_timeout: Duration,
 }
 
 // ~ public only to be shared inside the kafka crate; not supposed to
@@ -39,7 +40,7 @@ pub fn new(client: Option<KafkaClient>, hosts: Vec<String>) -> Builder {
     let mut b = Builder {
         client: client,
         hosts: hosts,
-        fetch_max_wait_time: client::DEFAULT_FETCH_MAX_WAIT_TIME,
+        fetch_max_wait_time: Duration::from_millis(client::DEFAULT_FETCH_MAX_WAIT_TIME_MILLIS),
         fetch_min_bytes: client::DEFAULT_FETCH_MIN_BYTES,
         fetch_max_bytes_per_partition: client::DEFAULT_FETCH_MAX_BYTES_PER_PARTITION,
         fetch_crc_validation: client::DEFAULT_FETCH_CRC_VALIDATION,
@@ -49,7 +50,7 @@ pub fn new(client: Option<KafkaClient>, hosts: Vec<String>) -> Builder {
         fallback_offset: DEFAULT_FALLBACK_OFFSET,
         security_config: None,
         group_offset_storage: client::DEFAULT_GROUP_OFFSET_STORAGE,
-        conn_idle_timeout: client::DEFAULT_CONNECTION_IDLE_TIMEOUT,
+        conn_idle_timeout: Duration::from_millis(client::DEFAULT_CONNECTION_IDLE_TIMEOUT_MILLIS),
     };
     if let Some(ref c) = b.client {
         b.fetch_max_wait_time = c.fetch_max_wait_time();
@@ -127,7 +128,7 @@ impl Builder {
     }
 
     /// See `KafkaClient::set_fetch_max_wait_time`
-    pub fn with_fetch_max_wait_time(mut self, max_wait_time: i32) -> Builder {
+    pub fn with_fetch_max_wait_time(mut self, max_wait_time: Duration) -> Builder {
         self.fetch_max_wait_time = max_wait_time;
         self
     }
@@ -185,7 +186,7 @@ impl Builder {
 
     /// Specifies the timeout for idle connections.
     /// See `KafkaClient::set_connection_idle_timeout`.
-    pub fn with_connection_idle_timeout(mut self, timeout: u32) -> Self {
+    pub fn with_connection_idle_timeout(mut self, timeout: Duration) -> Self {
         self.conn_idle_timeout = timeout;
         self
     }
@@ -215,21 +216,24 @@ impl Builder {
         if self.assignments.is_empty() {
             return Err(Error::NoTopicsAssigned);
         }
-
-        let mut client = match self.client {
-            Some(client) => client,
+        // ~ create the client if necessary
+        let (mut client, need_metadata) = match self.client {
+            Some(client) => (client, false),
             None => {
-                let mut client = Self::new_kafka_client(self.hosts, self.security_config);
-                try!(client.load_metadata_all());
-                client
+                (Self::new_kafka_client(self.hosts, self.security_config), true)
             }
         };
-        client.set_fetch_max_wait_time(self.fetch_max_wait_time);
+        // ~ apply configuration settings
+        try!(client.set_fetch_max_wait_time(self.fetch_max_wait_time));
         client.set_fetch_min_bytes(self.fetch_min_bytes);
         client.set_fetch_max_bytes_per_partition(self.fetch_max_bytes_per_partition);
         client.set_group_offset_storage(self.group_offset_storage);
         client.set_connection_idle_timeout(self.conn_idle_timeout);
-
+        // ~ load metadata if necessary
+        if need_metadata {
+            try!(client.load_metadata_all());
+        }
+        // ~ load consumer state
         let config = Config {
             group: self.group,
             fallback_offset: self.fallback_offset,
diff --git a/src/error.rs b/src/error.rs
index 7e52d06b1..d2ab4ddae 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -47,6 +47,8 @@ pub enum Error {
     NoHostReachable,
     /// Unable to set up `Consumer` due to missing topic assignments
     NoTopicsAssigned,
+    /// An invalid user-provided duration
+    InvalidDuration,
 }
 
 /// Various errors reported by a remote Kafka server.
@@ -207,6 +209,7 @@ impl Clone for Error {
             &Error::StringDecodeError => Error::StringDecodeError,
             &Error::NoHostReachable => Error::NoHostReachable,
             &Error::NoTopicsAssigned => Error::NoTopicsAssigned,
+            &Error::InvalidDuration => Error::InvalidDuration,
         }
     }
 }
@@ -227,6 +230,7 @@ impl error::Error for Error {
             Error::StringDecodeError => "String decoding error",
             Error::NoHostReachable => "No host reachable",
             Error::NoTopicsAssigned => "No topic assigned",
+            Error::InvalidDuration => "Invalid duration",
         }
     }
 
@@ -254,6 +258,7 @@ impl fmt::Display for Error {
             Error::StringDecodeError => write!(f, "String decoding error"),
             Error::NoHostReachable => write!(f, "No host reachable"),
             Error::NoTopicsAssigned => write!(f, "No topic assigned"),
+            Error::InvalidDuration => write!(f, "Invalid duration"),
         }
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index e13d96c8c..5a22c559e 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -31,6 +31,7 @@ extern crate test;
 
 pub mod error;
 pub mod client;
+mod client_internals;
 pub mod consumer;
 pub mod producer;
 mod utils;
@@ -39,3 +40,10 @@ mod protocol;
 mod compression;
 
 pub use self::error::{Error, Result};
+
+trait KafkaClientInternals {
+    fn internal_produce_messages<'a, 'b, I, J>(
+        &mut self, ack: client::RequiredAcks, ack_timeout: i32, messages: I)
+        -> Result<Vec<client::TopicPartitionOffset>>
+        where J: AsRef<client::ProduceMessage<'a, 'b>>, I: IntoIterator<Item=J>;
+}
diff --git a/src/producer.rs b/src/producer.rs
index dddac7a42..d0271304a 100644
--- a/src/producer.rs
+++ b/src/producer.rs
@@ -14,11 +14,12 @@
 //! # Example
 //! ```no_run
 //! use std::fmt::Write;
+//! use std::time::Duration;
 //! use kafka::producer::{Producer, Record, RequiredAcks};
 //!
 //! let mut producer =
 //!     Producer::from_hosts(vec!("localhost:9092".to_owned()))
-//!         .with_ack_timeout(1000)
+//!         .with_ack_timeout(Duration::from_secs(1))
 //!         .with_required_acks(RequiredAcks::One)
 //!         .create()
 //!         .unwrap();
@@ -61,6 +62,7 @@
 use std::collections::HashMap;
 use std::fmt;
 use std::hash::{Hasher, SipHasher, BuildHasher, BuildHasherDefault};
+use std::time::Duration;
 
 use client::{self, KafkaClient};
 // public re-exports
@@ -77,8 +79,11 @@ use client::SecurityConfig;
 #[cfg(not(feature = "security"))]
 type SecurityConfig = ();
 
-/// The default value for `Builder::with_acks_timeout`.
-pub const DEFAULT_ACK_TIMEOUT: i32 = 30 * 1000;
+use client_internals::KafkaClientInternals;
+use protocol;
+
+/// The default value for `Builder::with_ack_timeout`.
+pub const DEFAULT_ACK_TIMEOUT_MILLIS: u64 = 30 * 1000;
 
 /// The default value for `Builder::with_required_acks`.
 pub const DEFAULT_REQUIRED_ACKS: RequiredAcks = RequiredAcks::One;
@@ -215,12 +220,12 @@ struct State<P> {
 }
 
 struct Config {
-    /// The maximum time in millis to wait for acknowledgements. See
+    /// The maximum time to wait for acknowledgements. See
     /// `KafkaClient::produce_messages`.
     ack_timeout: i32,
     /// The number of acks to request. See
     /// `KafkaClient::produce_messages`.
-    required_acks: RequiredAcks,
+    required_acks: i16,
 }
 
 impl Producer {
@@ -249,7 +254,7 @@ impl<P: Partitioner> Producer<P> {
               V: AsBytes
     {
         let mut rs = try!(self.send_all(ref_slice(rec)));
-        if let RequiredAcks::None = self.config.required_acks {
+        if self.config.required_acks == 0 {
             // ~ with no required_acks we get no response and
             // consider the send-data request blindly as successful
             Ok(())
@@ -271,18 +276,19 @@ impl<P: Partitioner> Producer<P> {
         let client = &mut self.client;
         let config = &self.config;
 
-        client.produce_messages(config.required_acks,
-                                config.ack_timeout,
-                                recs.into_iter().map(|r| {
-                                    let mut m = client::ProduceMessage {
-                                        key: to_option(r.key.as_bytes()),
-                                        value: to_option(r.value.as_bytes()),
-                                        topic: r.topic,
-                                        partition: r.partition,
-                                    };
-                                    partitioner.partition(Topics::new(partitions), &mut m);
-                                    m
-                                }))
+        client.internal_produce_messages(
+            config.required_acks,
+            config.ack_timeout,
+            recs.into_iter().map(|r| {
+                let mut m = client::ProduceMessage {
+                    key: to_option(r.key.as_bytes()),
+                    value: to_option(r.value.as_bytes()),
+                    topic: r.topic,
+                    partition: r.partition,
+                };
+                partitioner.partition(Topics::new(partitions), &mut m);
+                m
+            }))
     }
 }
 
@@ -323,8 +329,8 @@ pub struct Builder<P = DefaultPartitioner> {
     client: Option<KafkaClient>,
     hosts: Vec<String>,
     compression: Compression,
-    ack_timeout: i32,
-    conn_idle_timeout: u32,
+    ack_timeout: Duration,
+    conn_idle_timeout: Duration,
     required_acks: RequiredAcks,
     partitioner: P,
     security_config: Option<SecurityConfig>,
@@ -336,8 +342,8 @@ impl Builder {
             client: client,
             hosts: hosts,
             compression: client::DEFAULT_COMPRESSION,
-            ack_timeout: DEFAULT_ACK_TIMEOUT,
-            conn_idle_timeout: client::DEFAULT_CONNECTION_IDLE_TIMEOUT,
+            ack_timeout: Duration::from_millis(DEFAULT_ACK_TIMEOUT_MILLIS),
+            conn_idle_timeout: Duration::from_millis(client::DEFAULT_CONNECTION_IDLE_TIMEOUT_MILLIS),
             required_acks: DEFAULT_REQUIRED_ACKS,
             partitioner: DefaultPartitioner::default(),
             security_config: None,
@@ -365,18 +371,18 @@ impl Builder {
         self
     }
 
-    /// Sets the maximum time in milliseconds the kafka brokers can
-    /// await the receipt of required acknowledgements (which is
-    /// specified through `Builder::with_required_acks`.)  Note that
-    /// Kafka explicitely documents this not to be a hard limit.
-    pub fn with_ack_timeout(mut self, timeout: i32) -> Self {
+    /// Sets the maximum time the kafka brokers can await the receipt
+    /// of required acknowledgements (which is specified through
+    /// `Builder::with_required_acks`.)  Note that Kafka explicitely
+    /// documents this not to be a hard limit.
+    pub fn with_ack_timeout(mut self, timeout: Duration) -> Self {
         self.ack_timeout = timeout;
         self
     }
 
     /// Specifies the timeout for idle connections.
     /// See `KafkaClient::set_connection_idle_timeout`.
-    pub fn with_connection_idle_timeout(mut self, timeout: u32) -> Self {
+    pub fn with_connection_idle_timeout(mut self, timeout: Duration) -> Self {
         self.conn_idle_timeout = timeout;
         self
     }
@@ -424,25 +430,30 @@ impl<P> Builder<P> {
     /// Finally creates/builds a new producer based on the so far
     /// supplied settings.
     pub fn create(self) -> Result<Producer<P>> {
-        let mut client = match self.client {
-            Some(client) => client,
+        // ~ create the client if necessary
+        let (mut client, need_metadata) = match self.client {
+            Some(client) => (client, false),
             None => {
-                let mut client = Self::new_kafka_client(self.hosts, self.security_config);
-                try!(client.load_metadata_all());
-                client
+                (Self::new_kafka_client(self.hosts, self.security_config), true)
             }
         };
+        // ~ apply configuration settings
         client.set_compression(self.compression);
         client.set_connection_idle_timeout(self.conn_idle_timeout);
-        let state = try!(State::new(&mut client, self.partitioner));
-        let config = Config {
-            ack_timeout: self.ack_timeout,
-            required_acks: self.required_acks,
+        let producer_config = Config {
+            ack_timeout: try!(protocol::to_millis_i32(self.ack_timeout)),
+            required_acks: self.required_acks as i16,
         };
+        // ~ load metadata if necessary
+        if need_metadata {
+            try!(client.load_metadata_all());
+        }
+        // ~ create producer state
+        let state = try!(State::new(&mut client, self.partitioner));
         Ok(Producer {
             client: client,
             state: state,
-            config: config,
+            config: producer_config,
         })
     }
 }
diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs
index f02950c60..fe789779a 100644
--- a/src/protocol/mod.rs
+++ b/src/protocol/mod.rs
@@ -1,5 +1,6 @@
 use std::io::{Read, Write};
 use std::mem;
+use std::time::Duration;
 
 use codecs::{ToByte, FromByte};
 use crc::crc32;
@@ -151,3 +152,44 @@ impl FromByte for HeaderResponse {
 pub fn to_crc(data: &[u8]) -> u32 {
     crc32::checksum_ieee(data)
 }
+
+// --------------------------------------------------------------------
+
+/// Safely converts a Duration into the number of milliseconds as a
+/// i32 as often required in the kafka protocol.
+pub fn to_millis_i32(d: Duration) -> Result<i32> {
+    use std::i32;
+    let m = d.as_secs().saturating_mul(1_000).saturating_add((d.subsec_nanos() / 1_000_000) as u64);
+    if m > i32::MAX as u64 {
+        Err(Error::InvalidDuration)
+    } else {
+        Ok(m as i32)
+    }
+}
+
+#[test]
+fn test_to_millis_i32() {
+    use std::{i32, u32, u64};
+
+    fn assert_invalid(d: Duration) {
+        match to_millis_i32(d) {
+            Err(Error::InvalidDuration) => {}
+            other => panic!("Expected Err(InvalidDuration) but got {:?}", other),
+        }
+    }
+    fn assert_valid(d: Duration, expected_millis: i32) {
+        let r = to_millis_i32(d);
+        match r {
+            Ok(m) =>
+                assert_eq!(expected_millis, m),
+            Err(e) =>
+                panic!("Expected Ok({}) but got Err({:?})", expected_millis, e),
+        }
+    }
+    assert_valid(Duration::from_millis(1_234), 1_234);
+    assert_valid(Duration::new(540, 123_456_789), 540_123);
+    assert_invalid(Duration::from_millis(u64::MAX));
+    assert_invalid(Duration::from_millis(u32::MAX as u64));
+    assert_invalid(Duration::from_millis(i32::MAX as u64 + 1));
+    assert_valid(Duration::from_millis(i32::MAX as u64 - 1), i32::MAX - 1);
+}


Test patch: