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

Issue:
producer: ensure distributing messages with the same key to the same partition
The "key" - if available - of a message is typically used to derive a consistent partition which a message should be delivered to. Usually one computes a hash code of the key and maps that consistently on one of the partitions of the topic.

Clients of the `Producer` API can achieve this already by supplying a custom `Partitioner` to the `Producer`. However, we shall make this the standard behaviour of the `DefaultPartitioner`.

Unfortunately, topic metadata held by the `Producer` (which is supplied by `KafkaClient`) deliver only "available" partitions, which makes it impossible to implement the "consistency" aspect of the requirement. Therefore, this will require an extension to `KafkaClient` as well.

Regarding the hash algorithm we don't need a cryptographic hash function. However, we need to use one which doesn't different depending the platform - this rules out farmhash for example. (it seems the java based producer uses murmur2-32bit, but i don't see the need to use the same since various kafka clients seem not to stick a "default" hash function, e.g. the java based producer uses a different hash function than the scala based one.)



Discussion / hints:
This is a requirement for some work I'm about to do. I'm new to Rust (coming from Java/Scala + Kafka background) but I could take a stab at this unless you're already working on it.

hello @datariot, it wasn't a priority for me right now, but i could give it precedence since the changes will have impact on other issues i planed to resolve. unfortunately, this will include some public api changes, though not as big as in the last 0.2 release.

I feel partitioning by key is a pretty important part of a scalable Kafka project, so if it's going to impact other things it might be good to prioritize. Also it would selfishly help me ;-)


Fix patch:
diff --git a/examples/console-producer.rs b/examples/console-producer.rs
index ca2ea12fd..e880aa753 100644
--- a/examples/console-producer.rs
+++ b/examples/console-producer.rs
@@ -110,6 +110,7 @@ fn produce_impl_inbatches(producer: &mut Producer, src: &mut BufRead, cfg: &Conf
             next_rec = 0;
         }
         let mut rec = &mut rec_stash[next_rec];
+        rec.value.clear();
         if try!(src.read_line(&mut rec.value)) == 0 {
             break; // ~ EOF reached
         }
diff --git a/examples/metadata.rs b/examples/metadata.rs
index a98550c06..d69c42469 100644
--- a/examples/metadata.rs
+++ b/examples/metadata.rs
@@ -20,7 +20,12 @@ fn main() {
 
     for t in client.topics() {
         for p in t.partitions() {
-            println!("{}\t{}\t{}\t{}", t.name(), p.id(), p.leader_id(), p.leader_host());
+            match p.leader() {
+                Some(leader) =>
+                    println!("{}\t{}\t{}\t{}", t.name(), p.id(), leader.id(), leader.host()),
+                None =>
+                    println!("{}\t{} => no leader!", t.name(), p.id()),
+            }
         }
     }
 }
diff --git a/examples/produce.rs b/examples/produce.rs
index d0de35d8e..46433cb7f 100644
--- a/examples/produce.rs
+++ b/examples/produce.rs
@@ -3,7 +3,7 @@ extern crate kafka;
 use kafka::producer::{Producer, Record};
 use kafka::error::Error as KafkaError;
 
-/// This program demonstrates sending single messagess through a
+/// This program demonstrates sending single message through a
 /// `Producer`.  This is a convenient higher-level client that will
 /// fit most use cases.
 fn main() {
diff --git a/src/client/metadata.rs b/src/client/metadata.rs
index 00fd9c166..ed33a085c 100644
--- a/src/client/metadata.rs
+++ b/src/client/metadata.rs
@@ -2,14 +2,18 @@
 //! Example: `KafkaClient::topics()`.
 
 use std::collections::hash_map;
-use std::slice;
 use std::fmt;
 
-use super::{KafkaClient, ClientState, TopicPartitions, TopicPartition};
+use super::KafkaClient;
+use super::state::{ClientState, TopicPartitions, TopicPartitionIter, TopicPartition};
+
+// public re-export
+pub use super::state::Broker;
+pub use super::state::TopicNames;
 
 /// A view on the loaded metadata about topics and their partitions.
 pub struct Topics<'a> {
-    state: &'a ClientState
+    state: &'a ClientState,
 }
 
 impl<'a> Topics<'a> {
@@ -23,10 +27,10 @@ impl<'a> Topics<'a> {
     /// Retrieves the number of the underlying topics.
     #[inline]
     pub fn len(&self) -> usize {
-        self.state.topic_partitions.len()
+        self.state.num_topics()
     }
 
-    /// Provides an iterator over the known topics.
+    /// Provides an iterator over the underlying topics.
     #[inline]
     pub fn iter(&'a self) -> TopicIter<'a> {
         TopicIter::new(self.state)
@@ -35,34 +39,26 @@ impl<'a> Topics<'a> {
     /// A conveniece method to return an iterator the topics' names.
     #[inline]
     pub fn names(&'a self) -> TopicNames<'a> {
-        TopicNames { iter: self.state.topic_partitions.keys() }
+        self.state.topic_names()
     }
 
     /// A convenience method to determine whether the specified topic
     /// is known.
     #[inline]
     pub fn contains(&'a self, topic: &str) -> bool {
-        self.state.topic_partitions.contains_key(topic)
+        self.state.contains_topic(topic)
     }
 
-    /// Retrieves the partitions of a known topic.
+    /// Retrieves the partitions of a specified topic.
     #[inline]
     pub fn partitions(&'a self, topic: &str) -> Option<Partitions<'a>> {
-        self.state.topic_partitions.get(topic).map(|tp| Partitions {
-            state: self.state,
-            tp: tp,
+        self.state.partitions_for(topic).map(|tp| {
+            Partitions {
+                state: self.state,
+                tp: tp,
+            }
         })
     }
-
-    /// Retrieves a snapshot/copy of the partition ids available for
-    /// the specified topic.  Note that the returned copy may get out
-    /// of date if the underlying client's metadata gets refreshed.
-    #[inline]
-    pub fn partition_ids(&'a self, topic: &str) -> Option<Vec<i32>> {
-        self.state.topic_partitions.get(topic)
-            .map(|tp| &tp.partitions)
-            .map(|ps| ps.iter().map(|p| p.partition_id).collect())
-    }
 }
 
 impl<'a> fmt::Debug for Topics<'a> {
@@ -107,7 +103,7 @@ impl<'a> TopicIter<'a> {
     fn new(state: &'a ClientState) -> TopicIter<'a> {
         TopicIter {
             state: state,
-            iter: state.topic_partitions.iter()
+            iter: state.topic_partitions().iter(),
         }
     }
 }
@@ -117,29 +113,16 @@ impl<'a> Iterator for TopicIter<'a> {
 
     #[inline]
     fn next(&mut self) -> Option<Self::Item> {
-        self.iter.next().map(|(name, tps) | Topic {
-            state: self.state,
-            name: &name[..],
-            tp: tps,
+        self.iter.next().map(|(name, tps)| {
+            Topic {
+                state: self.state,
+                name: &name[..],
+                tp: tps,
+            }
         })
     }
 }
 
-/// An iterator over the names of topics known to the originating
-/// kafka client.
-pub struct TopicNames<'a> {
-    iter: hash_map::Keys<'a, String, TopicPartitions>,
-}
-
-impl<'a> Iterator for TopicNames<'a> {
-    type Item=&'a str;
-
-    #[inline]
-    fn next(&mut self) -> Option<Self::Item> {
-        self.iter.next().map(|s| &s[..])
-    }
-}
-
 /// A view on the loaded metadata for a particular topic.
 pub struct Topic<'a> {
     state: &'a ClientState,
@@ -154,7 +137,7 @@ impl<'a> Topic<'a> {
         self.name
     }
 
-    /// Retrieves the list of known partitions for this topic.
+    /// Retrieves the list of all partitions for this topic.
     #[inline]
     pub fn partitions(&self) -> Partitions<'a> {
         Partitions {
@@ -162,20 +145,14 @@ impl<'a> Topic<'a> {
             tp: self.tp,
         }
     }
-
-    /// Retrieves a snapshot/copy of the partition ids available for
-    /// this topic.  Note that the returned copy may get out of date
-    /// if the underlying client's metadata gets refreshed.
-    #[inline]
-    pub fn partition_ids(&'a self) -> Vec<i32> {
-        self.tp.partitions.iter().map(|p| p.partition_id).collect()
-    }
 }
 
 impl<'a> fmt::Debug for Topic<'a> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "Topic {{ name: {}, partitions: {:?} }}",
-               self.name, self.partitions())
+        write!(f,
+               "Topic {{ name: {}, partitions: {:?} }}",
+               self.name,
+               self.partitions())
     }
 }
 
@@ -186,28 +163,39 @@ pub struct Partitions<'a> {
 }
 
 impl<'a> Partitions<'a> {
-    /// Retrieves the number of the underlying partitions.
+    /// Retrieves the number of the topic's partitions.
     #[inline]
     pub fn len(&self) -> usize {
-        self.tp.partitions.len()
+        self.tp.len()
     }
 
     /// Tests for `.len() > 0`.
     #[inline]
     pub fn is_empty(&self) -> bool {
-        self.tp.partitions.is_empty()
+        self.tp.is_empty()
     }
 
-    /// Retrieves an iterator of the partitions for the underlying topic.
+    /// Retrieves an iterator of the partitions of the underlying topic.
     #[inline]
     pub fn iter(&self) -> PartitionIter<'a> {
-        PartitionIter::new(self.state, self.tp.partitions.iter())
+        PartitionIter::new(self.state, self.tp)
     }
 
     /// Finds a specified partition identified by its id.
     #[inline]
     pub fn partition(&self, partition_id: i32) -> Option<Partition<'a>> {
-        self.tp.partition(partition_id).map(|p| Partition::new(self.state, p))
+        self.tp.partition(partition_id).map(|p| Partition::new(self.state, p, partition_id))
+    }
+
+    /// Convenience method to retrieve the identifiers of all
+    /// currently "available" partitions.  Such partitions are known
+    /// to have a leader broker and can be sent messages to.
+    #[inline]
+    pub fn available_ids(&self) -> Vec<i32> {
+        self.tp
+            .iter()
+            .filter_map(|(id, p)| p.broker(self.state).map(|_| id))
+            .collect()
     }
 }
 
@@ -239,19 +227,22 @@ impl<'a> IntoIterator for Partitions<'a> {
     type IntoIter=PartitionIter<'a>;
 
     fn into_iter(self) -> Self::IntoIter {
-        PartitionIter::new(self.state, self.tp.partitions.iter())
+        PartitionIter::new(self.state, self.tp)
     }
 }
 
 /// An interator over a topic's partitions.
 pub struct PartitionIter<'a> {
     state: &'a ClientState,
-    iter: slice::Iter<'a, TopicPartition>
+    iter: TopicPartitionIter<'a>,
 }
 
 impl<'a> PartitionIter<'a> {
-    fn new(state: &'a ClientState, iter: slice::Iter<'a, TopicPartition>) -> Self {
-        PartitionIter { state: state, iter: iter }
+    fn new(state: &'a ClientState, tp: &'a TopicPartitions) -> Self {
+        PartitionIter {
+            state: state,
+            iter: tp.iter(),
+        }
     }
 }
 
@@ -260,43 +251,59 @@ impl<'a> Iterator for PartitionIter<'a> {
 
     #[inline]
     fn next(&mut self) -> Option<Self::Item> {
-        self.iter.next().map(|tp| Partition::new(self.state, tp))
+        self.iter.next().map(|(id, p)| Partition::new(self.state, p, id))
     }
 }
 
 /// Metadata about a particular topic partition.
+///
+/// A partition can be seen as either available or not by
+/// `kafka-rust`.  "Available" partitions are partitions with an
+/// assigned leader broker and can be send messages to or fetched
+/// messages from.  Non-available partitions are ignored by
+/// `kafka-rust`.  Whether or not a partition is currently "available"
+/// can be determined by testing for `partition.leader().is_some()` or
+/// more directly through `partition.is_available()`.
 pub struct Partition<'a> {
     state: &'a ClientState,
     partition: &'a TopicPartition,
+    id: i32,
 }
 
 impl<'a> Partition<'a> {
-    fn new(state: &'a ClientState, partition: &'a TopicPartition) -> Partition<'a> {
-        Partition { state: state, partition: partition }
+    fn new(state: &'a ClientState, partition: &'a TopicPartition, id: i32) -> Partition<'a> {
+        Partition {
+            state: state,
+            partition: partition,
+            id: id,
+        }
     }
 
     /// Retrieves the identifier of this topic partition.
     #[inline]
     pub fn id(&self) -> i32 {
-        self.partition.partition_id
+        self.id
     }
 
-    /// Retrives the node_id of the current leader of this partition.
+    /// Retrieves the current leader broker of this partition - if
+    /// any.  A partition with a leader is said to be "available".
     #[inline]
-    pub fn leader_id(&self) -> i32 {
-        self.state.broker_by_partition(self.partition).node_id
+    pub fn leader(&self) -> Option<&Broker> {
+        self.partition.broker(self.state)
     }
 
-    /// Retrieves the host of the current leader of this partition.
-    #[inline]
-    pub fn leader_host(&self) -> &'a str {
-        &self.state.broker_by_partition(self.partition).host
+    /// Determines whether this partition is currently "available".
+    /// See `Partition::leader()`.
+    pub fn is_available(&self) -> bool {
+        self.leader().is_some()
     }
 }
 
 impl<'a> fmt::Debug for Partition<'a> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "Partition {{ id: {}, leader_id: {}, leader_host: {} }}",
-               self.id(), self.leader_id(), self.leader_host())
+        write!(f,
+               "Partition {{ id: {}, leader: {:?} }}",
+               self.id(),
+               self.leader())
     }
 }
diff --git a/src/client/mod.rs b/src/client/mod.rs
index e62df643a..d31a745b5 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -4,7 +4,7 @@
 //! The entry point into this module is `KafkaClient` obtained by a
 //! call to `KafkaClient::new`.
 
-use std::collections::hash_map::{HashMap, Entry};
+use std::collections::hash_map::HashMap;
 use std::io::Cursor;
 use std::io::Read;
 use std::iter::Iterator;
@@ -22,6 +22,7 @@ use error::{Result, Error, KafkaCode};
 use protocol::{self, FromResponse};
 
 pub mod metadata;
+mod state;
 
 const CLIENTID: &'static str = "kafka-rust";
 const DEFAULT_SO_TIMEOUT_SECS: i32 = 120; // socket read, write timeout seconds
@@ -53,7 +54,7 @@ pub struct KafkaClient {
     conn_pool: ConnectionPool,
 
     // ~ the current state of this client
-    state: ClientState,
+    state: state::ClientState,
 }
 
 #[derive(Debug)]
@@ -69,92 +70,6 @@ struct ClientConfig {
     fetch_max_bytes_per_partition: i32,
 }
 
-#[derive(Debug)]
-struct ClientState {
-    // ~ the last correlation used when communicating with kafka
-    // (see `#next_correlation_id`)
-    correlation: i32,
-
-    // ~ a list of known brokers referred to by the index in this
-    // vector.  This index is also referred to as `BrokerIndex` in
-    // this module.
-    //
-    // note: loading of additional topic metadata must preserve
-    // already present brokers in this vector at their position.
-    brokers: Vec<Broker>,
-
-    // ~ a mapping of topic to information about its partitions
-    // ~ `TopicPartitions#partitions` are kept in ascending order by `partition_id`
-    topic_partitions: HashMap<String, TopicPartitions>,
-}
-
-#[derive(Debug)]
-struct Broker {
-    /// ~ the kafka identifier of this broker
-    node_id: i32,
-    /// ~ host:port of this broker
-    host: String,
-}
-
-/// A representation of partitions for a single topic.
-#[derive(Debug)]
-struct TopicPartitions {
-    partitions: Vec<TopicPartition>,
-}
-
-/// A custom identifier for a broker. KafkaClient maintains an array
-/// of brokers where a BrokerId is the index into this array.
-#[derive(Debug, Copy, Clone)]
-struct BrokerIndex(u32);
-
-/// Metadata for a single topic partition.
-#[derive(Debug)]
-struct TopicPartition {
-    partition_id: i32,
-    // ~ an index into ClientState#brokers
-    broker_index: BrokerIndex,
-}
-
-impl TopicPartitions {
-    fn new(partitions: Vec<TopicPartition>) -> TopicPartitions {
-        TopicPartitions { partitions: partitions }
-    }
-
-    fn partition(&self, partition_id: i32) -> Option<&TopicPartition> {
-        self.partitions.binary_search_by(|e| {
-            e.partition_id.cmp(&partition_id)
-        }).ok().and_then(|i| {
-            // `get_unchecked` is safe here due to the preceeding binary_search
-            unsafe { Some(self.partitions.get_unchecked(i)) }
-        })
-    }
-}
-
-impl TopicPartition {
-    fn new(partition_id: i32, broker_index: BrokerIndex) -> TopicPartition {
-        TopicPartition { partition_id: partition_id, broker_index: broker_index }
-    }
-}
-
-impl ClientState {
-    fn next_correlation_id(&mut self) -> i32 {
-        self.correlation = (self.correlation + 1) % (1i32 << 30);
-        self.correlation
-    }
-
-    fn broker_by_partition(&self, partition: &TopicPartition) -> &Broker {
-        &self.brokers[partition.broker_index.0 as usize]
-    }
-
-    fn find_broker<'a>(&'a self, topic: &str, partition_id: i32) -> Option<&'a str> {
-        self.topic_partitions.get(topic)
-            .and_then(|tp| tp.partition(partition_id).map(|p| {
-                let host: &str = &self.broker_by_partition(p).host;
-                host
-            }))
-    }
-}
-
 #[derive(Debug)]
 struct ConnectionPool {
     conns: HashMap<String, KafkaConnection>,
@@ -373,11 +288,7 @@ impl KafkaClient {
                 fetch_max_bytes_per_partition: DEFAULT_FETCH_MAX_BYTES_PER_PARTITION,
             },
             conn_pool: ConnectionPool::new(DEFAULT_SO_TIMEOUT_SECS),
-            state: ClientState {
-                correlation: 0,
-                brokers: Vec::new(),
-                topic_partitions: HashMap::new(),
-            },
+            state: state::ClientState::new(),
         }
     }
 
@@ -499,11 +410,17 @@ impl KafkaClient {
     ///
     /// # Examples
     /// ```no_run
-    /// let mut client = kafka::client::KafkaClient::new(vec!("localhost:9092".to_owned()));
+    /// use kafka::client::KafkaClient;
+    /// use kafka::client::metadata::Broker;
+    ///
+    /// let mut client = KafkaClient::new(vec!["localhost:9092".to_owned()]);
     /// client.load_metadata_all().unwrap();
     /// for topic in client.topics() {
     ///   for partition in topic.partitions() {
-    ///     println!("{} #{} => {}", topic.name(), partition.id(), partition.leader_host());
+    ///     println!("{} #{} => {}", topic.name(), partition.id(),
+    ///              partition.leader()
+    ///                       .map(Broker::host)
+    ///                       .unwrap_or("no-leader!"));
     ///   }
     /// }
     /// ```
@@ -559,70 +476,14 @@ impl KafkaClient {
     #[inline]
     pub fn load_metadata<T: AsRef<str>>(&mut self, topics: &[T]) -> Result<()> {
         let resp = try!(self.fetch_metadata(topics));
-        self.load_metadata_impl(resp)
-    }
-
-    // ~ this is the rest of `load_metadata`. this code is _not_
-    // generic (and hence will _not_ get generated for each distinct
-    // input type T to the `load_metadata` method).
-    fn load_metadata_impl(&mut self, md: protocol::MetadataResponse) -> Result<()> {
-        // ~ build up an index of the already loaded brokers -- if any
-        let mut brokers: HashMap<i32, BrokerIndex> =
-            HashMap::with_capacity(md.brokers.len() + self.state.brokers.len());
-        for (i, broker) in (0u32..).zip(self.state.brokers.iter()) {
-            brokers.insert(broker.node_id, BrokerIndex(i));
-        }
-        // ~ now update add new brokers or updated existing ones
-        for broker in md.brokers {
-            let broker_host = format!("{}:{}", broker.host, broker.port);
-            match brokers.entry(broker.nodeid) {
-                Entry::Occupied(e) => {
-                    // ~ verify our information of the already tracked
-                    // broker is up-to-date
-                    let broker_index = *e.get();
-                    let b = &mut self.state.brokers[broker_index.0 as usize];
-                    if b.host != broker_host {
-                        b.host = broker_host;
-                    }
-                }
-                Entry::Vacant(e) => {
-                    // ~ insert the new broker
-                    self.state.brokers.push(Broker {
-                        node_id: broker.nodeid,
-                        host: broker_host,
-                    });
-                    // ~ track the new broker's index
-                    e.insert(BrokerIndex((self.state.brokers.len() - 1) as u32));
-                }
-            }
-        }
-
-        for t in md.topics {
-            let mut known_partitions = Vec::with_capacity(t.partitions.len());
-            for partition in t.partitions {
-                match brokers.get(&partition.leader) {
-                    Some(broker_index) => {
-                        known_partitions.push(TopicPartition::new(partition.id, *broker_index));
-                    },
-                    None => {
-                        debug!("unknown leader {} for topic-partition: {}:{}",
-                               partition.leader, t.topic, partition.id);
-                    }
-                }
-            }
-            known_partitions.sort_by(|a, b| a.partition_id.cmp(&b.partition_id));
-            self.state.topic_partitions.insert(
-                t.topic, TopicPartitions::new(known_partitions));
-        }
-        Ok(())
+        self.state.update_metadata(resp)
     }
 
     /// Clears metadata stored in the client.  You must load metadata
     /// after this call if you want to use the client.
     #[inline]
     pub fn reset_metadata(&mut self) {
-        self.state.topic_partitions.clear();
-        self.state.brokers.clear();
+        self.state.clear_metadata();
     }
 
     /// Fetches metadata about the specified topics from all of the
@@ -669,12 +530,11 @@ impl KafkaClient {
         let mut reqs: HashMap<&str, protocol::OffsetRequest> = HashMap::with_capacity(n_topics);
         for topic in topics {
             let topic = topic.as_ref();
-            if let Some(tp) = state.topic_partitions.get(topic) {
-                for p in &tp.partitions {
-                    let host: &str = &state.broker_by_partition(p).host;
+            if let Some(ps) = state.partitions_for(topic) {
+                for (id, host) in ps.iter().filter_map(|(id, p)| p.broker(&state).map(|b| (id, b.host()))) {
                     let entry = reqs.entry(host)
                         .or_insert_with(|| protocol::OffsetRequest::new(correlation, &config.client_id));
-                    entry.add(topic, p.partition_id, time);
+                    entry.add(topic, id, time);
                 }
             }
         }
@@ -1013,12 +873,9 @@ impl KafkaClient {
                                -> Result<Vec<TopicPartitionOffset>>
     {
         let tps: Vec<_> =
-            match self.state.topic_partitions.get(topic) {
+            match self.state.partitions_for(topic) {
                 None => return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)),
-                Some(tp) => tp.partitions
-                    .iter()
-                    .map(|p| FetchGroupOffset::new(topic, p.partition_id))
-                    .collect(),
+                Some(tp) => tp.iter().map(|(id, _)| FetchGroupOffset::new(topic, id)).collect(),
             };
         self.fetch_group_offsets(group, tps)
     }
diff --git a/src/client/state.rs b/src/client/state.rs
new file mode 100644
index 000000000..630ca5da6
--- /dev/null
+++ b/src/client/state.rs
@@ -0,0 +1,524 @@
+use std::collections::hash_map::{HashMap, Entry, Keys};
+use std::convert::AsRef;
+use std::slice;
+use std::u32;
+
+use error::Result;
+use protocol;
+
+#[derive(Debug)]
+pub struct ClientState {
+    // ~ the last correlation used when communicating with kafka
+    // (see `#next_correlation_id`)
+    correlation: i32,
+
+    // ~ a list of known brokers referred to by the index in this
+    // vector.  This index is also referred to as `BrokerRef` and is
+    // enforced by this module.
+    //
+    // Note: loading of additional topic metadata must preserve
+    // already present brokers in this vector at their position.
+    // See `ClientState::update_metadata`
+    brokers: Vec<Broker>,
+
+    // ~ a mapping of topic to information about its partitions
+    topic_partitions: HashMap<String, TopicPartitions>,
+}
+
+// --------------------------------------------------------------------
+
+// ~ note: this type is re-exported to the crate's public api through
+// client::metadata
+/// Describes a Kafka broker node `kafka-rust` is communicating with.
+#[derive(Debug)]
+pub struct Broker {
+    /// The identifier of this broker as understood in a Kafka
+    /// cluster.
+    node_id: i32,
+    /// "host:port" of this broker. This information is advertised by
+    /// and originating from Kafka cluster itself.
+    host: String,
+}
+
+impl Broker {
+    /// Retrives the node_id of this broker as identified with the
+    /// remote Kafka cluster.
+    #[inline]
+    pub fn id(&self) -> i32 {
+        self.node_id
+    }
+
+    /// Retrieves the host:port of the this Kafka broker.
+    #[inline]
+    pub fn host(&self) -> &str {
+        &self.host
+    }
+}
+
+// See `Brokerref`
+const UNKNOWN_BROKER_INDEX: u32 = u32::MAX;
+
+/// ~ A custom identifier for a broker.  This type hides the fact that
+/// a `TopicPartition` references a `Broker` indirectly, loosely
+/// through an index, thereby being able to share broker data without
+/// having to fallback to `Rc` or `Arc` or otherwise fighting the
+/// borrowck.
+// ~ The value `UNKNOWN_BROKER_INDEX` is artificial and represents a
+// index to an unknown broker (aka the null value.) Code indexing
+// `self.brokers` using a `BrokerRef` _must_ check against this
+// constant and/or treat it conditionally.
+#[derive(Debug, Copy, Clone)]
+pub struct BrokerRef {
+    _index: u32,
+}
+
+impl BrokerRef {
+    // ~ private constructor on purpose
+    fn new(index: u32) -> Self {
+        BrokerRef { _index: index }
+    }
+
+    fn set(&mut self, other: BrokerRef) {
+        if self._index != other._index {
+            self._index = other._index;
+        }
+    }
+
+    fn set_unknown(&mut self) {
+        self.set(BrokerRef::new(UNKNOWN_BROKER_INDEX))
+    }
+}
+
+// --------------------------------------------------------------------
+
+/// A representation of partitions for a single topic.
+#[derive(Debug)]
+pub struct TopicPartitions {
+    // ~ This list keeps information about each partition of the
+    // corresponding topic - even about partitions currently without a
+    // leader.  The index into this list specifies the partition
+    // identifier.  (This works due to Kafka numbering partitions 0..N
+    // where N is the number of partitions of the topic.)
+    partitions: Vec<TopicPartition>,
+}
+
+impl TopicPartitions {
+    /// Creates a new partitions vector with all partitions leaderless
+    fn new_with_partitions(n: usize) -> TopicPartitions {
+        TopicPartitions { partitions: (0..n).map(|_| TopicPartition::new()).collect() }
+    }
+
+    pub fn len(&self) -> usize {
+        self.partitions.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.partitions.is_empty()
+    }
+
+    pub fn partition(&self, partition_id: i32) -> Option<&TopicPartition> {
+        self.partitions.get(partition_id as usize)
+    }
+
+    pub fn iter(&self) -> TopicPartitionIter {
+        self.into_iter()
+    }
+}
+
+impl<'a> IntoIterator for &'a TopicPartitions {
+    type Item = (i32, &'a TopicPartition);
+    type IntoIter = TopicPartitionIter<'a>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        TopicPartitionIter {
+            partition_id: 0,
+            iter: self.partitions.iter(),
+        }
+    }
+}
+
+/// Metadata for a single topic partition.
+#[derive(Debug)]
+pub struct TopicPartition {
+    broker: BrokerRef,
+}
+
+impl TopicPartition {
+    fn new() -> TopicPartition {
+        TopicPartition { broker: BrokerRef::new(UNKNOWN_BROKER_INDEX) }
+    }
+
+    pub fn broker<'a>(&self, state: &'a ClientState) -> Option<&'a Broker> {
+        state.brokers.get(self.broker._index as usize)
+    }
+}
+
+/// An iterator over a topic's partitions.
+pub struct TopicPartitionIter<'a> {
+    iter: slice::Iter<'a, TopicPartition>,
+    partition_id: i32,
+}
+
+impl<'a> Iterator for TopicPartitionIter<'a> {
+    type Item = (i32, &'a TopicPartition);
+    fn next(&mut self) -> Option<Self::Item> {
+        self.iter.next().map(|tp| {
+            let partition_id = self.partition_id;
+            self.partition_id += 1;
+            (partition_id, tp)
+        })
+    }
+}
+
+// --------------------------------------------------------------------
+
+// ~ note: this type is re-exported to the crate's public api through
+// client::metadata
+/// An iterator over the topic names.
+pub struct TopicNames<'a> {
+    iter: Keys<'a, String, TopicPartitions>,
+}
+
+impl<'a> Iterator for TopicNames<'a> {
+    type Item=&'a str;
+
+    #[inline]
+    fn next(&mut self) -> Option<Self::Item> {
+        self.iter.next().map(AsRef::as_ref)
+    }
+}
+
+impl ClientState {
+    pub fn new() -> Self {
+        ClientState {
+            correlation: 0,
+            brokers: Vec::new(),
+            topic_partitions: HashMap::new(),
+        }
+    }
+
+    pub fn num_topics(&self) -> usize {
+        self.topic_partitions.len()
+    }
+
+    pub fn contains_topic(&self, topic: &str) -> bool {
+        self.topic_partitions.contains_key(topic)
+    }
+
+    pub fn topic_names(&self) -> TopicNames {
+        TopicNames { iter: self.topic_partitions.keys() }
+    }
+
+    // exposed for the sake of the metadata module
+    pub fn topic_partitions(&self) -> &HashMap<String, TopicPartitions> {
+        &self.topic_partitions
+    }
+
+    pub fn partitions_for<'a>(&'a self, topic: &str) -> Option<&'a TopicPartitions> {
+        self.topic_partitions.get(topic)
+    }
+
+    pub fn next_correlation_id(&mut self) -> i32 {
+        self.correlation = (self.correlation + 1) % (1i32 << 30);
+        self.correlation
+    }
+
+    pub fn find_broker<'a>(&'a self, topic: &str, partition_id: i32) -> Option<&'a str> {
+        self.topic_partitions
+            .get(topic)
+            .and_then(|tp| tp.partition(partition_id))
+            .and_then(|p| p.broker(self))
+            .map(|b| &b.host[..])
+    }
+
+    /// Clears all metadata.
+    pub fn clear_metadata(&mut self) {
+        // ~ important to clear both since one references the other
+        // through `BrokerIndex`
+        self.topic_partitions.clear();
+        self.brokers.clear();
+    }
+
+    /// Loads new and updates existing metadata from the given
+    /// metadata response.
+    pub fn update_metadata(&mut self, md: protocol::MetadataResponse) -> Result<()> {
+        debug!("updating metadata from: {:?}", md);
+
+        // ~ register new brokers with self.brokers and obtain an
+        // index over them by broker-node-id
+        let brokers = self.update_brokers(&md);
+
+        // ~ now update partitions
+        for t in md.topics {
+            // ~ get a mutable reference to the partitions vector
+            // (maintained in self.topic_partitions) for the topic
+            let tps = match self.topic_partitions.entry(t.topic) {
+                Entry::Occupied(e) => {
+                    let ps = &mut e.into_mut().partitions;
+                    match (ps.len(), t.partitions.len()) {
+                        (n, m) if n > m => ps.truncate(m),
+                        (n, m) if n < m => {
+                            ps.reserve(m);
+                            for _ in 0..(m - n) {
+                                ps.push(TopicPartition::new());
+                            }
+                        }
+                        _ => {}
+                    }
+                    ps
+                }
+                Entry::Vacant(e) => {
+                    &mut e.insert(TopicPartitions::new_with_partitions(t.partitions.len()))
+                          .partitions
+                }
+            };
+            // ~ sync the partitions vector with the new information
+            for partition in t.partitions {
+                let tp = &mut tps[partition.id as usize];
+                if let Some(bref) = brokers.get(&partition.leader) {
+                    tp.broker.set(*bref)
+                } else {
+                    tp.broker.set_unknown()
+                }
+            }
+        }
+        Ok(())
+    }
+
+    /// Updates self.brokers from the given metadata returning an
+    /// index `NodeId -> BrokerRef`
+    fn update_brokers(&mut self, md: &protocol::MetadataResponse) -> HashMap<i32, BrokerRef> {
+        // ~ build an index of the already loaded brokers -- if any
+        let mut brokers = HashMap::with_capacity(self.brokers.len() + md.brokers.len());
+        for (i, broker) in (0u32..).zip(self.brokers.iter()) {
+            brokers.insert(broker.node_id, BrokerRef::new(i));
+        }
+
+        // ~ now add new brokers or updated existing ones while
+        // keeping the above 'broker' index up-to-date
+        for broker in &md.brokers {
+            let broker_host = format!("{}:{}", broker.host, broker.port);
+            match brokers.entry(broker.node_id) {
+                Entry::Occupied(e) => {
+                    // ~ verify our information of the already tracked
+                    // broker is up-to-date
+                    let bref = *e.get();
+                    let b = &mut self.brokers[bref._index as usize];
+                    if b.host != broker_host {
+                        b.host = broker_host;
+                    }
+                }
+                Entry::Vacant(e) => {
+                    // ~ insert the new broker
+                    let new_index = self.brokers.len();
+                    self.brokers.push(Broker {
+                        node_id: broker.node_id,
+                        host: broker_host,
+                    });
+                    // ~ track the pushed broker's index
+                    e.insert(BrokerRef::new(new_index as u32));
+                }
+            }
+        }
+        brokers
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::ClientState;
+    use protocol;
+    use protocol::metadata_int as md;
+
+    fn new_partition(id: i32, leader: i32) -> md::PartitionMetadata {
+        md::PartitionMetadata {
+            error: 0,
+            id: id,
+            leader: leader,
+            replicas: vec![],
+            isr: vec![],
+        }
+    }
+
+    /// Utility to sort the given vector and return it.
+    fn sorted<O: Ord>(mut xs: Vec<O>) -> Vec<O> {
+        xs.sort();
+        xs
+    }
+
+    // mock data for an initial kafka metadata response
+    fn metadata_response_initial() -> protocol::MetadataResponse {
+        protocol::MetadataResponse {
+            header: protocol::HeaderResponse { correlation: 1 },
+            brokers: vec![md::BrokerMetadata {
+                              node_id: 10,
+                              host: "gin1.dev".to_owned(),
+                              port: 1234,
+                          },
+                          md::BrokerMetadata {
+                              node_id: 50,
+                              host: "gin2.dev".to_owned(),
+                              port: 9876,
+                          },
+                          md::BrokerMetadata {
+                              node_id: 30,
+                              host: "gin3.dev".to_owned(),
+                              port: 9092,
+                          }],
+            topics: vec![md::TopicMetadata {
+                             error: 0,
+                             topic: "tee-one".to_owned(),
+                             partitions: vec![new_partition(0, 50),
+                                              new_partition(1, 10),
+                                              new_partition(2, 30),
+                                              new_partition(3, -1),
+                                              new_partition(4, 50)],
+                         },
+                         md::TopicMetadata {
+                             error: 0,
+                             topic: "tee-two".to_owned(),
+                             partitions: vec![new_partition(0, 30),
+                                              new_partition(1, -1),
+                                              new_partition(2, -1),
+                                              new_partition(3, 10)],
+                         },
+                         md::TopicMetadata {
+                             error: 0,
+                             topic: "tee-three".to_owned(),
+                             partitions: vec![],
+                         }],
+        }
+    }
+
+
+    fn assert_partitions(state: &ClientState,
+                         topic: &str,
+                         expected: &[(i32, Option<(i32, &str)>)]) {
+        let partitions = state.partitions_for(topic).unwrap();
+        assert_eq!(expected.len(), partitions.len());
+        assert_eq!(expected.len() == 0, partitions.is_empty());
+        assert_eq!(expected,
+                   &partitions.iter()
+                              .map(|(id, tp)| {
+                                  let broker = tp.broker(&state).map(|b| (b.id(), b.host()));
+                                  // ~ verify that find_broker delivers the same information
+                                  assert_eq!(broker.map(|b| b.1), state.find_broker(topic, id));
+                                  (id, broker)
+                              })
+                              .collect::<Vec<_>>()[..]);
+    }
+
+    fn assert_initial_metadata_load(state: &ClientState) {
+        assert_eq!(vec!["tee-one", "tee-three", "tee-two"],
+                   sorted(state.topic_names().collect::<Vec<_>>()));
+        assert_eq!(3, state.num_topics());
+
+        assert_eq!(true, state.contains_topic("tee-one"));
+        assert!(state.partitions_for("tee-one").is_some());
+
+        assert_eq!(true, state.contains_topic("tee-two"));
+        assert!(state.partitions_for("tee-two").is_some());
+
+        assert_eq!(true, state.contains_topic("tee-three"));
+        assert!(state.partitions_for("tee-three").is_some());
+
+        assert_eq!(false, state.contains_topic("foobar"));
+        assert!(state.partitions_for("foobar").is_none());
+
+        assert_partitions(&state,
+                          "tee-one",
+                          &[(0, Some((50, "gin2.dev:9876"))),
+                            (1, Some((10, "gin1.dev:1234"))),
+                            (2, Some((30, "gin3.dev:9092"))),
+                            (3, None),
+                            (4, Some((50, "gin2.dev:9876")))]);
+        assert_partitions(&state,
+                          "tee-two",
+                          &[(0, Some((30, "gin3.dev:9092"))),
+                            (1, None),
+                            (2, None),
+                            (3, Some((10, "gin1.dev:1234")))]);
+        assert_partitions(&state, "tee-three", &[]);
+    }
+
+    fn metadata_response_update() -> protocol::MetadataResponse {
+        protocol::MetadataResponse {
+            header: protocol::HeaderResponse { correlation: 2 },
+            brokers: vec![md::BrokerMetadata {
+                              node_id: 10,
+                              host: "gin1.dev".to_owned(),
+                              port: 1234,
+                          },
+                          // note: compared to the initial metadata
+                          // response this broker moved to a different
+                          // machine
+                          md::BrokerMetadata {
+                              node_id: 50,
+                              host: "aladin1.dev".to_owned(),
+                              port: 9091,
+                          },
+                          md::BrokerMetadata {
+                              node_id: 30,
+                              host: "gin3.dev".to_owned(),
+                              port: 9092,
+                          }],
+            // metadata for topic "tee-two" only
+            topics: vec![md::TopicMetadata {
+                             error: 0,
+                             topic: "tee-two".to_owned(),
+                             partitions: vec![new_partition(0, 10),
+                                              new_partition(1, 10),
+                                              new_partition(2, 50),
+                                              new_partition(3, -1),
+                                              new_partition(4, 30)],
+                         }],
+        }
+    }
+
+    fn assert_updated_metadata_load(state: &ClientState) {
+        assert_eq!(vec!["tee-one", "tee-three", "tee-two"],
+                   sorted(state.topic_names().collect::<Vec<_>>()));
+        assert_eq!(3, state.num_topics());
+
+        assert_eq!(true, state.contains_topic("tee-one"));
+        assert!(state.partitions_for("tee-one").is_some());
+
+        assert_eq!(true, state.contains_topic("tee-two"));
+        assert!(state.partitions_for("tee-two").is_some());
+
+        assert_eq!(true, state.contains_topic("tee-three"));
+        assert!(state.partitions_for("tee-three").is_some());
+
+        assert_eq!(false, state.contains_topic("foobar"));
+        assert!(state.partitions_for("foobar").is_none());
+
+        assert_partitions(&state,
+                          "tee-one",
+                          &[(0, Some((50, "aladin1.dev:9091"))),
+                            (1, Some((10, "gin1.dev:1234"))),
+                            (2, Some((30, "gin3.dev:9092"))),
+                            (3, None),
+                            (4, Some((50, "aladin1.dev:9091")))]);
+        assert_partitions(&state,
+                          "tee-two",
+                          &[(0, Some((10, "gin1.dev:1234"))),
+                            (1, Some((10, "gin1.dev:1234"))),
+                            (2, Some((50, "aladin1.dev:9091"))),
+                            (3, None),
+                            (4, Some((30, "gin3.dev:9092")))]);
+        assert_partitions(&state, "tee-three", &[]);
+    }
+
+    #[test]
+    fn test_loading_metadata() {
+        let mut state = ClientState::new();
+        // Test loading metadata into a new, empty client state.
+        state.update_metadata(metadata_response_initial()).unwrap();
+        assert_initial_metadata_load(&state);
+
+        // Test loading a metadata update into a client state with
+        // already some initial metadata loaded.
+        state.update_metadata(metadata_response_update()).unwrap();
+        assert_updated_metadata_load(&state);
+    }
+}
diff --git a/src/producer.rs b/src/producer.rs
index 45d62c702..e428b868a 100644
--- a/src/producer.rs
+++ b/src/producer.rs
@@ -1,11 +1,15 @@
-//! Kafka Producer - A higher-level API for sending messages to kafka topics.
+//! Kafka Producer - A higher-level API for sending messages to Kafka
+//! topics.
 //!
-//! A multi-topic capable producer for a Kafka cluster providing a
-//! convenient API for sending messages.  So far the producer has only
-//! synchronous capabilities.
+//! This module hosts a multi-topic capable producer for a Kafka
+//! cluster providing a convenient API for sending messages
+//! synchronously.
 //!
-//! In Kafka, each message is basically a key/value pair.  A `Record`
-//! is all the data necessary to produce such a message to Kafka.
+//! In Kafka, each message is a key/value pair where one or the other
+//! is optional.  A `Record` represents all the data necessary to
+//! produce such a message to Kafka using the `Producer`.  It
+//! specifies the target topic and the target partition the message is
+//! supposed to be delivered to as well as the key and the value.
 //!
 //! # Example
 //! ```no_run
@@ -27,23 +31,40 @@
 //! }
 //! ```
 //!
-//! In this example, when the method call to `producer.send` returns
+//! In this example, when the `producer.send(..)` returns
 //! successfully, we are guaranteed the message is delivered to Kafka
 //! and persisted by at least one Kafka broker.  However, when sending
-//! multiple messages just like this example, it is more efficient to
-//! send them in batches using `Producer::send_all`.
+//! multiple messages just like in this example, it is more efficient
+//! to send them in batches using `Producer::send_all`.
+//!
+//! Since some of the `Record`s attributes are optional, convenience
+//! methods exist to ease their creation.  In this example, the call
+//! to `Record::from_value` creates a key-less, value-only record with
+//! an unspecified partition.  The `Record` struct, however, is
+//! intended to provide full control over its lifecycle to client
+//! code, and, hence, is fully open.  Its current constructor methods
+//! are provided for convience only.
+//!
+//! Beside the target topic, key, and the value of a `Record`, client
+//! code is allowed to specify the topic partition the message is
+//! supposed to be delivered to.  If the partition of a `Record` is
+//! not specified - more precisely speaking if it's negative -
+//! `Producer` will rely on its underlying `Partitioner` to find a
+//! suitable one.  A `Partitioner` implementation can be supplied by
+//! client code at the `Producer`'s construction time and defaults to
+//! `DefaultPartitioner`.  See that for more information for its
+//! strategy to find a partition.
 
 // XXX 1) rethink return values for the send_all() method
-// XXX 2) extend DefaultPartitioner to dispatch based on message key
-// XXX 3) Handle recoverable errors behind the scenes through retry attempts
-// XXX 4) maintain a background thread to provide an async version of the send* methods
+// XXX 2) Handle recoverable errors behind the scenes through retry attempts
 
 use std::collections::HashMap;
 use std::fmt;
+use std::hash::{Hasher, SipHasher, BuildHasher, BuildHasherDefault};
 
 use client::{self, KafkaClient};
 // public re-exports
-pub use client::{Compression};
+pub use client::Compression;
 use error::Result;
 use utils::TopicPartitionOffset;
 
@@ -65,7 +86,9 @@ pub trait AsBytes {
 }
 
 impl AsBytes for () {
-    fn as_bytes(&self) -> &[u8] { &[] }
+    fn as_bytes(&self) -> &[u8] {
+        &[]
+    }
 }
 
 // There seems to be some compiler issue with this:
@@ -75,19 +98,27 @@ impl AsBytes for () {
 
 // for now we provide the impl for some standard library types
 impl AsBytes for String {
-    fn as_bytes(&self) -> &[u8] { self.as_ref() }
+    fn as_bytes(&self) -> &[u8] {
+        self.as_ref()
+    }
 }
 impl AsBytes for Vec<u8> {
-    fn as_bytes(&self) -> &[u8] { self.as_ref() }
+    fn as_bytes(&self) -> &[u8] {
+        self.as_ref()
+    }
 }
 
 impl<'a> AsBytes for &'a [u8] {
-    fn as_bytes(&self) -> &[u8] { self }
+    fn as_bytes(&self) -> &[u8] {
+        self
+    }
 }
 impl<'a> AsBytes for &'a str {
-    fn as_bytes(&self) -> &[u8] { str::as_bytes(self) }
+    fn as_bytes(&self) -> &[u8] {
+        str::as_bytes(self)
+    }
 }
- 
+
 // --------------------------------------------------------------------
 
 /// A structure representing a message to be sent to Kafka through the
@@ -111,13 +142,17 @@ pub struct Record<'a, K, V> {
 }
 
 impl<'a, K, V> Record<'a, K, V> {
-
     /// Convenience function to create a new key/value record with an
     /// "unspecified" partition - this is, a partition set to a negative
     /// value.
     #[inline]
     pub fn from_key_value(topic: &'a str, key: K, value: V) -> Record<'a, K, V> {
-        Record { key: key, value: value, topic: topic, partition: -1 }
+        Record {
+            key: key,
+            value: value,
+            topic: topic,
+            partition: -1,
+        }
     }
 
     /// Convenience method to set the partition.
@@ -134,14 +169,23 @@ impl<'a, V> Record<'a, (), V> {
     /// value.
     #[inline]
     pub fn from_value(topic: &'a str, value: V) -> Record<'a, (), V> {
-        Record { key: (), value: value, topic: topic, partition: -1 }
+        Record {
+            key: (),
+            value: value,
+            topic: topic,
+            partition: -1,
+        }
     }
 }
 
 impl<'a, K: fmt::Debug, V: fmt::Debug> fmt::Debug for Record<'a, K, V> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "Record {{ topic: {}, partition: {}, key: {:?}, value: {:?} }}",
-               self.topic, self.partition, self.key, self.value)
+        write!(f,
+               "Record {{ topic: {}, partition: {}, key: {:?}, value: {:?} }}",
+               self.topic,
+               self.partition,
+               self.key,
+               self.value)
     }
 }
 
@@ -158,7 +202,7 @@ pub struct Producer<P = DefaultPartitioner> {
 
 struct State<P> {
     /// A list of available partition IDs for each topic.
-    partition_ids: HashMap<String, Vec<i32>>,
+    partitions: HashMap<String, Partitions>,
     /// The partitioner to decide how to distribute messages
     partitioner: P,
 }
@@ -192,11 +236,10 @@ impl Producer {
 
 
 impl<P: Partitioner> Producer<P> {
-
     /// Synchronously send the specified message to Kafka.
-    pub fn send<'a, K, V>(&mut self, rec: &Record<'a, K, V>)
-                          -> Result<()>
-        where K: AsBytes, V: AsBytes
+    pub fn send<'a, K, V>(&mut self, rec: &Record<'a, K, V>) -> Result<()>
+        where K: AsBytes,
+              V: AsBytes
     {
         let mut rs = try!(self.send_all(ref_slice(rec)));
         assert_eq!(1, rs.len());
@@ -204,27 +247,29 @@ impl<P: Partitioner> Producer<P> {
     }
 
     /// Synchronously send all of the specified messages to Kafka.
-    pub fn send_all<'a, K, V>(&mut self, recs: &[Record<'a, K, V>])
+    pub fn send_all<'a, K, V>(&mut self,
+                              recs: &[Record<'a, K, V>])
                               -> Result<Vec<TopicPartitionOffset>>
-        where K: AsBytes, V: AsBytes
+        where K: AsBytes,
+              V: AsBytes
     {
         let partitioner = &mut self.state.partitioner;
-        let partition_ids = &self.state.partition_ids;
+        let partitions = &self.state.partitions;
         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(partition_ids), &mut m);
-                m
-            }))
+        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
+                                }))
     }
 }
 
@@ -243,9 +288,17 @@ impl<P> State<P> {
         let ts = client.topics();
         let mut ids = HashMap::with_capacity(ts.len());
         for t in ts {
-            ids.insert(t.name().to_owned(), t.partition_ids());
+            let ps = t.partitions();
+            ids.insert(t.name().to_owned(),
+                       Partitions {
+                           available_ids: ps.available_ids(),
+                           num_all_partitions: ps.len() as u32,
+                       });
         }
-        Ok(State{partition_ids: ids, partitioner: partitioner})
+        Ok(State {
+            partitions: ids,
+            partitioner: partitioner,
+        })
     }
 }
 
@@ -263,9 +316,7 @@ pub struct Builder<P = DefaultPartitioner> {
 }
 
 impl Builder {
-    fn new(client: Option<KafkaClient>, hosts: Vec<String>)
-           -> Builder<DefaultPartitioner>
-    {
+    fn new(client: Option<KafkaClient>, hosts: Vec<String>) -> Builder<DefaultPartitioner> {
         let mut b = Builder {
             client: client,
             hosts: hosts,
@@ -314,7 +365,6 @@ impl Builder {
 }
 
 impl<P> Builder<P> {
-
     /// Sets the partitioner to dispatch when sending messages without
     /// an explicit partition assignment.
     pub fn with_partitioner<Q: Partitioner>(self, partitioner: Q) -> Builder<Q> {
@@ -345,7 +395,7 @@ impl<P> Builder<P> {
             ack_timeout: self.ack_timeout,
             required_acks: self.required_acks,
         };
-        Ok(Producer{
+        Ok(Producer {
             client: client,
             state: state,
             config: config,
@@ -359,18 +409,51 @@ impl<P> Builder<P> {
 ///
 /// Indented for use by `Partitioner`s.
 pub struct Topics<'a> {
-    partition_ids: &'a HashMap<String, Vec<i32>>,
+    partitions: &'a HashMap<String, Partitions>,
+}
+
+/// Producer relevant partition information of a particular topic.
+///
+/// Indented for use by `Partition`s.
+pub struct Partitions {
+    available_ids: Vec<i32>,
+    num_all_partitions: u32,
+}
+
+impl Partitions {
+    /// Retrieves the list of the identifiers of currently "available"
+    /// partitions for the given topic.  This list excludes partitions
+    /// which do not have a leader broker assigned.
+    #[inline]
+    pub fn available_ids(&self) -> &[i32] {
+        &self.available_ids
+    }
+
+    /// Retrieves the number of "available" partitions. This is a
+    /// merely a convenience method. See `Partitions::available_ids`.
+    #[inline]
+    pub fn num_available(&self) -> u32 {
+        self.available_ids.len() as u32
+    }
+
+    /// The total number of partitions of the underlygin topic.  This
+    /// number includes also partitions without a current leader
+    /// assignment.
+    #[inline]
+    pub fn num_all(&self) -> u32 {
+        self.num_all_partitions
+    }
 }
 
 impl<'a> Topics<'a> {
-    fn new(partition_ids: &'a HashMap<String, Vec<i32>>) -> Topics<'a> {
-        Topics { partition_ids: partition_ids }
+    fn new(partitions: &'a HashMap<String, Partitions>) -> Topics<'a> {
+        Topics { partitions: partitions }
     }
 
-    /// Retrieves a list of the identifiers of available partitions
-    /// for the given topic.
-    pub fn partition_ids(&self, topic: &'a str) -> Option<&[i32]> {
-        self.partition_ids.get(topic).map(|ps| &ps[..])
+    /// Retrieves informationa about a topic's partitions.
+    #[inline]
+    pub fn partitions(&'a self, topic: &str) -> Option<&'a Partitions> {
+        self.partitions.get(topic)
     }
 }
 
@@ -394,58 +477,214 @@ pub trait Partitioner {
 /// As its name implies `DefaultPartitioner` is the default
 /// partitioner for `Producer`.
 ///
-/// Every received message with a non-negative value will not be
-/// changed by this partitioner and will be passed through as is.
+/// For every message it proceedes as follows:
+///
+/// - If the messages contains a non-negative partition value it
+/// leaves the message untouched.  This will cause `Producer` to try
+/// to send the message to exactly that partition to.
 ///
-/// For every message with an "unspecified" `partition` - this is a
-/// negative partition - it will try to find one.  In a very simple
-/// manner, it tries to distribute such messsages across available
-/// partitions in a round robin fashion.
+/// - Otherwise, if the message has an "unspecified" `partition` -
+/// this is, it has a negative partition value - and a specified key,
+/// `DefaultPartitioner` will compute a hash from the key using the
+/// underlying hasher and take `hash % num_all_partitions` to derive
+/// the partition to send the message to.  This will consistently
+/// cause messages with the same key to be sent to the same partition.
 ///
-/// This behavior may not suffice every workload.  If you're
-/// application is dependent on a particular distribution scheme, you
-/// want to provide your own partioner to the `Producer` at
-/// initialization time.
+/// - Otherwise - a message with an "unspecified" `partition` and no
+/// key - `DefaultPartitioner` will "randomly" pick one from the
+/// "available" partitions trying to distribute the messages across
+/// the multiple partitions.  In particular, it tries to distribute
+/// such messsages across the "available" partitions in a round robin
+/// fashion.  "Available" it this context means partitions with a
+/// known leader.
+///
+/// This behavior may not suffice every workload.  If your application
+/// is dependent on a particular distribution scheme different from
+/// the one outlined above, you want to provide your own partioner to
+/// the `Producer` at its initialization time.
 ///
 /// See `Builder::with_partitioner`.
-pub struct DefaultPartitioner {
+#[derive(Default)]
+pub struct DefaultPartitioner<H = BuildHasherDefault<SipHasher>> {
+    // ~ a hasher builder; used to consistently hash keys
+    hash_builder: H,
     // ~ a counter incremented with each partitioned message to
     // achieve a different partition assignment for each message
-    cntr: u32
+    cntr: u32,
+}
+
+impl DefaultPartitioner {
+    /// Creates a new partitioner which will use the given hash
+    /// builder to hash message keys.
+    pub fn with_hasher<B: BuildHasher>(hash_builder: B) -> DefaultPartitioner<B> {
+        DefaultPartitioner {
+            hash_builder: hash_builder.into(),
+            cntr: 0,
+        }
+    }
+
+    pub fn with_default_hasher<B>() -> DefaultPartitioner<BuildHasherDefault<B>>
+        where B: Hasher + Default
+    {
+        DefaultPartitioner {
+            hash_builder: BuildHasherDefault::<B>::default(),
+            cntr: 0,
+        }
+    }
 }
 
-impl Partitioner for DefaultPartitioner {
+impl<H: BuildHasher> Partitioner for DefaultPartitioner<H> {
     #[allow(unused_variables)]
     fn partition(&mut self, topics: Topics, rec: &mut client::ProduceMessage) {
-        if rec.partition < 0 {
-            match topics.partition_ids(rec.topic) {
-                Some(ps) if !ps.is_empty() => {
-
-                    // ~ XXX dispatch to partition by the hash of the
-                    // key - if that is available, otherwise continue
-                    // just like below
-                    //
-                    // ~ XXX dispatching by key will require that we
-                    // take '% "all-partitions"' (not just the
-                    // "currently available" ones) to guard agaist
-                    // sending the same key to a different partition
-                    // due to a temporal unavailability of a
-                    // particular partition.
-
-                    // ~ get a partition
-                    rec.partition = ps[self.cntr as usize % ps.len()];
+        if rec.partition >= 0 {
+            // ~ partition explicitely defined, trust the user
+            return;
+        }
+        let partitions = match topics.partitions(rec.topic) {
+            None => return, // ~ unknown topic, this is not the place to deal with it.
+            Some(partitions) => partitions,
+        };
+        match rec.key {
+            Some(key) => {
+                let num_partitions = partitions.num_all();
+                if num_partitions == 0 {
+                    // ~ no partitions at all ... a rather strange
+                    // topic. again, this is not the right place to
+                    // deal with it.
+                    return;
+                }
+                let mut h = self.hash_builder.build_hasher();
+                h.write(key);
+                // ~ unconditionally dispatch to partitions no matter
+                // whether they are currently available or not.  this
+                // guarantees consistency which is the point of
+                // partitioning by key.  other behaviour - if desired
+                // - can be implemented in custom, user provided
+                // partitioners.
+                let hash = h.finish() as u32;
+                // if `num_partitions == u32::MAX` this can lead to a
+                // negative partition ... such a partition count is very
+                // unlikely though
+                rec.partition = (hash % num_partitions) as i32;
+            }
+            None => {
+                // ~ no key available, determine a partition from the
+                // available ones.
+                let avail = partitions.available_ids();
+                if avail.len() > 0 {
+                    rec.partition = avail[self.cntr as usize % avail.len()];
                     // ~ update internal state so that the next time we choose
                     // a different partition
                     self.cntr = self.cntr.wrapping_add(1);
                 }
-                _ => {}
             }
         }
     }
 }
 
-impl Default for DefaultPartitioner {
-    fn default() -> Self {
-        DefaultPartitioner { cntr: 0 }
+// --------------------------------------------------------------------
+
+#[cfg(test)]
+mod default_partitioner_tests {
+    use std::hash::{Hasher, SipHasher, BuildHasherDefault};
+    use std::collections::HashMap;
+
+    use client;
+    use super::{DefaultPartitioner, Partitioner, Partitions, Topics};
+
+    fn topics_map(topics: Vec<(&str, Partitions)>) -> HashMap<String, Partitions> {
+        let mut h = HashMap::new();
+        for topic in topics {
+            h.insert(topic.0.into(), topic.1);
+        }
+        h
+    }
+
+    fn assert_partitioning<P: Partitioner>(topics: &HashMap<String, Partitions>,
+                                           p: &mut P,
+                                           topic: &str,
+                                           key: &str)
+                                           -> i32 {
+        let mut msg = client::ProduceMessage {
+            key: Some(key.as_bytes()),
+            value: None,
+            topic: topic,
+            partition: -1,
+        };
+        p.partition(Topics::new(topics), &mut msg);
+        let num_partitions = topics.get(topic).unwrap().num_all_partitions as i32;
+        assert!(msg.partition >= 0 && msg.partition < num_partitions);
+        msg.partition
+    }
+
+    /// Validate consistent partitioning on a message's key
+    #[test]
+    fn test_key_partitioning() {
+        let h = topics_map(vec![("foo",
+                                 Partitions {
+                                    available_ids: vec![0, 1, 4],
+                                    num_all_partitions: 5,
+                                }),
+                                ("bar",
+                                 Partitions {
+                                    available_ids: vec![0, 1],
+                                    num_all_partitions: 2,
+                                })]);
+
+        let mut p: DefaultPartitioner<BuildHasherDefault<SipHasher>> = Default::default();
+
+        // ~ validate that partitioning by the same key leads to the same
+        // partition
+        let h1 = assert_partitioning(&h, &mut p, "foo", "foo-key");
+        let h2 = assert_partitioning(&h, &mut p, "foo", "foo-key");
+        assert_eq!(h1, h2);
+
+        // ~ validate that partitioning by different keys leads to
+        // different partitions (the keys are chosen such that they lead
+        // to different partitions)
+        let h3 = assert_partitioning(&h, &mut p, "foo", "foo-key");
+        let h4 = assert_partitioning(&h, &mut p, "foo", "bar-key");
+        assert!(h3 != h4);
+    }
+
+    #[derive(Default)]
+    struct MyCustomHasher(u64);
+
+    impl Hasher for MyCustomHasher {
+        fn finish(&self) -> u64 {
+            self.0
+        }
+        fn write(&mut self, bytes: &[u8]) {
+            self.0 = bytes[0] as u64;
+        }
+    }
+
+    /// Validate it is possible to register a custom hasher with the
+    /// default partitioner
+    #[test]
+    fn default_partitioner_with_custom_hasher_default() {
+        // this must compile
+        let mut p = DefaultPartitioner::with_default_hasher::<MyCustomHasher>();
+
+        let h = topics_map(vec![("confirms",
+                                 Partitions {
+                                    available_ids: vec![0, 1],
+                                    num_all_partitions: 2,
+                                }),
+                                ("contents",
+                                 Partitions {
+                                    available_ids: vec![0, 1, 9],
+                                    num_all_partitions: 10,
+                                })]);
+
+        // verify also the partitioner derives the correct partition
+        // ... this is hash modulo num_all_partitions. here it is a
+        // topic with a total of 2 partitions.
+        let p1 = assert_partitioning(&h, &mut p, "confirms", "A" /* ascii: 65 */);
+        assert_eq!(1, p1);
+
+        // here it is a topic with a total of 10 partitions
+        let p2 = assert_partitioning(&h, &mut p, "contents", "B" /* ascii: 66 */);
+        assert_eq!(6, p2);
     }
 }
diff --git a/src/protocol/consumer.rs b/src/protocol/consumer_int.rs
similarity index 100%
rename from src/protocol/consumer.rs
rename to src/protocol/consumer_int.rs
diff --git a/src/protocol/fetch_req.rs b/src/protocol/fetch_int.rs
similarity index 94%
rename from src/protocol/fetch_req.rs
rename to src/protocol/fetch_int.rs
index fe569e0cf..cdb3ac027 100644
--- a/src/protocol/fetch_req.rs
+++ b/src/protocol/fetch_int.rs
@@ -1,6 +1,7 @@
-//! A representation of the requests for fetching messages. This code
+//! A representation of the requests for fetching messages.  This code
 //! is kept apart from the corresponding "responses" module, since the
-//! latter is publicly exported by this crate.
+//! latter is publicly exported by this crate but we don't want to
+//! expose the "request" structures.
 
 use error::Result;
 use std::io::Write;
diff --git a/src/protocol/metadata.rs b/src/protocol/metadata_int.rs
similarity index 97%
rename from src/protocol/metadata.rs
rename to src/protocol/metadata_int.rs
index 8a7d91943..180f80831 100644
--- a/src/protocol/metadata.rs
+++ b/src/protocol/metadata_int.rs
@@ -43,7 +43,7 @@ pub struct MetadataResponse {
 
 #[derive(Default, Debug)]
 pub struct BrokerMetadata {
-    pub nodeid: i32,
+    pub node_id: i32,
     pub host: String,
     pub port: i32
 }
@@ -83,7 +83,7 @@ impl FromByte for BrokerMetadata {
     #[allow(unused_must_use)]
     fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
         try_multi!(
-            self.nodeid.decode(buffer),
+            self.node_id.decode(buffer),
             self.host.decode(buffer),
             self.port.decode(buffer)
         )
diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs
index af2b055fb..1e38f7589 100644
--- a/src/protocol/mod.rs
+++ b/src/protocol/mod.rs
@@ -12,24 +12,26 @@ macro_rules! try_multi {
     })
 }
 
-// ~ these modules will see the above defined macro
-mod produce;
-mod offset;
-mod metadata;
-mod consumer;
+// ~ exposed merely for unit testing purposes with this crate.
+// however, there are _not_ intended to be a public artifact of this
+// crate and, therefore, are marked as "int"ernal.
+pub mod produce_int;
+pub mod offset_int;
+pub mod metadata_int;
+pub mod consumer_int;
+pub mod fetch_int;
 
 mod zreader;
-mod fetch_req;
 pub mod fetch;
 
-// ~ re-exports for request/response types defined in submodules
-pub use self::fetch_req::FetchRequest;
-pub use self::produce::{ProduceRequest, ProduceResponse};
-pub use self::offset::{OffsetRequest, OffsetResponse};
-pub use self::metadata::{MetadataRequest, MetadataResponse};
-// pub use self::consumer::{ConsumerMetadataRequest, ConsumerMetadataResponse};
-pub use self::consumer::{OffsetFetchRequest, OffsetFetchResponse,
-                         OffsetCommitRequest, OffsetCommitResponse};
+// ~ convenient re-exports for request/response types defined in the
+// submodules
+pub use self::fetch_int::FetchRequest;
+pub use self::produce_int::{ProduceRequest, ProduceResponse};
+pub use self::offset_int::{OffsetRequest, OffsetResponse};
+pub use self::metadata_int::{MetadataRequest, MetadataResponse};
+pub use self::consumer_int::{OffsetFetchRequest, OffsetFetchResponse,
+                             OffsetCommitRequest, OffsetCommitResponse};
 
 // --------------------------------------------------------------------
 
diff --git a/src/protocol/offset.rs b/src/protocol/offset_int.rs
similarity index 100%
rename from src/protocol/offset.rs
rename to src/protocol/offset_int.rs
diff --git a/src/protocol/produce.rs b/src/protocol/produce_int.rs
similarity index 100%
rename from src/protocol/produce.rs
rename to src/protocol/produce_int.rs


Test patch: