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

Issue:
Consumer performance
The `Consumer` currently clones every messages it finally delivers to its client code (and further continues to reference the originally cloned message.) Given that messages can be pretty large this needs to be changed. (I fear we might need to drop the `Iterator` implementation and provide some other - potentially more complicated - means of iterating the fetched messages.)



Discussion / hints:
A quick and dirty proof-of-concept I've prepared locally (not yet pushed) indicates we can get much faster at parsing uncompressed "fetch responses" and providing the "message payload" to client code. The consumer is very likely to benefit as well. I'll push as soon as I clean up my code, though, I intend this to end up only in `branch-0.2.x`.


Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 71c31a7b9..b2e20a694 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,12 +17,14 @@ byteorder = "0.4"
 libc = "0.2"
 flate2 = "0.2"
 crc = "1.2"
+log = "0.3"
 
 [build-dependencies]
 pkg-config = "0.3"
 
 [dev-dependencies]
 getopts = "0.2"
+env_logger = "0.3"
 
 [features]
 nightly = []
diff --git a/examples/console-consumer.rs b/examples/console-consumer.rs
new file mode 100644
index 000000000..e3b7a3db5
--- /dev/null
+++ b/examples/console-consumer.rs
@@ -0,0 +1,113 @@
+extern crate kafka;
+extern crate getopts;
+extern crate env_logger;
+
+use std::{env, io, fmt, process};
+
+use kafka::client::{KafkaClient, FetchOffset};
+use kafka::consumer::Consumer;
+
+/// This is a very simple command line application reading from a
+/// specific kafka topic and dumping the messages to standard output.
+fn main() {
+    env_logger::init().unwrap();
+
+    let cfg = match Config::from_cmdline() {
+        Ok(cfg) => cfg,
+        Err(e) => {
+            println!("{}", e);
+            process::exit(1);
+        }
+    };
+    if let Err(e) = process(&cfg) {
+        println!("{}", e);
+        process::exit(1);
+    }
+}
+
+fn process(cfg: &Config) -> Result<(), Error> {
+    // XXX too verbose to set up the consumer ... try avoiding the
+    // indirection through KafkaClient
+    let mut client = KafkaClient::new(cfg.brokers.clone());
+    client.set_fetch_max_bytes_per_partition(1024*1024);
+    try!(client.load_metadata_all());
+
+    let mut c = Consumer::new(client, cfg.group.clone(), cfg.topic.clone())
+        .with_fallback_offset(FetchOffset::Earliest);
+
+    loop {
+        for ms in try!(c.poll()).iter() {
+            for m in ms.messages() {
+                let s = String::from_utf8_lossy(m.value);
+                println!("{}:{}@{}: {}", ms.topic(), ms.partition(), m.offset, s.trim());
+            }
+            c.consume_messageset(ms);
+        }
+        try!(c.commit_consumed());
+    }
+}
+
+// --------------------------------------------------------------------
+
+enum Error {
+    Kafka(kafka::error::Error),
+    Io(io::Error),
+    Literal(String),
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            &Error::Kafka(ref e) => write!(f, "kafka-error: {}", e),
+            &Error::Io(ref e) => write!(f, "io-error: {}", e),
+            &Error::Literal(ref s) => write!(f, "{}", s),
+        }
+    }
+}
+
+impl From<kafka::error::Error> for Error {
+    fn from(e: kafka::error::Error) -> Self { Error::Kafka(e) }
+}
+
+impl From<io::Error> for Error {
+    fn from(e: io::Error) -> Self { Error::Io(e) }
+}
+
+// --------------------------------------------------------------------
+
+struct Config {
+    brokers: Vec<String>,
+    group: String,
+    topic: String,
+}
+
+impl Config {
+    fn from_cmdline() -> Result<Config, Error> {
+        let args: Vec<_> = env::args().collect();
+        let mut opts = getopts::Options::new();
+        opts.optflag("h", "help", "Print this help screen");
+        opts.optopt("", "brokers", "Specify kafka brokers (comma separated)", "HOSTS");
+        opts.optopt("", "topic", "Specify target topic", "NAME");
+        opts.optopt("", "group", "Specify the group_id file", "NAME");
+
+        let m = match opts.parse(&args[1..]) {
+            Ok(m) => m,
+            Err(e) => return Err(Error::Literal(e.to_string())),
+        };
+        if m.opt_present("help") {
+            let brief = format!("{} [options]", args[0]);
+            return Err(Error::Literal(opts.usage(&brief)));
+        }
+        Ok(Config {
+            brokers: m.opt_str("brokers")
+                .unwrap_or_else(|| "localhost:9092".to_owned())
+                .split(',')
+                .map(|s| s.trim().to_owned())
+                .collect(),
+            group: m.opt_str("group")
+                .unwrap_or_else(|| "my-group".to_owned()),
+            topic: m.opt_str("topic")
+                .unwrap_or_else(|| "my-topic".to_owned()),
+        })
+   }
+}
diff --git a/examples/consume.rs b/examples/consume.rs
index 8c84692b9..3e0092681 100644
--- a/examples/consume.rs
+++ b/examples/consume.rs
@@ -1,11 +1,13 @@
 extern crate kafka;
 
 use kafka::client::{KafkaClient, FetchOffset};
+use kafka::consumer::Consumer;
 
-/// This program demonstrates consuming messages through `KafkaClient`. This is the top level
-/// client that will fit most use cases. Note that consumed messages are tracked by Kafka so you
-/// can only consume them once. This is what you want for most use cases, you can look at
-/// examples/fetch.rs for a lower level API.
+/// This program demonstrates consuming messages through a `Consumer`.
+/// This is a convenient client that will fit most use cases.  Note
+/// that consumed messages are tracked by Kafka so you can only
+/// consume them once.  This is what you want for most use cases, you
+/// can look at `examples/fetch.rs` for a lower level API.
 fn main() {
     let broker = "localhost:9092";
     let topic = "my-topic";
@@ -26,11 +28,25 @@ fn main() {
         return;
     }
 
-    let con = kafka::consumer::Consumer::new(client, "test-group".to_owned(), topic.to_owned())
-        .fallback_offset(FetchOffset::Earliest);
-
-    for msg in con {
-        println!("{:?}", msg);
+    let mut con = Consumer::new(client, "test-group".to_owned(), topic.to_owned())
+        .with_fallback_offset(FetchOffset::Earliest);
+    loop {
+        match con.poll() {
+            Err(e) => {
+                println!("Error consuming messages: {}", e);
+                break;
+            }
+            Ok(mss) => {
+                if mss.is_empty() {
+                    println!("No more messages.");
+                    break;
+                }
+                for ms in mss.iter() {
+                    for m in ms.messages() {
+                        println!("{}:{}@{}: {:?}", ms.topic(), ms.partition(), m.offset, m.value);
+                    }
+                }
+            }
+        }
     }
-    println!("No more messages.")
 }
diff --git a/examples/fetch.rs b/examples/fetch.rs
index c98865c89..04d0005af 100644
--- a/examples/fetch.rs
+++ b/examples/fetch.rs
@@ -1,6 +1,7 @@
 extern crate kafka;
 
 use kafka::client::KafkaClient;
+use kafka::utils::TopicPartitionOffset;
 
 /// This program demonstrates the low level api for fetching messages.
 /// Please look at examles/consume.rs for an easier to use API.
@@ -27,15 +28,30 @@ fn main() {
         return;
     }
 
-    match client.fetch_messages(topic, partition, offset) {
+    match client.fetch_messages(&[TopicPartitionOffset::new(topic, partition, offset)]) {
         Err(e) => {
             println!("Failed to fetch messages: {}", e);
         }
-        Ok(msgs) => {
-            for msg in msgs {
-                println!("{:?}", msg);
+        Ok(resps) => {
+            for resp in resps {
+                for t in resp.topics() {
+                    for p in t.partitions() {
+                        match p.data() {
+                            &Err(ref e) => {
+                                println!("partition error: {}:{}: {}", t.topic(), p.partition(), e)
+                            }
+                            &Ok(ref data) => {
+                                println!("topic: {} / partition: {} / latest available message offset: {}",
+                                         t.topic(), p.partition(), data.highwatermark_offset());
+                                for msg in data.messages() {
+                                    println!("topic: {} / partition: {} / message.offset: {} / message.len: {}",
+                                             t.topic(), p.partition(), msg.offset, msg.value.len());
+                                }
+                            }
+                        }
+                    }
+                }
             }
-            println!("No more messages.");
         }
     }
 }
diff --git a/src/client.rs b/src/client.rs
index 5d999aeb2..70bcc9fe4 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -13,13 +13,12 @@ use std::cell::Cell;
 
 // pub re-export
 pub use compression::Compression;
+pub use protocol::fetch;
 
 use codecs::{ToByte, FromByte};
 use connection::KafkaConnection;
 use error::{Result, Error, KafkaCode};
 use protocol::{self, FromResponse};
-// pub re-export
-pub use protocol::zfetch;
 use utils;
 
 
@@ -228,7 +227,8 @@ impl FetchOffset {
 
 // view objects on the currently loaded topics metadata ---------------
 
-/// An immutable iterator over a kafka client's known topics.
+/// An immutable view on the load metadata about topics and their partitions.
+#[derive(Debug)]
 pub struct Topics<'a> {
     topic_partitions: &'a HashMap<String, TopicPartitions>,
 }
@@ -401,11 +401,14 @@ impl KafkaClient {
     /// # Example
     ///
     /// ```no_run
-    /// let mut client = kafka::client::KafkaClient::new(vec!["localhost:9092".to_owned()]);
+    /// use kafka::client::KafkaClient;
+    /// use kafka::utils::TopicPartitionOffset;
+    ///
+    /// 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_min_bytes(64 * 1024);
-    /// let r = client.fetch_messages("my-topic", 0, 0);
+    /// let r = client.fetch_messages(&[TopicPartitionOffset::new("my-topic", 0, 0)]);
     /// ```
     ///
     /// See also `KafkaClient::set_fetch_max_wait_time(..)` and
@@ -609,8 +612,9 @@ impl KafkaClient {
     /// ```
     /// Returns a vector of offset data for each available partition.
     /// PartitionOffset will contain parition and offset info Or Error code as returned by Kafka.
+    // XXX rename to fetch_topic_offsets (plural)
     pub fn fetch_topic_offset<T: AsRef<str>>(&mut self, topic: T, offset: FetchOffset)
-                                             -> Result<Vec<utils::PartitionOffset>>
+                                              -> Result<Vec<utils::PartitionOffset>>
     {
         let topic = topic.as_ref();
 
@@ -623,67 +627,22 @@ impl KafkaClient {
         }
     }
 
-    /// Fetch messages from Kafka (multiple topic, partition, offset)
-    ///
-    /// It takes a vector of `utils:TopicPartitionOffset` and returns a vector of `utils::TopicMessage`
-    /// or error::Error
-    ///
-    /// # Examples
-    ///
-    /// ```no_run
-    /// use kafka::client::KafkaClient;
-    /// use kafka::utils::TopicPartitionOffset;
-    ///
-    /// let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
-    /// client.load_metadata_all().unwrap();
-    /// let reqs = &[TopicPartitionOffset{ topic: "my-topic", partition: 0, offset: 0 },
-    ///              TopicPartitionOffset{ topic: "my-topic-2", partition: 0, offset: 0 }];
-    /// let msgs = client.fetch_messages_multi(reqs).unwrap();
-    /// ```
-    pub fn fetch_messages_multi<'a, I, J>(&mut self, input: I) -> Result<Vec<utils::TopicMessage>>
-        where J: AsRef<utils::TopicPartitionOffset<'a>>, I: IntoIterator<Item=J>
-    {
-        let reqs = __prepare_fetch_messages_requests(&mut self.state, &self.config, input);
-        __fetch_messages_multi(&mut self.conn_pool, reqs)
-    }
-
-    /// Fetch messages from Kafka (single topic, partition, offset)
-    ///
-    /// It takes a single topic, parition and offset and return a vector of messages (`utils::TopicMessage`)
-    /// or error::Error
-    ///
-    /// # Examples
-    ///
-    /// ```no_run
-    /// use kafka::client::KafkaClient;
-    ///
-    /// let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
-    /// client.load_metadata_all().unwrap();
-    /// let msgs = client.fetch_messages("my-topic", 0, 0);
-    /// ```
-    pub fn fetch_messages<T: AsRef<str>>(&mut self, topic: T, partition: i32, offset: i64)
-                                         -> Result<Vec<utils::TopicMessage>>
-    {
-        self.fetch_messages_multi(&[utils::TopicPartitionOffset::new(topic.as_ref(), partition, offset)])
-    }
-
     /// Fetch messages from Kafka (multiple topic, partition, offset)
     /// exposing low level details.
     ///
-    /// Just as `fetch_messages_multi` it takes a vector specifying
-    /// the partitions and their offsets as of which to fetch
-    /// messages.  However, unlike `fetch_messages_multi` this method
-    /// exposes the result in a raw, more complicated manner but
-    /// allows for more efficient consumption possibilities. In
-    /// particular, each of the returned fetch responses directly
-    /// corresponds to fetch requests to the underlying kafka brokers.
-    /// Except of transparently uncompressing compressed messages, the
-    /// result is not otherwise prepared.
+    /// It takes a vector specifying the partitions and their offsets
+    /// as of which to fetch messages.  The result is exposed in a
+    /// raw, complicated manner but allows for very efficient
+    /// consumption possibilities.  In particular, each of the
+    /// returned fetch responses directly corresponds to fetch
+    /// requests to the underlying kafka brokers.  Except of
+    /// transparently uncompressing compressed messages, the result is
+    /// not otherwise prepared.
     ///
     /// All of the data available through the returned fetch responses
     /// is bound to their lifetime as that data is merely a "view"
     /// into parts of the response structs.  If you need to keep
-    /// individual messages for a longer time then the whole fetch
+    /// individual messages for a longer time than the whole fetch
     /// responses, you'll need to make a copy of the message data.
     ///
     /// # Example
@@ -699,29 +658,51 @@ impl KafkaClient {
     /// client.load_metadata_all().unwrap();
     /// let reqs = &[TopicPartitionOffset{ topic: "my-topic", partition: 0, offset: 0 },
     ///              TopicPartitionOffset{ topic: "my-topic-2", partition: 0, offset: 0 }];
-    /// let resps = client.zfetch_messages_multi(reqs).unwrap();
+    /// let resps = client.fetch_messages(reqs).unwrap();
     /// for resp in resps {
     ///   for t in resp.topics() {
     ///     for p in t.partitions() {
-    ///       match p.messages() {
-    ///         Err(e) => println!("partition error: {}:{}: {}", t.topic(), p.partition(), e),
-    ///         Ok(messages) => {
+    ///       match p.data() {
+    ///         &Err(ref e) => {
+    ///           println!("partition error: {}:{}: {}", t.topic(), p.partition(), e)
+    ///         }
+    ///         &Ok(ref data) => {
     ///           println!("topic: {} / partition: {} / latest available message offset: {}",
-    ///                    t.topic(), p.partition(), messages.highwatermark_offset());
-    ///           for msg in messages {
+    ///                    t.topic(), p.partition(), data.highwatermark_offset());
+    ///           for msg in data.messages() {
     ///             println!("topic: {} / partition: {} / message.offset: {} / message.len: {}",
-    ///                      t.topic(), p.partition(), msg.offset(), msg.value().len());
+    ///                      t.topic(), p.partition(), msg.offset, msg.value.len());
     ///           }
     ///         }
     ///       }
     ///     }
     ///   }
     /// }
-    pub fn zfetch_messages_multi<'a, I, J>(&mut self, input: I) -> Result<Vec<zfetch::FetchResponse>>
-        where J: AsRef<utils::TopicPartitionOffset<'a>>, I: IntoIterator<Item=J> {
+    /// ```
+    /// See also `kafka::consumer`.
+    pub fn fetch_messages<'a, I, J>(&mut self, input: I) -> Result<Vec<fetch::FetchResponse>>
+        where J: AsRef<utils::TopicPartitionOffset<'a>>, I: IntoIterator<Item=J>
+    {
+        let state = &mut self.state;
+        let config = &self.config;
+
+        let correlation = state.next_correlation_id();
 
-        let reqs = __prepare_fetch_messages_requests(&mut self.state, &self.config, input);
-        __zfetch_messages_multi(&mut self.conn_pool, reqs)
+        // Map topic and partition to the corresponding broker
+        let mut reqs: HashMap<&str, protocol::FetchRequest> = HashMap::new();
+        for tpo in input {
+            let tpo = tpo.as_ref();
+            if let Some(broker) = state.find_broker(tpo.topic, tpo.partition) {
+                reqs.entry(broker)
+                    .or_insert_with(|| {
+                        protocol::FetchRequest::new(correlation, &config.client_id,
+                                                    config.fetch_max_wait_time, config.fetch_min_bytes)
+                    })
+                    .add(tpo.topic, tpo.partition, tpo.offset, config.fetch_max_bytes_per_partition);
+            }
+        }
+
+        __fetch_messages(&mut self.conn_pool, reqs)
     }
 
     /// Send a message to Kafka
@@ -902,6 +883,7 @@ impl KafkaClient {
     /// or `error::Error`
     ///
     /// # Examples
+
     ///
     /// ```no_run
     /// use kafka::utils;
@@ -910,7 +892,8 @@ impl KafkaClient {
     /// let resp = client.fetch_group_offsets("my-group", "my-topic");
     /// ```
     pub fn fetch_group_offsets(&mut self, group: &str, topic: &str)
-                               -> Result<Vec<utils::TopicPartitionOffsetError>> {
+                               -> Result<Vec<utils::TopicPartitionOffsetError>>
+    {
         let tps: Vec<_> =
             match self.state.topic_partitions.get(topic) {
                 None => return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)),
@@ -948,52 +931,14 @@ fn __fetch_group_offsets_multi(conn_pool: &mut ConnectionPool,
     Ok(res)
 }
 
-fn __prepare_fetch_messages_requests<'a, 'b, 'c, I, J>(
-    state: &'c mut ClientState, config: &'b ClientConfig, input: I)
-    -> HashMap<&'c str, protocol::FetchRequest<'b, 'a>>
-    where J: AsRef<utils::TopicPartitionOffset<'a>>, I: IntoIterator<Item=J> {
-
-        let correlation = state.next_correlation_id();
-
-        // Map topic and partition to the corresponding broker
-        let mut reqs: HashMap<&str, protocol::FetchRequest> = HashMap::new();
-        for tpo in input {
-            let tpo = tpo.as_ref();
-            if let Some(broker) = state.find_broker(tpo.topic, tpo.partition) {
-                reqs.entry(broker)
-                    .or_insert_with(|| {
-                        protocol::FetchRequest::new(correlation, &config.client_id,
-                                                    config.fetch_max_wait_time, config.fetch_min_bytes)
-                    })
-                    .add(tpo.topic, tpo.partition, tpo.offset, config.fetch_max_bytes_per_partition);
-            }
-        }
-        reqs
-}
-
-/// ~ carries out the given fetch requests and returns the response
-fn __fetch_messages_multi(conn_pool: &mut ConnectionPool,
-                          reqs: HashMap<&str, protocol::FetchRequest>)
-                          -> Result<Vec<utils::TopicMessage>>
-{
-    // Call each broker with the request formed earlier
-    let mut res: Vec<utils::TopicMessage> = vec!();
-    for (host, req) in reqs {
-        let resp = try!(__send_receive::<protocol::FetchRequest, protocol::FetchResponse>(conn_pool, host, req));
-        res.extend(resp.into_messages());
-    }
-    Ok(res)
-}
-
 /// ~ carries out the given fetch requests and returns the response
-fn __zfetch_messages_multi(conn_pool: &mut ConnectionPool,
-                           reqs: HashMap<&str, protocol::FetchRequest>)
-                           -> Result<Vec<zfetch::FetchResponse>>
+fn __fetch_messages(conn_pool: &mut ConnectionPool, reqs: HashMap<&str, protocol::FetchRequest>)
+                    -> Result<Vec<fetch::FetchResponse>>
 {
     // Call each broker with the request formed earlier
     let mut res = Vec::with_capacity(reqs.len());
     for (host, req) in reqs {
-        let resp = try!(__z_send_receive::<protocol::FetchRequest, zfetch::FetchResponse>(conn_pool, host, req));
+        let resp = try!(__z_send_receive::<protocol::FetchRequest, fetch::FetchResponse>(conn_pool, host, req));
         res.push(resp);
     }
     Ok(res)
diff --git a/src/consumer.rs b/src/consumer.rs
index 014f74476..01d858049 100644
--- a/src/consumer.rs
+++ b/src/consumer.rs
@@ -1,221 +1,508 @@
 //! Kafka Consumer
 //!
-//! A simple consumer based on KafkaClient. Accepts an instance of KafkaClient, a group and a
-//! topic. Partitions can be specified using builder pattern (Assumes all partitions if not
-//! specified).
+//! A consumer for a single Kafka topic on behalf of a specified group
+//! providing help in offset management.  The consumer can be
+//! optionally advised to consume only particular partitions of the
+//! underlying topic.
 //!
 //! # Example
-//!
 //! ```no_run
-//! let mut client = kafka::client::KafkaClient::new(vec!("localhost:9092".to_owned()));
-//! let res = client.load_metadata_all();
-//! let con = kafka::consumer::Consumer::new(client, "test-group".to_owned(), "my-topic".to_owned())
-//!             .partition(0)
-//!             .fallback_offset(kafka::client::FetchOffset::Earliest);
-//! for msg in con {
-//!     println!("{:?}", msg);
+//! use kafka::client::{KafkaClient, FetchOffset};
+//! use kafka::consumer::Consumer;
+//!
+//! let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
+//! client.load_metadata_all().unwrap();
+//! let mut consumer = Consumer::new(client, "my-group".to_owned(), "my-topic".to_owned())
+//!                     .with_partitions(&[0, 1])
+//!                     .with_fallback_offset(FetchOffset::Earliest);
+//! loop {
+//!   for ms in consumer.poll().unwrap().iter() {
+//!     for m in ms.messages() {
+//!       println!("{:?}", m);
+//!     }
+//!     consumer.consume_messageset(ms);
+//!   }
+//!   consumer.commit_consumed();
 //! }
 //! ```
 //!
-//! Consumer auto-commits the offsets after consuming COMMIT_INTERVAL messages (Currently set at
-//! 100)
+//! A `.poll()` will ask for the next available "chunk of data" for
+//! client code to process.  The returned data are `MessageSet`s - at
+//! most one for each partition of the consumed topic.
 //!
-//! Consumer implements Iterator.
+//! The consumer helps in keeping track of already consumed messages
+//! by maintaining a map of the consumed offsets.  Messages can be
+//! told "consumed" either through `consume_message` or
+//! `consume_messages` methods.  Once these consumed messages are
+//! committed to Kafka using `commit_consumed`, the consumer will
+//! start fetching messages from here after restart. Since committing
+//! is a certain overhead, it is up to the client to decide the
+//! frequency of the commits.  The consumer will *not* commit any
+//! messages to Kafka automatically.
 
-use std::collections::HashMap;
-use error::{Error, Result, KafkaCode};
-use utils::{TopicMessage, TopicPartitionOffset, TopicPartitionOffsetError, PartitionOffset};
-use client::{KafkaClient, FetchOffset};
+use std::collections::hash_map::{HashMap,Entry};
+use std::slice;
 
-const COMMIT_INTERVAL: i32 = 100; // Commit after every 100 message
+use client::{KafkaClient, FetchOffset, Topics};
+use error::{Error, KafkaCode, Result};
+use utils::{TopicPartition, TopicPartitionOffset};
 
-#[derive(Debug)]
+// public re-exports
+pub use client::fetch::Message;
+use client::fetch::{FetchResponse, TopicFetchResponse, PartitionFetchResponse};
+
+/// The Kafka Consumer
+///
+/// See module level documentation.
 pub struct Consumer {
-    state: ConsumerState,
     client: KafkaClient,
+    state: State,
+    config: Config,
+}
+
+#[derive(Debug)]
+struct State {
+    /// Contains the offsets which to fetch next for each partition
+    fetch_offsets: HashMap<i32, i64>, // XXX might want to change into a Vec<i64> where index denotes the partition_id
+
+    /// Contains the offsets of messages marked as "consumed" (to be
+    /// committed)
+    consumed_offsets: HashMap<i32, i64>, // XXX might want to change into a Vec<i64> where index denotes the partition_id
+    /// `true` iff the consumed_offsets contain data which needs to be
+    /// committed. Set to `false` after commit.
+    consumed_offsets_dirty: bool,
 }
 
-#[derive(Default, Debug)]
-struct ConsumerState {
+struct Config {
     group: String,
     topic: String,
     partitions: Vec<i32>,
-    initialized: bool,
-    messages: Vec<TopicMessage>,
-    index: usize,
-    offsets: HashMap<i32, i64>,
-    consumed: i32,
-
     fallback_offset: Option<FetchOffset>,
 }
 
-impl ConsumerState {
-    fn commit_offsets(&mut self, client: &mut KafkaClient) -> Result<()> {
-        client.commit_offsets(&self.group,
-                              self.offsets.iter()
-                              .map(|(p, o)| TopicPartitionOffset{
-                                  topic: &self.topic,
-                                  partition: *p,
-                                  offset: *o
-                              }))
-    }
-
-    fn fetch_offsets(&mut self, client: &mut KafkaClient) -> Result<()> {
-        if self.partitions.is_empty() {
-            // ~ fails if the underlying topic is unkonwn to the given
-            // client; this actually is what we want
-            self.partitions = match client.topics().partitions(&self.topic) {
-                None => return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)),
-                Some(tp) => tp.as_slice().iter().map(|p| p.id()).collect(),
-            };
+impl Consumer {
+
+    /// Create a new consumer over the specified managing offsets on
+    /// behalf of the specified group.
+    pub fn new(client: KafkaClient, group: String, topic: String) -> Consumer {
+        Consumer {
+            client: client,
+            state: State::new(),
+            config: Config {
+                group: group,
+                topic: topic,
+                partitions: Vec::new(),
+                fallback_offset: None,
+            },
         }
+    }
 
-        // ~ fetch the so far commited group offsets
-        let mut tpos = try!(client.fetch_group_offsets(&self.group, &self.topic));
+    /// Specify the offset to use when none was committed for the
+    /// underlying group yet.
+    ///
+    /// Running the underlying group for the first time against a
+    /// topic results in the question where to start reading from the
+    /// topic?  It might already contain a lot of messages.  Common
+    /// strategies are starting at the earliest available message
+    /// (thereby consuming whatever is currently in the topic) or at
+    /// the latest one (thereby staring to consume only newly arriving
+    /// messages.)  The parameter here corresponds to `time` in
+    /// `KafkaClient::fetch_offsets`.
+    ///
+    /// Unless this method is called and there is no offset committed
+    /// for the underlying group yet, this consumer will _not_ retrieve
+    /// any messages from the underlying topic.
+    // XXX rename to set_fallback_offset
+    pub fn with_fallback_offset(mut self, fallback_offset_time: FetchOffset) -> Consumer {
+        self.config.fallback_offset = Some(fallback_offset_time);
+        // XXX potentially reinitialize offsets and pre-fetched data
+        self
+    }
 
-        // ~ it might well that there were no group offsets committed
-        // yet ... fallback to default offsets.
-        try!(self.set_fallback_offsets(client, &mut tpos));
+    /// Explicitely specifies the partitions to consume. This will
+    /// override any previous assignments.
+    ///
+    /// If this function is never called, all available partitions of
+    /// the underlying topic will be consumed assumed.
+    pub fn with_partitions(mut self, partition: &[i32]) -> Consumer {
+        self.config.partitions.extend(partition);
+        // XXX might need to reinitialize offsets and potentially pre-fetched data
+        self
+    }
 
-        // ~ now initialized from the fetched offsets
-        for tpo in &tpos {
-            if self.partitions.contains(&tpo.partition) {
-                self.offsets.insert(tpo.partition, tpo.offset);
-            }
-        }
-        Ok(())
+    /// Destroys this consumer returning back the underlying kafka client.
+    pub fn client(self) -> KafkaClient {
+        self.client
     }
 
-    /// Try setting the "fallback offsets" for all of `tpos` where
-    /// `offset == -1`. Fails if retrieving the fallback offset is not
-    /// possible for some reason for the affected elements from
-    /// `tpos`.
-    fn set_fallback_offsets(&mut self, client: &mut KafkaClient, tpos: &mut [TopicPartitionOffsetError])
-                            -> Result<()>
-    {
-        // ~ it looks like kafka (0.8.2.1) is sending an error code
-        // (even though it documents it won't: https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetFetchResponse)
-        // so let's check only for `offset == -1` (and don't verify `error == None`)
-        let has_missing_offs = tpos.iter().filter(|tpo| tpo.offset == -1).next().is_some();
-        if has_missing_offs {
-            // ~ if the user specified a "fallback offset" strategy
-            // will try to fetch the current offset for that strategy
-            // and start consuming starting at those
-            if let Some(fallback_offset) = self.fallback_offset {
-                // ~ now fetch the offset according to the specified strategy
-                let new_offs = try!(client.fetch_topic_offset(&self.topic, fallback_offset));
-                // ehm ... not really fast (O(n^2))
-                for tpo in tpos.iter_mut() {
-                    if tpo.offset == -1 {
-                        if let Some(&PartitionOffset {offset: Ok(offset), ..})
-                            = new_offs.iter().find(|pt| pt.partition == tpo.partition)
-                        {
-                            tpo.offset = offset;
-                            tpo.error = None;
+    /// Poll for the next available message data.
+    pub fn poll(&mut self) -> Result<MessageSets> {
+        if !self.state.initialized() {
+            try!(self.state.initialize(&self.config, &mut self.client));
+            debug!("initialized: (topic: {} / group: {} / state: {:?})",
+                   self.config.topic, self.config.group, self.state);
+        }
+
+        let resps = try!(self.fetch_messages());
+
+        // XXX in future, immediately issue one more fetch_messages in
+        // the background such that the next time the client polls
+        // that request's response will likely be ready for
+        // consumption
+
+        let mut empty = true;
+        for resp in &resps {
+            for t in resp.topics() {
+                for p in t.partitions() {
+                    // XXX handle partitions with errors (will
+                    // probably need to refetch metadata)
+
+                    // XXX when a partition is empty but has a higher
+                    // highwatermark-offset than the one we fetched
+                    // from ... try to increase the max-fetch-size in
+                    // the next fetch request
+
+                    if let &Ok(ref data) = p.data() {
+                        if let Some(last_msg) = data.messages().last() {
+                            empty = false;
+                            self.state.fetch_offsets.insert(p.partition(), last_msg.offset + 1);
                         }
                     }
                 }
-            } else {
-                // XXX might want to produce some log message and return a dedicated error code
-                return Err(Error::Kafka(KafkaCode::Unknown));
             }
         }
+
+        Ok(MessageSets{ responses: resps, empty: empty })
+    }
+
+    /// Retrieves the offset of the last "consumed" message in the
+    /// specified partition. Results in `None` if there is no such
+    /// "consumed" message for this consumer's group in the underlying
+    /// topic.
+    pub fn last_consumed_message(&self, partition: i32) -> Option<i64> {
+        self.state.consumed_offsets.get(&partition).cloned()
+    }
+
+    /// Marks the message at the specified offset in the specified
+    /// partition as consumed by the caller.
+    ///
+    /// Note: a message with a "later/higher" offset automatically
+    /// marks all preceeding messages as "consumed", this is messages
+    /// with "earlier/lower" offsets in the same partition.
+    /// Therefore, it is not neccessary to invoke this method for
+    /// every consumed message.
+    pub fn consume_message(&mut self, partition: i32, offset: i64) {
+        match self.state.consumed_offsets.entry(partition) {
+            Entry::Vacant(v) => {
+                v.insert(offset);
+                self.state.consumed_offsets_dirty = true;
+            }
+            Entry::Occupied(mut v) => {
+                let o = v.get_mut();
+                if offset > *o {
+                     *o = offset;
+                    self.state.consumed_offsets_dirty = true;
+                }
+            }
+        }
+    }
+
+    /// A convience method to mark the given message set consumed as a
+    /// whole by the caller.  This is equivalent to marking the last
+    /// message of the given set as consumed.
+    pub fn consume_messageset<'a>(&mut self, msgs: MessageSet<'a>) {
+        debug_assert_eq!(msgs.topic, self.config.topic);
+
+        if !msgs.messages.is_empty() {
+            self.consume_message(
+                msgs.partition, msgs.messages.last().unwrap().offset);
+        }
+    }
+
+    /// Persists the so-far "marked as consumed" messages (on behalf
+    /// of this consumer's group for the underlying topic.)
+    ///
+    /// See also `Consumer::consume_message` and `Consumer::consume_messetset`.
+    // XXX offer an async version as well
+    pub fn commit_consumed(&mut self) -> Result<()> {
+        if !self.state.consumed_offsets_dirty {
+            debug!("no new consumed offsets to commit.");
+            return Ok(());
+        }
+
+        let client = &mut self.client;
+        let (group, topic) = (&self.config.group, &self.config.topic);
+        let offsets = &self.state.consumed_offsets;
+        debug!("commiting consumed offsets (topic: {} / group: {} / offsets: {:?}",
+               topic, group, offsets);
+        try!(client.commit_offsets(
+            group,
+            offsets.iter()
+                .map(|(&p, &o)| TopicPartitionOffset {
+                    topic: topic,
+                    partition: p,
+                    offset: o,
+                })));
+        self.state.consumed_offsets_dirty = false;
         Ok(())
     }
 
-    fn make_request(&mut self, client: &mut KafkaClient) -> Result<()> {
-        if ! self.initialized {
-            try!(self.fetch_offsets(client));
+    fn fetch_messages(&mut self) -> Result<Vec<FetchResponse>> {
+        let topic = &self.config.topic;
+        let fetch_offsets = &self.state.fetch_offsets;
+        debug!("fetching messages: (topic: {} / fetch-offsets: {:?})", topic, fetch_offsets);
+        let reqs = fetch_offsets.iter().map(|(&p, &o)| TopicPartitionOffset::new(topic, p, o));
+        self.client.fetch_messages(reqs)
+    }
+}
+
+// --------------------------------------------------------------------
+
+impl State {
+    fn new() -> State {
+        State {
+            fetch_offsets: HashMap::new(),
+            consumed_offsets: HashMap::new(),
+            consumed_offsets_dirty: false,
         }
-        let msg =
-            try!(client.fetch_messages_multi(
-                self.offsets.iter()
-                    .map(|(p, o)| TopicPartitionOffset::new(&self.topic, *p, *o))));
-        self.messages = msg;
-        self.initialized = true;
-        self.index = 0;
+    }
+
+    fn initialized(&self) -> bool {
+        !self.fetch_offsets.is_empty()
+    }
+
+    fn initialize(&mut self, config: &Config, client: &mut KafkaClient) -> Result<()> {
+        let partitions = try!(determine_partitions(config, client.topics()));
+        let consumed_offsets = try!(load_consumed_offsets(config, client, &partitions));
+        let fetch_offsets = try!(load_fetch_offsets(config, client, &partitions, &consumed_offsets));
+
+        self.fetch_offsets = fetch_offsets;
+        self.consumed_offsets = consumed_offsets;
+        self.consumed_offsets_dirty = false;
         Ok(())
     }
+}
 
-    fn next_message(&mut self, client: &mut KafkaClient) -> Option<TopicMessage> {
-        if self.initialized {
-            self.index += 1;
-            self.consumed += 1;
-            if self.consumed % COMMIT_INTERVAL == 0 {
-                let _ = self.commit_offsets(client);
-            }
-            if self.index <= self.messages.len() {
-                if self.messages[self.index-1].message.is_ok() {
-                    let curr = self.offsets.entry(self.messages[self.index-1].partition).or_insert(0);
-                    *curr = *curr+1;
-                    return Some(self.messages[self.index-1].clone());
+/// Determines the partitions to be consumed according to the
+/// specified configuration. Returns a ordered list of the determined
+/// partition ids.
+fn determine_partitions(config: &Config, metadata: Topics) -> Result<Vec<i32>> {
+    let avail_partitions = match metadata.partitions(&config.topic) {
+        // ~ fail if the underlying topic is unkonwn to the given client
+        None => {
+            debug!("no such topic: {} (all metadata: {:?})", config.topic, metadata);
+            return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
+        }
+        Some(tp) => tp,
+    };
+
+    let ps = if config.partitions.is_empty() {
+        // ~ no partitions configured ... use all available
+        avail_partitions.iter().map(|p| p.id()).collect()
+    } else {
+        // ~ validate that all partitions we're going to consume are
+        // available
+        let mut ps: Vec<i32> = Vec::with_capacity(config.partitions.len());
+        for &p in config.partitions.iter() {
+            match avail_partitions.partition(p) {
+                None => {
+                    debug!("no such partition: {} (all metadata for {}: {:?})",
+                           p, config.topic, avail_partitions.as_slice());
+                    return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
                 }
-                return None;
-            }
-            let _ = self.commit_offsets(client);
-            if self.messages.is_empty() {
-                return None;
-            }
+                Some(_) => ps.push(p),
+            };
         }
-        match self.make_request(client) {
-            Err(_) => None,
-            Ok(_) => self.next_message(client)
+        ps.sort();
+        ps.dedup();
+        ps
+    };
+
+    Ok(ps)
+}
+
+// Fetches the so-far commited/consumed offsets for the configured
+// group/topic/partitions.
+fn load_consumed_offsets(config: &Config, client: &mut KafkaClient, partitions: &[i32])
+                         -> Result<HashMap<i32, i64>>
+{
+    assert!(!partitions.is_empty());
+
+    let topic = &config.topic;
+    let tpos = try!(client.fetch_group_offsets_multi(
+        &config.group,
+        partitions.iter().map(|&p_id | TopicPartition::new(topic, p_id))));
+
+    let mut offs = HashMap::with_capacity(partitions.len());
+    for tpo in tpos {
+        if tpo.error.is_none() && tpo.offset != -1 {
+            offs.insert(tpo.partition, tpo.offset);
         }
     }
+    Ok(offs)
 }
 
-impl Consumer {
+/// Fetches the "next fetch" offsets based on the specified
+/// configuration and the given consumed offsets.
+fn load_fetch_offsets(config: &Config,
+                      client: &mut KafkaClient,
+                      partitions: &[i32],
+                      consumed_offsets: &HashMap<i32, i64>)
+                      -> Result<HashMap<i32, i64>>
+{
+    fn load_partition_offsets(client: &mut KafkaClient, topic: &str, offset: FetchOffset)
+                              -> Result<HashMap<i32, i64>> {
+        let offs = try!(client.fetch_topic_offset(topic, offset));
+        let mut m = HashMap::with_capacity(offs.len());
+        for off in offs {
+            m.insert(off.partition, off.offset.unwrap_or(-1));
+        }
+        Ok(m)
+    }
 
-    /// Constructor
-    ///
-    /// Create a new consumer. Expects a KafkaClient, group, and topic as arguments.
-    pub fn new(client: KafkaClient, group: String, topic: String) -> Consumer {
-        Consumer {
-            state: ConsumerState {
-                group: group,
-                topic: topic,
-                initialized: false,
-                index: 0,
-                .. ConsumerState::default()
-            },
-            client: client,
+    // fetch the earliest and latest available offsets
+    let latest = try!(load_partition_offsets(client, &config.topic, FetchOffset::Latest));
+    let earliest = try!(load_partition_offsets(client, &config.topic, FetchOffset::Earliest));
+
+    // ~ for each partition if we have a consumed_offset verify it is
+    // in the earliest/latest range and use that consumed_offset+1 as
+    // the fetch_message.
+    let mut fetch_offsets = HashMap::new();
+    for p in partitions {
+        let (&l_off, &e_off) = (latest.get(p).unwrap_or(&-1), earliest.get(p).unwrap_or(&-1));
+        fetch_offsets.insert(
+            *p, match consumed_offsets.get(p) {
+                Some(&co) if co >= e_off && co < l_off => co + 1,
+                _ => {
+                    match config.fallback_offset {
+                        Some(FetchOffset::Latest) => l_off,
+                        Some(FetchOffset::Earliest) => e_off,
+                        _ => {
+                            debug!("cannot determine fetch offset (group: {} / topic: {} / partition: {})",
+                                   &config.group, &config.topic, p);
+                            return Err(Error::Kafka(KafkaCode::Unknown));
+                        }
+                    }
+                }
+            }
+            );
+    }
+    Ok(fetch_offsets)
+}
+
+// --------------------------------------------------------------------
+
+/// Messages retrieved from kafka in one fetch request.  This is a
+/// concatenation of blocks of messages successfully retrieved from
+/// the consumed topic partitions.  Each such partitions is guaranteed
+/// to be present at most once in this structure.
+pub struct MessageSets {
+    responses: Vec<FetchResponse>,
+
+    /// Precomputed; Says whether there are some messages or whether
+    /// the responses actually contain consumeable messages
+    empty: bool
+}
+
+impl MessageSets {
+    /// Determines efficiently whether there are any consumeable
+    /// messages in this data set.
+    pub fn is_empty(&self) -> bool {
+        self.empty
+    }
+
+    /// Iterates over the message sets delivering the fetched message
+    /// data of consumed topic partitions.
+    pub fn iter(&self) -> MessageSetsIter {
+        let mut responses = self.responses.iter();
+        let mut topics = responses.next().map(|r| r.topics().iter());
+        let (curr_topic, partitions) =
+            topics.as_mut()
+            .and_then(|t| t.next())
+            .map_or((None, None), |t| (Some(t.topic()), Some(t.partitions().iter())));
+        MessageSetsIter {
+            responses: responses,
+            topics: topics,
+            curr_topic: curr_topic.unwrap_or(""),
+            partitions: partitions,
         }
     }
+}
 
-    /// Set the partitions of this consumer.
-    ///
-    /// If this function is never called, all partitions are assumed.
-    /// This function call be called multiple times to add more than 1 partitions.
-    pub fn partition(mut self, partition: i32) -> Consumer {
-        self.state.partitions.push(partition);
-        self
+/// A set of messages succesfully retrieved from a specific topic
+/// partition.
+pub struct MessageSet<'a> {
+    topic: &'a str,
+    partition: i32,
+    messages: &'a [Message<'a>],
+}
+
+impl<'a> MessageSet<'a> {
+    #[inline]
+    pub fn topic(&self) -> &'a str {
+        self.topic
     }
 
-    /// Specify the offset to use when none was committed for the
-    /// underlying group yet.
-    ///
-    /// Running the underlying group for the first time against a
-    /// topic results in the question where to start reading from the
-    /// topic?  It might already contain a lot of messages.  Common
-    /// strategies are starting at the earliest available message
-    /// (thereby consuming whatever is currently in the topic) or at
-    /// the latest one (thereby staring to consume only newly arriving
-    /// messages.)  The parameter here corresponds to `time` in
-    /// `KafkaClient::fetch_offsets`.
-    ///
-    /// Unless this method is called and there is no offset committed
-    /// for the underlying group yet, this consumer will _not_ retrieve
-    /// any messages from the underlying topic.
-    pub fn fallback_offset(mut self, fallback_offset_time: FetchOffset) -> Consumer {
-        self.state.fallback_offset = Some(fallback_offset_time);
-        self
+    #[inline]
+    pub fn partition(&self) -> i32 {
+        self.partition
+    }
+
+    #[inline]
+    pub fn messages(&self) -> &'a [Message<'a>] {
+        self.messages
     }
 }
 
-impl Iterator for Consumer {
-    type Item = TopicMessage;
+/// An iterator over the consumed topic partition message sets.
+pub struct MessageSetsIter<'a> {
+    responses: slice::Iter<'a, FetchResponse>,
+    topics: Option<slice::Iter<'a, TopicFetchResponse<'a>>>,
+    curr_topic: &'a str,
+    partitions: Option<slice::Iter<'a, PartitionFetchResponse<'a>>>,
+}
+
+impl<'a> Iterator for MessageSetsIter<'a> {
+    type Item = MessageSet<'a>;
 
-    fn next(&mut self) -> Option<TopicMessage> {
-        self.state.next_message(&mut self.client)
+    fn next(&mut self) -> Option<Self::Item> {
+        loop {
+            // ~ then the next available partition
+            if let Some(p) = self.partitions.as_mut().and_then(|p| p.next()) {
+                // ~ skip errornous partitions
+                // ~ skip empty partitions
+                match p.data() {
+                    &Err(_) => {
+                        continue;
+                    }
+                    &Ok(ref pdata) => {
+                        let msgs = pdata.messages();
+                        if msgs.is_empty() {
+                            continue;
+                        } else {
+                            return Some(MessageSet {
+                                topic: self.curr_topic,
+                                partition: p.partition(),
+                                messages: msgs,
+                            });
+                        }
+                    }
+                }
+            }
+            // ~ then the next available topic
+            if let Some(t) = self.topics.as_mut().and_then(|t| t.next()) {
+                self.curr_topic = t.topic();
+                self.partitions = Some(t.partitions().iter());
+                continue;
+            }
+            // ~ then the next available response
+            if let Some(r) = self.responses.next() {
+                self.curr_topic = "";
+                self.topics = Some(r.topics().iter());
+                continue;
+            }
+            // ~ finally we know there's nothing available anymore
+            return None;
+        }
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 13fd61d67..455e90b51 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -4,6 +4,9 @@ extern crate byteorder;
 extern crate flate2;
 extern crate crc;
 
+#[macro_use]
+extern crate log;
+
 #[cfg(all(test, feature = "nightly"))]
 extern crate test;
 
diff --git a/src/protocol/fetch.rs b/src/protocol/fetch.rs
index afb7a1642..bad672a08 100644
--- a/src/protocol/fetch.rs
+++ b/src/protocol/fetch.rs
@@ -1,360 +1,322 @@
-use std::io::{Cursor, Read, Write};
+//! A representation of fetched messages from Kafka.
 
-use codecs::{ToByte, FromByte};
-use compression::snappy::SnappyReader;
-use compression::{Compression, gzip};
-use error::{Error, Result};
-use utils::{TopicMessage, OffsetMessage};
+// Note: you'll find the request related code in `fetch_req.rs`.
 
-use super::{HeaderRequest, HeaderResponse};
-use super::{API_KEY_FETCH, API_VERSION};
+use std::borrow::Cow;
+use std::io::Read;
+use std::mem;
 
-#[derive(Debug)]
-pub struct FetchRequest<'a, 'b> {
-    pub header: HeaderRequest<'a>,
-    pub replica: i32,
-    pub max_wait_time: i32,
-    pub min_bytes: i32,
-    pub topic_partitions: Vec<TopicPartitionFetchRequest<'b>>
-}
+use error::{Error, Result};
+use compression::{gzip, Compression};
+use compression::snappy::SnappyReader;
 
-#[derive(Debug)]
-pub struct TopicPartitionFetchRequest<'a> {
-    pub topic: &'a str,
-    pub partitions: Vec<PartitionFetchRequest>
+use super::FromResponse;
+use super::zreader::ZReader;
+
+// ~ helper macro to aid parsing arrays of values (as defined by the
+// Kafka protocol.)
+macro_rules! array_of {
+    ($zreader:ident, $parse_elem:expr) => {{
+        let n_elems = try!($zreader.read_array_len());
+        let mut array = Vec::with_capacity(n_elems);
+        for _ in 0..n_elems {
+            array.push(try!($parse_elem));
+        }
+        array
+    }}
 }
 
-#[derive(Debug)]
-pub struct PartitionFetchRequest {
-    pub partition: i32,
-    pub offset: i64,
-    pub max_bytes: i32
+/// The result of a "fetch messages" request from a particular Kafka
+/// broker. Such a response can contain messages for multiple topic
+/// partitions.
+pub struct FetchResponse {
+    // used to "own" the data all other references of this struct
+    // point to.
+    #[allow(dead_code)]
+    raw_data: Vec<u8>,
+
+    correlation_id: i32,
+
+    // ~ static is used here to get around the fact that we don't want
+    // FetchResponse have a lifetime parameter as well. the field is
+    // exposed only through an accessor which binds the exposed
+    // lifetime to the lifetime of the FetchResponse instance
+    topics: Vec<TopicFetchResponse<'static>>,
 }
 
-impl<'a, 'b> FetchRequest<'a, 'b> {
-
-    pub fn new(correlation_id: i32, client_id: &'a str, max_wait_time: i32, min_bytes: i32)
-               -> FetchRequest<'a, 'b> {
-        FetchRequest {
-            header: HeaderRequest::new(
-                API_KEY_FETCH, API_VERSION, correlation_id, client_id),
-            replica: -1,
-            max_wait_time: max_wait_time,
-            min_bytes: min_bytes,
-            topic_partitions: vec!()
-        }
-    }
-
-    pub fn add(&mut self, topic: &'b str, partition: i32, offset: i64, max_bytes: i32) {
-        for tp in &mut self.topic_partitions {
-            if tp.topic == topic {
-                tp.add(partition, offset, max_bytes);
-                return;
-            }
-        }
-        let mut tp = TopicPartitionFetchRequest::new(topic);
-        tp.add(partition, offset, max_bytes);
-        self.topic_partitions.push(tp);
+impl FromResponse for FetchResponse {
+    fn from_response(response: Vec<u8>) -> Result<Self> {
+        FetchResponse::from_vec(response)
     }
 }
 
-impl<'a> TopicPartitionFetchRequest<'a> {
-    pub fn new(topic: &'a str) -> TopicPartitionFetchRequest<'a> {
-        TopicPartitionFetchRequest {
-            topic: topic,
-            partitions: vec!()
-        }
+impl FetchResponse {
+    /// Parses a FetchResponse from binary data as defined by the
+    /// Kafka Protocol.
+    fn from_vec(response: Vec<u8>) -> Result<FetchResponse> {
+        let slice = unsafe { mem::transmute(&response[..]) };
+        let mut r = ZReader::new(slice);
+        let correlation_id = try!(r.read_i32());
+        let topics = array_of!(r, TopicFetchResponse::read(&mut r));
+        Ok(FetchResponse {
+            raw_data: response,
+            correlation_id: correlation_id,
+            topics: topics
+        })
     }
 
-    pub fn add(&mut self, partition: i32, offset: i64, max_bytes: i32) {
-        self.partitions.push(PartitionFetchRequest:: new(partition, offset, max_bytes))
+    /// Retrieves the id corresponding to the fetch messages request
+    /// (provided for debugging purposes only).
+    #[inline]
+    pub fn correlation_id(&self) -> i32 {
+        self.correlation_id
     }
-}
 
-impl PartitionFetchRequest {
-    pub fn new(partition: i32, offset: i64, max_bytes: i32) -> PartitionFetchRequest {
-        PartitionFetchRequest {
-            partition: partition,
-            offset: offset,
-            max_bytes: max_bytes,
-        }
+    /// Provides an iterator over all the topics and the fetched data
+    /// relative to these topics.
+    #[inline]
+    pub fn topics<'a>(&'a self) -> &[TopicFetchResponse<'a>] {
+        &self.topics
     }
 }
 
-impl<'a, 'b> ToByte for FetchRequest<'a, 'b> {
-    fn encode<W: Write>(&self, buffer: &mut W) -> Result<()> {
-        try_multi!(
-            self.header.encode(buffer),
-            self.replica.encode(buffer),
-            self.max_wait_time.encode(buffer),
-            self.min_bytes.encode(buffer),
-            self.topic_partitions.encode(buffer)
-        )
-    }
+/// The result of a "fetch messages" request from a particular Kafka
+/// broker for a single topic only.  Beside the name of the topic,
+/// this structure provides an iterator over the topic partitions from
+/// which messages were requested.
+pub struct TopicFetchResponse<'a> {
+    topic: &'a str,
+    partitions: Vec<PartitionFetchResponse<'a>>,
 }
 
-impl<'a> ToByte for TopicPartitionFetchRequest<'a> {
-    fn encode<W: Write>(&self, buffer: &mut W) -> Result<()> {
-        try_multi!(
-            self.topic.encode(buffer),
-            self.partitions.encode(buffer)
-        )
+impl<'a> TopicFetchResponse<'a> {
+    fn read(r: &mut ZReader<'a>) -> Result<TopicFetchResponse<'a>> {
+        let name = try!(r.read_str());
+        let partitions = array_of!(r, PartitionFetchResponse::read(r));
+        Ok(TopicFetchResponse {
+            topic: name,
+            partitions: partitions,
+        })
     }
-}
 
-impl ToByte for PartitionFetchRequest {
-    fn encode<T:Write>(&self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.partition.encode(buffer),
-            self.offset.encode(buffer),
-            self.max_bytes.encode(buffer)
-        )
+    /// Retrieves the identifier/name of the represented topic.
+    #[inline]
+    pub fn topic(&self) -> &'a str {
+        self.topic
     }
-}
-
-// --------------------------------------------------------------------
-
-#[derive(Default, Debug)]
-pub struct FetchResponse {
-    pub header: HeaderResponse,
-    pub topic_partitions: Vec<TopicPartitionFetchResponse>,
-}
 
-#[derive(Default, Debug)]
-pub struct TopicPartitionFetchResponse {
-    pub topic: String,
-    pub partitions: Vec<PartitionFetchResponse>,
+    /// Provides an iterator over all the partitions of this topic for
+    /// which messages were requested.
+    #[inline]
+    pub fn partitions(&self) -> &[PartitionFetchResponse<'a>] {
+        &self.partitions
+    }
 }
 
-#[derive(Default, Debug)]
-pub struct PartitionFetchResponse {
-    pub partition: i32,
-    pub error: i16,
-    pub highwatermark_offset: i64,
-    pub message_set: MessageSet
+/// The result of a "fetch messages" request from a particular Kafka
+/// broker for a single topic partition only.  Beside the partition
+/// identifier, this structure provides an iterator over the actually
+/// requested message data.
+///
+/// Note: There might have been a (recoverable) error for a particular
+/// partition (but not for another).
+pub struct PartitionFetchResponse<'a> {
+    /// The identifier of the represented partition.
+    partition: i32,
+
+    /// Either an error or the partition data.
+    data: Result<PartitionData<'a>>,
 }
 
-
-impl FetchResponse {
-    pub fn into_messages(self) -> Vec<TopicMessage> {
-        self.topic_partitions
-            .into_iter()
-            .flat_map(|tp| tp.into_messages())
-            .collect()
+impl<'a> PartitionFetchResponse<'a> {
+    fn read(r: &mut ZReader<'a>) -> Result<PartitionFetchResponse<'a>> {
+        let partition = try!(r.read_i32());
+        let error = Error::from_protocol_error(try!(r.read_i16()));
+        // we need to parse the rest even if there was an error to
+        // consume the input stream (zreader)
+        let highwatermark = try!(r.read_i64());
+        let msgset = try!(MessageSet::from_slice(try!(r.read_bytes())));
+        Ok(PartitionFetchResponse {
+            partition: partition,
+            data: match error {
+                Some(error) => Err(error),
+                None => Ok(PartitionData {
+                    highwatermark_offset: highwatermark,
+                    message_set: msgset,
+                }),
+            },
+        })
     }
-}
 
-impl TopicPartitionFetchResponse {
-    pub fn into_messages(self) -> Vec<TopicMessage> {
-        let topic = self.topic;
-        self.partitions
-            .into_iter()
-            .flat_map(|p| p.into_messages(topic.clone()))
-            .collect()
+    /// Retrieves the identifier of the represented partition.
+    #[inline]
+    pub fn partition(&self) -> i32 {
+        self.partition
     }
-}
 
-impl PartitionFetchResponse {
-    pub fn into_messages(self, topic: String) -> Vec<TopicMessage> {
-        if let Some(e) = Error::from_protocol_error(self.error) {
-            return vec!(TopicMessage {
-                topic: topic,
-                partition: self.partition,
-                message: Err(e),
-            });
-        }
-        let partition = self.partition;
-        self.message_set
-            .into_messages()
-            .into_iter()
-            .map(|om| TopicMessage {
-                topic: topic.clone(),
-                partition: partition,
-                message: Ok(om)
-            })
-            .collect()
+    /// Retrieves the data payload for this partition.
+    pub fn data(&'a self) -> &'a Result<PartitionData<'a>> {
+        &self.data
     }
 }
 
-impl FromByte for FetchResponse {
-    type R = FetchResponse;
-
-    #[allow(unused_must_use)]
-    fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.header.decode(buffer),
-            self.topic_partitions.decode(buffer)
-        )
-    }
+/// The successfully fetched data payload for a particular partition.
+pub struct PartitionData<'a> {
+    highwatermark_offset: i64,
+    message_set: MessageSet<'a>,
 }
 
-impl FromByte for TopicPartitionFetchResponse {
-    type R = TopicPartitionFetchResponse;
-
-    #[allow(unused_must_use)]
-    fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.topic.decode(buffer),
-            self.partitions.decode(buffer)
-        )
+impl<'a> PartitionData<'a> {
+    /// Retrieves the so-called "high water mark offset" indicating
+    /// the "latest" offset for this partition at the remote broker.
+    /// This can be used by clients to find out how much behind the
+    /// latest available message they are.
+    #[inline]
+    pub fn highwatermark_offset(&self) -> i64 {
+        self.highwatermark_offset
     }
-}
 
-impl FromByte for PartitionFetchResponse {
-    type R = PartitionFetchResponse;
-
-    #[allow(unused_must_use)]
-    fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.partition.decode(buffer),
-            self.error.decode(buffer),
-            self.highwatermark_offset.decode(buffer),
-            self.message_set.decode(buffer)
-        )
+    /// Retrieves the fetched message data for this partition.
+    #[inline]
+    pub fn messages(&self) -> &[Message<'a>] {
+        return &self.message_set.messages
     }
 }
 
-#[derive(Default, Debug)]
-pub struct MessageSet {
-    pub message: Vec<MessageSetInner>
+struct MessageSet<'a> {
+    #[allow(dead_code)]
+    raw_data: Cow<'a, [u8]>, // ~ this field is used to potentially "own" the underlying vector
+    messages: Vec<Message<'a>>,
 }
 
-#[derive(Default, Debug)]
-pub struct MessageSetInner {
+/// A fetched message from a remote Kafka broker for a particular
+/// topic partition.
+#[derive(Debug)]
+pub struct Message<'a> {
+    /// The offset at which this message resides in the remote kafka
+    /// broker topic partition.
     pub offset: i64,
-    pub messagesize: i32,
-    pub message: Message
-}
-
-#[derive(Default, Debug)]
-pub struct Message {
-    pub crc: i32,
-    pub magic: i8,
-    pub attributes: i8,
-    pub key: Vec<u8>,
-    pub value: Vec<u8>
-}
-
-impl MessageSet {
-    fn into_messages(self) -> Vec<OffsetMessage> {
-        self.message
-            .into_iter()
-            .flat_map(|m| m.into_messages())
-            .collect()
-    }
-}
 
-impl MessageSetInner {
-    fn into_messages(self) -> Vec<OffsetMessage>{
-        self.message.into_messages(self.offset)
-    }
-}
+    /// The "key" data of this message.  Empty if there is no such
+    /// data for this message.
+    pub key: &'a [u8],
 
-impl Message {
-    fn into_messages(self, offset: i64) -> Vec<OffsetMessage> {
-        match self.attributes & 3 {
-            codec if codec == Compression::NONE as i8 => vec!(OffsetMessage{
-                offset:offset,
-                value: self.value
-            }),
-            codec if codec == Compression::GZIP as i8 => message_decode_gzip(self.value),
-            codec if codec == Compression::SNAPPY as i8 => message_decode_snappy(self.value),
-            _ => vec!()
-        }
-    }
-}
-
-fn message_decode_snappy(value: Vec<u8>) -> Vec<OffsetMessage> {
-    // XXX avoid unwrap, have this return an error!!!
-    let mut r = SnappyReader::new(&value).unwrap();
-    let mset = MessageSet::decode_new(&mut r).unwrap();
-    mset.into_messages()
+    /// The value data of this message.  Empty if there is no such
+    /// data for this message.
+    pub value: &'a [u8],
 }
 
-fn message_decode_gzip(value: Vec<u8>) -> Vec<OffsetMessage>{
-    // Gzip
-    let mut buffer = Cursor::new(value);
-    // XXX is the loop really necessary?
-    match message_decode_loop_gzip(&mut buffer) {
-        Ok(x) => x,
-        Err(_) => vec!()
+impl<'a> MessageSet<'a> {
+    fn from_vec(data: Vec<u8>) -> Result<MessageSet<'a>> {
+        // since we're going to keep the original
+        // uncompressed vector around without
+        // further modifying it and providing
+        // publicly no mutability possibilities
+        // this is safe
+        let ms = try!(MessageSet::from_slice(unsafe {
+            mem::transmute(&data[..])
+        }));
+        return Ok(MessageSet {
+            raw_data: Cow::Owned(data),
+            messages: ms.messages,
+        });
     }
-}
 
-fn message_decode_loop_gzip<T:Read>(buffer: &mut T) -> Result<Vec<OffsetMessage>> {
-    let msg = try!(gzip::uncompress(buffer));
-    let mset = try!(MessageSet::decode_new(&mut Cursor::new(msg)));
-    Ok(mset.into_messages())
-}
-
-impl FromByte for MessageSet {
-    type R = MessageSet;
-
-    fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        let mssize = try!(i32::decode_new(buffer));
-        if mssize <= 0 { return Ok(()); }
-
-        let mssize = mssize as u64;
-        let mut buf = buffer.take(mssize);
-
-        while buf.limit() > 0 {
-            match MessageSetInner::decode_new(&mut buf) {
-                Ok(val) => self.message.push(val),
-                // handle partial trailing messages (see #17)
-                Err(Error::UnexpectedEOF) => (),
-                Err(err) => return Err(err)
-            }
+    fn from_slice<'b>(raw_data: &'b [u8]) -> Result<MessageSet<'b>> {
+        let mut r = ZReader::new(raw_data);
+        let mut msgs = Vec::new();
+        while !r.is_empty() {
+            match MessageSet::next_message(&mut r) {
+                // this is the last messages which might be
+                // incomplete; a valid case to be handled by
+                // consumers
+                Err(Error::UnexpectedEOF) => {
+                    break;
+                }
+                Err(e) => {
+                    return Err(e);
+                }
+                Ok((offset, pmsg)) => {
+                    // handle compression (denoted by the last 2 bits
+                    // of the attr field)
+                    match pmsg.attr & 0x03 {
+                        c if c == Compression::NONE as i8 => {
+                            msgs.push(Message {
+                                offset: offset,
+                                key: pmsg.key,
+                                value: pmsg.value
+                            });
+                        }
+                        // XXX handle recursive compression in future
+                        c if c == Compression::GZIP as i8 => {
+                            let v = try!(gzip::uncompress(pmsg.value));
+                            return Ok(try!(MessageSet::from_vec(v)));
+                        }
+                        c if c == Compression::SNAPPY as i8 => {
+                            let mut v = Vec::new();
+                            try!(try!(SnappyReader::new(pmsg.value)).read_to_end(&mut v));
+                            return Ok(try!(MessageSet::from_vec(v)));
+                        }
+                        _ => panic!("Unknown compression type!"),
+                    }
+                }
+            };
         }
-        Ok(())
+        Ok(MessageSet {
+            raw_data: Cow::Borrowed(raw_data),
+            messages: msgs,
+        })
     }
 
-    fn decode_new<T: Read>(buffer: &mut T) -> Result<Self::R> {
-        let mut temp: Self::R = Default::default();
-        while let Ok(mi) = MessageSetInner::decode_new(buffer) {
-            temp.message.push(mi);
-        }
-        if temp.message.is_empty() {
-            return Err(Error::UnexpectedEOF)
-        }
-        Ok(temp)
+    fn next_message<'b>(r: &mut ZReader<'b>) -> Result<(i64, ProtocolMessage<'b>)> {
+        let offset = try!(r.read_i64());
+        let msg_data = try!(r.read_bytes());
+        Ok((offset, try!(ProtocolMessage::from_slice(msg_data))))
     }
 }
 
-impl FromByte for MessageSetInner {
-    type R = MessageSetInner;
-
-    #[allow(unused_must_use)]
-    fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.offset.decode(buffer),
-            self.messagesize.decode(buffer),
-            self.message.decode(buffer)
-        )
-    }
+/// Represents a messages exactly as defined in the protocol.
+struct ProtocolMessage<'a> {
+    crc: i32,
+    magic: i8,
+    attr: i8,
+    key: &'a [u8],
+    value: &'a [u8],
 }
 
-impl FromByte for Message {
-    type R = Message;
-
-    fn decode<T: Read>(&mut self, buffer: &mut T) -> Result<()> {
-        try_multi!(
-            self.crc.decode(buffer),
-            self.magic.decode(buffer),
-            self.attributes.decode(buffer),
-            self.key.decode(buffer),
-            self.value.decode(buffer)
-        )
+impl<'a> ProtocolMessage<'a> {
+    /// Parses a raw message from the given byte slice.  Does _not_
+    /// handle any compression.
+    fn from_slice<'b>(raw_data: &'b [u8]) -> Result<ProtocolMessage<'b>> {
+        let mut r = ZReader::new(raw_data);
+
+        let msg_crc = try!(r.read_i32());
+        // XXX later validate the checksum
+        let msg_magic = try!(r.read_i8());
+        // XXX validate that `msg_magic == 0`
+        let msg_attr = try!(r.read_i8());
+        let msg_key = try!(r.read_bytes());
+        let msg_val = try!(r.read_bytes());
+
+        debug_assert!(r.is_empty());
+
+        Ok(ProtocolMessage {
+            crc: msg_crc,
+            magic: msg_magic,
+            attr: msg_attr,
+            key: msg_key,
+            value: msg_val,
+        })
     }
 }
 
-// --------------------------------------------------------------------
+// tests --------------------------------------------------------------
 
 #[cfg(test)]
 mod tests {
     use std::str;
 
-    use super::FetchResponse;
-    use codecs::FromByte;
+    use super::{FetchResponse, Message};
 
     static FETCH1_TXT: &'static str =
         include_str!("../../test-data/fetch1.txt");
@@ -367,76 +329,102 @@ mod tests {
     static FETCH1_FETCH_RESPONSE_GZIP_K0821: &'static [u8] =
         include_bytes!("../../test-data/fetch1.mytopic.1p.gzip.kafka.0821");
 
-    fn test_decode_new_fetch_response(msg_per_line: &str, mut fetch_response_bytes: &[u8]) {
-        let resp = FetchResponse::decode_new(&mut fetch_response_bytes);
-        let resp = resp.unwrap();
+    fn into_messages<'a>(r: &'a FetchResponse) -> Vec<&'a Message<'a>> {
+        let mut all_msgs = Vec::new();
+        for t in r.topics() {
+            for p in t.partitions() {
+                match p.data() {
+                    &Err(_) => {
+                        println!("Skipping error partition: {}:{}", t.topic, p.partition);
+                    }
+                    &Ok(ref data) => {
+                        all_msgs.extend(data.messages());
+                    }
+                }
+            }
+        }
+        all_msgs
+    }
 
-        // ~ response for exactly one topic expected
-        assert_eq!(1, resp.topic_partitions.len());
+    fn test_decode_new_fetch_response(msg_per_line: &str, response: Vec<u8>) {
+        let resp = FetchResponse::from_vec(response);
+        let resp = resp.unwrap();
 
-        let msgs = resp.into_messages();
         let original: Vec<_> = msg_per_line.lines().collect();
+
+        // ~ response for exactly one topic expected
+        assert_eq!(1, resp.topics.len());
+        // ~ topic name
+        assert_eq!("my-topic", resp.topics[0].topic);
+        // ~ exactly one partition
+        assert_eq!(1, resp.topics[0].partitions.len());
+        // ~ the first partition
+        assert_eq!(0, resp.topics[0].partitions[0].partition);
+        // ~ no error
+        assert!(resp.topics[0].partitions[0].data.is_ok());
+
+        let msgs = into_messages(&resp);
         assert_eq!(original.len(), msgs.len());
         for (msg, orig) in msgs.into_iter().zip(original.iter()) {
-            assert_eq!(str::from_utf8(&msg.message.unwrap().value).unwrap(), *orig);
+            assert_eq!(str::from_utf8(msg.value).unwrap(), *orig);
         }
     }
 
     #[test]
-    fn test_decode_new_fetch_response_nocompression_k0821() {
-        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821);
+    fn test_from_slice_nocompression_k0821() {
+        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned());
     }
 
     #[test]
-    fn test_decode_new_fetch_response_snappy_k0821() {
-        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_SNAPPY_K0821);
+    fn test_from_slice_snappy_k0821() {
+        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned());
     }
 
     #[test]
-    fn test_decode_new_fetch_response_snappy_k0822() {
-        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_SNAPPY_K0822);
+    fn test_from_slice_snappy_k0822() {
+        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned());
     }
 
     #[test]
-    fn test_decode_new_fetch_response_gzip_k0821() {
-        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_GZIP_K0821);
+    fn test_from_slice_gzip_k0821() {
+        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned());
     }
 
     #[cfg(feature = "nightly")]
     mod benches {
-        use std::io::Cursor;
         use test::{black_box, Bencher};
 
-        use codecs::FromByte;
         use super::super::FetchResponse;
+        use super::into_messages;
 
-        fn bench_decode_new_fetch_response(b: &mut Bencher, data: &[u8]) {
+        fn bench_decode_new_fetch_response(b: &mut Bencher, data: Vec<u8>) {
             b.bytes = data.len() as u64;
             b.iter(|| {
-                let r = black_box(FetchResponse::decode_new(&mut Cursor::new(data)).unwrap());
-                let v = black_box(r.into_messages());
+                let data = data.clone();
+                let r = black_box(FetchResponse::from_vec(data).unwrap());
+                let v = black_box(into_messages(&r));
                 v.len()
             });
         }
 
         #[bench]
         fn bench_decode_new_fetch_response_nocompression_k0821(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821)
+            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned())
         }
 
         #[bench]
         fn bench_decode_new_fetch_response_snappy_k0821(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0821)
+            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned())
         }
 
         #[bench]
         fn bench_decode_new_fetch_response_snappy_k0822(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0822)
+            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned())
         }
 
         #[bench]
         fn bench_decode_new_fetch_response_gzip_k0821(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_GZIP_K0821)
+            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned())
         }
     }
 }
diff --git a/src/protocol/fetch_req.rs b/src/protocol/fetch_req.rs
new file mode 100644
index 000000000..aeed7bfd8
--- /dev/null
+++ b/src/protocol/fetch_req.rs
@@ -0,0 +1,114 @@
+//! 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.
+
+use error::Result;
+use std::io::Write;
+
+use codecs::ToByte;
+use super::HeaderRequest;
+use super::{API_KEY_FETCH, API_VERSION};
+
+
+#[derive(Debug)]
+pub struct FetchRequest<'a, 'b> {
+    pub header: HeaderRequest<'a>,
+    pub replica: i32,
+    pub max_wait_time: i32,
+    pub min_bytes: i32,
+    pub topic_partitions: Vec<TopicPartitionFetchRequest<'b>>
+}
+
+#[derive(Debug)]
+pub struct TopicPartitionFetchRequest<'a> {
+    pub topic: &'a str,
+    pub partitions: Vec<PartitionFetchRequest>
+}
+
+#[derive(Debug)]
+pub struct PartitionFetchRequest {
+    pub partition: i32,
+    pub offset: i64,
+    pub max_bytes: i32
+}
+
+impl<'a, 'b> FetchRequest<'a, 'b> {
+
+    pub fn new(correlation_id: i32, client_id: &'a str, max_wait_time: i32, min_bytes: i32)
+               -> FetchRequest<'a, 'b> {
+        FetchRequest {
+            header: HeaderRequest::new(
+                API_KEY_FETCH, API_VERSION, correlation_id, client_id),
+            replica: -1,
+            max_wait_time: max_wait_time,
+            min_bytes: min_bytes,
+            topic_partitions: vec!()
+        }
+    }
+
+    pub fn add(&mut self, topic: &'b str, partition: i32, offset: i64, max_bytes: i32) {
+        for tp in &mut self.topic_partitions {
+            if tp.topic == topic {
+                tp.add(partition, offset, max_bytes);
+                return;
+            }
+        }
+        let mut tp = TopicPartitionFetchRequest::new(topic);
+        tp.add(partition, offset, max_bytes);
+        self.topic_partitions.push(tp);
+    }
+}
+
+impl<'a> TopicPartitionFetchRequest<'a> {
+    pub fn new(topic: &'a str) -> TopicPartitionFetchRequest<'a> {
+        TopicPartitionFetchRequest {
+            topic: topic,
+            partitions: vec!()
+        }
+    }
+
+    pub fn add(&mut self, partition: i32, offset: i64, max_bytes: i32) {
+        self.partitions.push(PartitionFetchRequest:: new(partition, offset, max_bytes))
+    }
+}
+
+impl PartitionFetchRequest {
+    pub fn new(partition: i32, offset: i64, max_bytes: i32) -> PartitionFetchRequest {
+        PartitionFetchRequest {
+            partition: partition,
+            offset: offset,
+            max_bytes: max_bytes,
+        }
+    }
+}
+
+impl<'a, 'b> ToByte for FetchRequest<'a, 'b> {
+    fn encode<W: Write>(&self, buffer: &mut W) -> Result<()> {
+        try_multi!(
+            self.header.encode(buffer),
+            self.replica.encode(buffer),
+            self.max_wait_time.encode(buffer),
+            self.min_bytes.encode(buffer),
+            self.topic_partitions.encode(buffer)
+        )
+    }
+}
+
+impl<'a> ToByte for TopicPartitionFetchRequest<'a> {
+    fn encode<W: Write>(&self, buffer: &mut W) -> Result<()> {
+        try_multi!(
+            self.topic.encode(buffer),
+            self.partitions.encode(buffer)
+        )
+    }
+}
+
+impl ToByte for PartitionFetchRequest {
+    fn encode<T:Write>(&self, buffer: &mut T) -> Result<()> {
+        try_multi!(
+            self.partition.encode(buffer),
+            self.offset.encode(buffer),
+            self.max_bytes.encode(buffer)
+        )
+    }
+}
diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs
index 9414b7b03..af2b055fb 100644
--- a/src/protocol/mod.rs
+++ b/src/protocol/mod.rs
@@ -16,17 +16,17 @@ macro_rules! try_multi {
 mod produce;
 mod offset;
 mod metadata;
-mod fetch;
 mod consumer;
 
 mod zreader;
-pub mod zfetch;
+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::fetch::{FetchRequest, FetchResponse};
 // pub use self::consumer::{ConsumerMetadataRequest, ConsumerMetadataResponse};
 pub use self::consumer::{OffsetFetchRequest, OffsetFetchResponse,
                          OffsetCommitRequest, OffsetCommitResponse};
diff --git a/src/protocol/zfetch.rs b/src/protocol/zfetch.rs
deleted file mode 100644
index 03d50399c..000000000
--- a/src/protocol/zfetch.rs
+++ /dev/null
@@ -1,587 +0,0 @@
-//! An efficient response parser for the "fetch messages" use-case.
-
-use std::borrow::Cow;
-use std::io::Read;
-use std::mem;
-use std::slice::Iter;
-
-use error::{Error, Result};
-use compression::{gzip, Compression};
-use compression::snappy::SnappyReader;
-
-use super::FromResponse;
-use super::zreader::ZReader;
-
-// ~ helper macro to aid parsing arrays of values (as defined by the
-// Kafka protocol.)
-macro_rules! array_of {
-    ($zreader:ident, $parse_elem:expr) => {{
-        let n_elems = try!($zreader.read_array_len());
-        let mut array = Vec::with_capacity(n_elems);
-        for _ in 0..n_elems {
-            array.push(try!($parse_elem));
-        }
-        array
-    }}
-}
-
-/// The result of a "fetch messages" request from a particular Kafka
-/// broker. Such a response can contain messages for multiple topic
-/// partitions.
-pub struct FetchResponse {
-    // used to "own" the data all other references of this struct
-    // point to.
-    #[allow(dead_code)]
-    raw_data: Vec<u8>,
-
-    correlation_id: i32,
-
-    // ~ static is used here to get around the fact that we don't want
-    // FetchResponse have a lifetime parameter as well. the field is
-    // exposed only through an accessor which binds the exposed
-    // lifetime to the lifetime of the FetchResponse instance
-    topics: Vec<TopicFetchResponse<'static>>,
-}
-
-impl FromResponse for FetchResponse {
-    fn from_response(response: Vec<u8>) -> Result<Self> {
-        FetchResponse::from_vec(response)
-    }
-}
-
-impl FetchResponse {
-    /// Parses a FetchResponse from binary data as defined by the
-    /// Kafka Protocol.
-    fn from_vec(response: Vec<u8>) -> Result<FetchResponse> {
-        let slice = unsafe { mem::transmute(&response[..]) };
-        let mut r = ZReader::new(slice);
-        let correlation_id = try!(r.read_i32());
-        let topics = array_of!(r, TopicFetchResponse::read(&mut r));
-        Ok(FetchResponse {
-            raw_data: response,
-            correlation_id: correlation_id,
-            topics: topics
-        })
-    }
-
-    /// Retrieves the id corresponding to the fetch messages request
-    /// (provided for debugging purposes only).
-    #[inline]
-    pub fn correlation_id(&self) -> i32 {
-        self.correlation_id
-    }
-
-    /// Provides an iterator over all the topics and the fetched data
-    /// relative to these topics.
-    #[inline]
-    pub fn topics<'a>(&'a self) -> Topics<'a> {
-        Topics { iter: self.topics.iter() }
-    }
-}
-
-/// An iterator over the topics of a `FetchResponse`.
-pub struct Topics<'a> {
-    iter: Iter<'a, TopicFetchResponse<'a>>,
-}
-
-impl<'a> Iterator for Topics<'a> {
-    type Item = &'a TopicFetchResponse<'a>;
-
-    #[inline]
-    fn next(&mut self) -> Option<Self::Item> {
-        self.iter.next()
-    }
-}
-
-/// The result of a "fetch messages" request from a particular Kafka
-/// broker for a single topic only.  Beside the name of the topic,
-/// this structure provides an iterator over the topic partitions from
-/// which messages were requested.
-pub struct TopicFetchResponse<'a> {
-    topic: &'a str,
-    partitions: Vec<PartitionFetchResponse<'a>>,
-}
-
-impl<'a> TopicFetchResponse<'a> {
-    fn read(r: &mut ZReader<'a>) -> Result<TopicFetchResponse<'a>> {
-        let name = try!(r.read_str());
-        let partitions = array_of!(r, PartitionFetchResponse::read(r));
-        Ok(TopicFetchResponse {
-            topic: name,
-            partitions: partitions,
-        })
-    }
-
-    /// Retrieves the identifier/name of the represented topic.
-    #[inline]
-    pub fn topic(&self) -> &'a str {
-        self.topic
-    }
-
-    /// Provides an iterator over all the partitions of this topic for
-    /// which messages were requested.
-    #[inline]
-    pub fn partitions<'b>(&'b self) -> Partitions<'b, 'a> {
-        Partitions { iter: self.partitions.iter() }
-    }
-}
-
-/// An iterator over the partitions of a `TopicFetchResponse`.
-pub struct Partitions<'a, 'b: 'a> {
-    iter: Iter<'a, PartitionFetchResponse<'b>>,
-}
-
-impl<'a, 'b> Iterator for Partitions<'a, 'b> {
-    type Item = &'a PartitionFetchResponse<'b>;
-
-    #[inline]
-    fn next(&mut self) -> Option<Self::Item> {
-        self.iter.next()
-    }
-}
-
-
-/// The result of a "fetch messages" request from a particular Kafka
-/// broker for a single topic partition only.  Beside the partition
-/// identifier, this structure provides an iterator over the actually
-/// requested message data.
-///
-/// Note: There might have been an error for a particular partition
-/// (but not for another), in which case data accessors of this
-/// structure will result in that error.
-pub struct PartitionFetchResponse<'a> {
-    /// The identifier of the represented partition.
-    partition: i32,
-
-    /// Either an error or the partition data.
-    data: Result<PartitionData<'a>>,
-}
-
-struct PartitionData<'a> {
-    highwatermark_offset: i64,
-    message_set: MessageSet<'a>,
-}
-
-impl<'a> PartitionFetchResponse<'a> {
-    fn read(r: &mut ZReader<'a>) -> Result<PartitionFetchResponse<'a>> {
-        let partition = try!(r.read_i32());
-        let error = Error::from_protocol_error(try!(r.read_i16()));
-        // we need to parse the rest even if there was an error to
-        // consume the input stream (zreader)
-        let highwatermark = try!(r.read_i64());
-        let msgset = try!(MessageSet::from_slice(try!(r.read_bytes())));
-        Ok(PartitionFetchResponse {
-            partition: partition,
-            data: match error {
-                Some(error) => Err(error),
-                None => Ok(PartitionData {
-                    highwatermark_offset: highwatermark,
-                    message_set: msgset,
-                }),
-            },
-        })
-    }
-
-    /// Retrieves the identifier of the represented partition.
-    #[inline]
-    pub fn partition(&self) -> i32 {
-        self.partition
-    }
-
-    /// Retrieves the data payload for this partition.
-    #[inline]
-    pub fn messages<'s>(&'s self) -> Result<Messages<'s, 'a>> {
-        match self.data {
-            Ok(ref data) => Ok(Messages {
-                highwatermark_offset: data.highwatermark_offset,
-                iter: data.message_set.messages.iter()
-            }),
-            Err(ref e) => Err(e.clone()),
-        }
-    }
-}
-
-/// An iterator over the messages of a `PartitionFetchResponse`.
-pub struct Messages<'a, 'b: 'a> {
-    highwatermark_offset: i64,
-    iter: Iter<'a, Message<'b>>,
-}
-
-impl<'a, 'b: 'a> Messages<'a, 'b> {
-    /// Retrieves the so-called "high water mark offset" indicating
-    /// the "latest" offset for the partition of this message set at
-    /// the remote broker.  This can be used by clients to find out
-    /// how much behind the latest message available in the particular
-    /// partition they are.
-    #[inline]
-    pub fn highwatermark_offset(&self) -> i64 {
-        self.highwatermark_offset
-    }
-}
-
-impl<'a, 'b> Iterator for Messages<'a, 'b> {
-    type Item = &'a Message<'b>;
-
-    #[inline]
-    fn next(&mut self) -> Option<Self::Item> {
-        self.iter.next()
-    }
-}
-
-
-struct MessageSet<'a> {
-    #[allow(dead_code)]
-    raw_data: Cow<'a, [u8]>, // ~ this field is used to potentially "own" the underlying vector
-    messages: Vec<Message<'a>>,
-}
-
-/// A fetched messages from a remote Kafka broker for a particular
-/// topic partition.
-pub struct Message<'a> {
-    offset: i64,
-    key: &'a [u8],
-    value: &'a [u8],
-}
-
-impl<'a> Message<'a> {
-    /// Retrieves the offset at which this message resides in the
-    /// remote kafka broker topic partition.
-    #[inline]
-    pub fn offset(&self) -> i64 {
-        self.offset
-    }
-
-    /// Retrieves the "key" data of this message.  Empty if there is
-    /// no such data for this message.
-    #[inline]
-    pub fn key(&self) -> &'a [u8] {
-        self.key
-    }
-
-    /// Retrieves the value data of this message.  Empty if there is
-    /// no such data for this message.
-    #[inline]
-    pub fn value(&self) -> &'a [u8] {
-        self.value
-    }
-}
-
-impl<'a> MessageSet<'a> {
-    fn from_vec(data: Vec<u8>) -> Result<MessageSet<'a>> {
-        // since we're going to keep the original
-        // uncompressed vector around without
-        // further modifying it and providing
-        // publicly no mutability possibilities
-        // this is safe
-        let ms = try!(MessageSet::from_slice(unsafe {
-            mem::transmute(&data[..])
-        }));
-        return Ok(MessageSet {
-            raw_data: Cow::Owned(data),
-            messages: ms.messages,
-        });
-    }
-
-    fn from_slice<'b>(raw_data: &'b [u8]) -> Result<MessageSet<'b>> {
-        let mut r = ZReader::new(raw_data);
-        let mut msgs = Vec::new();
-        while !r.is_empty() {
-            match MessageSet::next_message(&mut r) {
-                // this is the last messages which might be
-                // incomplete; a valid case to be handled by
-                // consumers
-                Err(Error::UnexpectedEOF) => {
-                    break;
-                }
-                Err(e) => {
-                    return Err(e);
-                }
-                Ok((offset, pmsg)) => {
-                    // handle compression (denoted by the last 2 bits
-                    // of the attr field)
-                    match pmsg.attr & 0x03 {
-                        c if c == Compression::NONE as i8 => {
-                            msgs.push(Message {
-                                offset: offset,
-                                key: pmsg.key,
-                                value: pmsg.value
-                            });
-                        }
-                        // XXX handle recursive compression in future
-                        c if c == Compression::GZIP as i8 => {
-                            let v = try!(gzip::uncompress(pmsg.value));
-                            return Ok(try!(MessageSet::from_vec(v)));
-                        }
-                        c if c == Compression::SNAPPY as i8 => {
-                            let mut v = Vec::new();
-                            try!(try!(SnappyReader::new(pmsg.value)).read_to_end(&mut v));
-                            return Ok(try!(MessageSet::from_vec(v)));
-                        }
-                        _ => panic!("Unknown compression type!"),
-                    }
-                }
-            };
-        }
-        Ok(MessageSet {
-            raw_data: Cow::Borrowed(raw_data),
-            messages: msgs,
-        })
-    }
-
-    fn next_message<'b>(r: &mut ZReader<'b>) -> Result<(i64, ProtocolMessage<'b>)> {
-        let offset = try!(r.read_i64());
-        let msg_data = try!(r.read_bytes());
-        Ok((offset, try!(ProtocolMessage::from_slice(msg_data))))
-    }
-}
-
-/// Represents a messages exactly as defined in the protocol.
-struct ProtocolMessage<'a> {
-    crc: i32,
-    magic: i8,
-    attr: i8,
-    key: &'a [u8],
-    value: &'a [u8],
-}
-
-impl<'a> ProtocolMessage<'a> {
-    /// Parses a raw message from the given byte slice.  Does _not_
-    /// handle any compression.
-    fn from_slice<'b>(raw_data: &'b [u8]) -> Result<ProtocolMessage<'b>> {
-        let mut r = ZReader::new(raw_data);
-
-        let msg_crc = try!(r.read_i32());
-        // XXX later validate the checksum
-        let msg_magic = try!(r.read_i8());
-        // XXX validate that `msg_magic == 0`
-        let msg_attr = try!(r.read_i8());
-        let msg_key = try!(r.read_bytes());
-        let msg_val = try!(r.read_bytes());
-
-        debug_assert!(r.is_empty());
-
-        Ok(ProtocolMessage {
-            crc: msg_crc,
-            magic: msg_magic,
-            attr: msg_attr,
-            key: msg_key,
-            value: msg_val,
-        })
-    }
-}
-
-// --------------------------------------------------------------------
-
-/// A convenience helper for iterating "fetch messages" responses.
-pub struct ResponseIter<'a> {
-    responses: Iter<'a, FetchResponse>,
-    topics: Option<Topics<'a>>,
-    curr_topic: &'a str,
-    partitions: Option<Partitions<'a, 'a>>,
-}
-
-/// A responce for a set of messages from a single topic partition.
-pub struct Response<'a> {
-    /// The name of the topic this response corresponds to.
-    pub topic: &'a str,
-    /// The partition this response correponds to.
-    pub partition: i32,
-    /// Either an error or the set of messages retrieved for the
-    /// underlying topic partition.
-    pub messages: Result<Messages<'a, 'a>>,
-}
-
-/// Provide a partition level iterator over all specified fetch
-/// responses. Since there might be failures on the level of
-/// partitions, it is the lowest unit at which a response can be
-/// processed. The returned iterator will iterate all partitions in
-/// the given responses in their specified order.
-///
-/// # Examples
-///
-/// ```no_run
-/// use kafka::client::KafkaClient;
-/// use kafka::utils::TopicPartitionOffset;
-/// use kafka::client::zfetch::iter_responses;
-///
-/// // Fetch some data for two topic partitions
-/// let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
-/// client.load_metadata_all().unwrap();
-/// let reqs = &[TopicPartitionOffset{ topic: "my-topic", partition: 0, offset: 0 },
-///              TopicPartitionOffset{ topic: "my-topic-2", partition: 0, offset: 0 }];
-/// let resps = client.zfetch_messages_multi(reqs).unwrap();
-///
-/// // Iterate all the responses for all the partitions specified to be fetched data from
-/// for r in iter_responses(&resps) {
-///   match r.messages {
-///     Err(e) => println!("error for {}:{}: {}", r.topic, r.partition, e),
-///     Ok(msgs) => {
-///       for msg in msgs {
-///         println!("{}:{}: {:?}", r.topic, r.partition, msg.value());
-///       }
-///     }
-///   }
-/// }
-/// ```
-pub fn iter_responses<'a>(responses: &'a [FetchResponse]) -> ResponseIter<'a> {
-    let mut responses = responses.iter();
-    let mut topics = responses.next().map(|r| r.topics());
-    let (curr_topic, partitions) =
-        topics.as_mut()
-        .and_then(|t| t.next())
-        .map_or((None, None), |t| (Some(t.topic()), Some(t.partitions())));
-    ResponseIter {
-        responses: responses,
-        topics: topics,
-        curr_topic: curr_topic.unwrap_or(""),
-        partitions: partitions,
-    }
-}
-
-impl<'a> Iterator for ResponseIter<'a> {
-    type Item = Response<'a>;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        // ~ then the next available partition
-        if let Some(p) = self.partitions.as_mut().and_then(|p| p.next()) {
-            return Some(Response {
-                topic: self.curr_topic,
-                partition: p.partition(),
-                messages: p.messages(),
-            });
-        }
-        // ~ then the next available topic
-        if let Some(t) = self.topics.as_mut().and_then(|t| t.next()) {
-            self.curr_topic = t.topic();
-            self.partitions = Some(t.partitions());
-            return self.next();
-        }
-        // ~ then the next available response
-        if let Some(r) = self.responses.next() {
-            self.curr_topic = "";
-            self.topics = Some(r.topics());
-            return self.next();
-        }
-        // ~ finally we know there's nothing available anymore
-        None
-    }
-}
-
-// tests --------------------------------------------------------------
-
-#[cfg(test)]
-mod tests {
-    use std::str;
-
-    use super::{FetchResponse, Message};
-
-    static FETCH1_TXT: &'static str =
-        include_str!("../../test-data/fetch1.txt");
-    static FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821: &'static [u8] =
-        include_bytes!("../../test-data/fetch1.mytopic.1p.nocompression.kafka.0821");
-    static FETCH1_FETCH_RESPONSE_SNAPPY_K0821: &'static [u8] =
-        include_bytes!("../../test-data/fetch1.mytopic.1p.snappy.kafka.0821");
-    static FETCH1_FETCH_RESPONSE_SNAPPY_K0822: &'static [u8] =
-        include_bytes!("../../test-data/fetch1.mytopic.1p.snappy.kafka.0822");
-    static FETCH1_FETCH_RESPONSE_GZIP_K0821: &'static [u8] =
-        include_bytes!("../../test-data/fetch1.mytopic.1p.gzip.kafka.0821");
-
-    fn into_messages<'a>(data: &'a FetchResponse) -> Vec<&'a Message<'a>> {
-        let mut all_msgs = Vec::new();
-        for t in data.topics() {
-            for p in t.partitions() {
-                match p.messages() {
-                    Err(_) => {
-                        println!("Skipping error partition: {}:{}", t.topic, p.partition);
-                    }
-                    Ok(msgs) => {
-                        all_msgs.extend(msgs);
-                    }
-                }
-            }
-        }
-        all_msgs
-    }
-
-    fn test_decode_new_fetch_response(msg_per_line: &str, response: Vec<u8>) {
-        let resp = FetchResponse::from_vec(response);
-        let resp = resp.unwrap();
-
-        let original: Vec<_> = msg_per_line.lines().collect();
-
-        // ~ response for exactly one topic expected
-        assert_eq!(1, resp.topics.len());
-        // ~ topic name
-        assert_eq!("my-topic", resp.topics[0].topic);
-        // ~ exactly one partition
-        assert_eq!(1, resp.topics[0].partitions.len());
-        // ~ the first partition
-        assert_eq!(0, resp.topics[0].partitions[0].partition);
-        // ~ no error
-        assert!(resp.topics[0].partitions[0].data.is_ok());
-
-        let msgs = into_messages(&resp);
-        assert_eq!(original.len(), msgs.len());
-        for (msg, orig) in msgs.into_iter().zip(original.iter()) {
-            assert_eq!(str::from_utf8(msg.value).unwrap(), *orig);
-        }
-    }
-
-    #[test]
-    fn test_from_slice_nocompression_k0821() {
-        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned());
-    }
-
-    #[test]
-    fn test_from_slice_snappy_k0821() {
-        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned());
-    }
-
-    #[test]
-    fn test_from_slice_snappy_k0822() {
-        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned());
-    }
-
-    #[test]
-    fn test_from_slice_gzip_k0821() {
-        test_decode_new_fetch_response(FETCH1_TXT, FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned());
-    }
-
-    #[cfg(feature = "nightly")]
-    mod benches {
-        use test::{black_box, Bencher};
-
-        use super::super::FetchResponse;
-        use super::into_messages;
-
-        fn bench_decode_new_fetch_response(b: &mut Bencher, data: Vec<u8>) {
-            b.bytes = data.len() as u64;
-            b.iter(|| {
-                let data = data.clone();
-                let r = black_box(FetchResponse::from_vec(data).unwrap());
-                let v = black_box(into_messages(&r));
-                v.len()
-            });
-        }
-
-        #[bench]
-        fn bench_decode_new_fetch_response_nocompression_k0821(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned())
-        }
-
-        #[bench]
-        fn bench_decode_new_fetch_response_snappy_k0821(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned())
-        }
-
-        #[bench]
-        fn bench_decode_new_fetch_response_snappy_k0822(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned())
-        }
-
-        #[bench]
-        fn bench_decode_new_fetch_response_gzip_k0821(b: &mut Bencher) {
-            bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned())
-        }
-    }
-}
diff --git a/src/utils.rs b/src/utils.rs
index e43a5d94b..5a601872d 100644
--- a/src/utils.rs
+++ b/src/utils.rs
@@ -76,6 +76,13 @@ pub struct TopicPartition<'a> {
     pub partition: i32
 }
 
+impl<'a> TopicPartition<'a> {
+    #[inline]
+    pub fn new(topic: &str, partition: i32) -> TopicPartition {
+        TopicPartition { topic: topic, partition: partition }
+    }
+}
+
 impl<'a> AsRef<TopicPartition<'a>> for TopicPartition<'a> {
     fn as_ref(&self) -> &Self {
         &self


Test patch: