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

Issue:
consumer: allow consuming multiple topics
I'm trying to use a Rust consumer to read from multiple topics. This is the code I have now:

``` rust
extern crate kafka;
use kafka::client::KafkaClient;
use kafka::consumer::Consumer;
use kafka::utils;
fn main(){
    let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
    let res = client.load_metadata_all();
    let topics = client.topic_partitions.keys().cloned().collect(); 
    let offsets = client.fetch_offsets(topics, -1);
    for topic in &topics {
    let mut con = Consumer::new(client, "test-consumer-group".to_owned(), "topic".to_owned()).partition(0);
    let mut messagedata = 0;
    for msg in con {
        println!("{}", str::from_utf8(&msg.message).unwrap().to_string());
    }
  }
}

```

below is the error:

```
    src/main.rs:201:19: 201:25 error: use of moved value: `topics` [E0382]
src/main.rs:201     for topic in &topics {
                                  ^~~~~~
    note: in expansion of for loop expansion
    src/main.rs:201:5: 210:6 note: expansion site
    src/main.rs:167:40: 167:46 note: `topics` moved here because it has type `collections::vec::Vec<collections::string::String>`, which is non-copyable
    src/main.rs:167     let offsets = client.fetch_offsets(topics, -1);
                                                       ^~~~~~
    src/main.rs:203:37: 203:43 error: use of moved value: `client` [E0382]
    src/main.rs:203     let mut con = Consumer::new(client, "test-consumer-group".to_owned(), "topicname".to_owned()).partition(0);
                                                    ^~~~~~
    note: in expansion of for loop expansion
    src/main.rs:201:5: 210:6 note: expansion site
    note: `client` was previously moved here because it has type     `kafka::client::KafkaClient`, which is non-copyable
    error: aborting due to 2 previous errors
```

To better explain my question, here is my partial workable code for just one topic:

``` rust
let mut con = Consumer::new(client, "test-consumer-group".to_owned(), "testtopic".to_owned()).partition(0);

for msg in con {
    println!("{}", str::from_utf8(&msg.message).unwrap().to_string());
}
```

And I tested the fetch_message function, it works for multiple topics, but the result I have (msgs) is Topicmessage, I don't know how to get message from Topicmessage.

``` rust
let msgs = client.fetch_messages_multi(vec!(utils::TopicPartitionOffset{
                                            topic: "topic1".to_string(),
                                            partition: 0,
                                            offset: 0 //from the begining
                                            },
                                        utils::TopicPartitionOffset{
                                            topic: "topic2".to_string(),
                                            partition: 0,
                                            offset: 0
                                        },
                                        utils::TopicPartitionOffset{
                                            topic: "topic3".to_string(),
                                            partition: 0,
                                            offset: 0
                                        }));
for msg in msgs{
    println!("{}", msg);
}
```



Discussion / hints:
hello @Lingling7,

the error message regarding `error: use of moved value: topics` is due to the fact that the call to `client.fetch_offsets(topics, -1)` "consumes" `topics` (a nice explanation and more details on this is the [ownership chapter of the rust book](https://doc.rust-lang.org/stable/book/ownership.html#move-semantics)). In particular, as part of #19 I would like to avoid the `client.fetch_offsets` and some other methods "consume" the topics in the long term. But for know you could work around this as follows:

```
    ...
    let topics: Vec<String> = client.topic_partitions.keys().cloned().collect(); 
    let offsets = client.fetch_offsets(topics.clone(), -1);
    ...
```

However, you'll still be left with the second error message. Unfortunately, `Consumer` does currently not provide the possibility to consume multiple topics (this might be possible in some future point in time; I'll leave this issue open for the moment in case we want to address this.)

As you noted you can use `client.fetch_messages_multi`, but you're left with managing the offsets yourself (here's an example of [a small program consuming all messages of a given list of topics](https://github.com/xitep/monchan/blob/6239b9bc3c8bc49b9bac90379c3446c7364604e9/src/main.rs#L184)).

Regarding your question about how to get to the "message" of a `TopicMessage`. You're already there :) Use "`{:?}`" instead of "`{}`" when printing the topic-message or access the members of [TopicMessages](https://github.com/spicavigo/kafka-rust/blob/master/src/utils.rs#L12) directly:

```
let msgs = client.fetch_messages_multi(...);
for msg in msgs {
    println!("msg.topic: {}", msg.topic);
    println!("msg.partition: {}", msg.partition);
    println!("msg.message: {}", msg.message); // this is the actual data of the message (you again need to "convert" it from raw bytes to something meaningful, e.g. `str::from_utf8(&msg.message).unwrap()`)
}
```

I hope this helps a little bit

@xitep  Thank you so much! I changed my code as you guided, still have the following error:

```
src/main.rs:192:37: 192:46 error: attempted access of field `topic` on type `collections::vec::Vec<kafka::utils::TopicMessage>`, but no field with that name was found
src/main.rs:192         println!("msg.topic: {:?}", msg.topic);
                                                    ^~~~~~~~~
note: in expansion of format_args!
<std macros>:2:25: 2:56 note: expansion site
<std macros>:1:1: 2:62 note: in expansion of print!
<std macros>:3:1: 3:54 note: expansion site
<std macros>:1:1: 3:58 note: in expansion of println!
src/main.rs:192:9: 192:48 note: expansion site
note: in expansion of for loop expansion
src/main.rs:190:5: 195:6 note: expansion site
src/main.rs:193:41: 193:54 error: attempted access of field `partition` on type `collections::vec::Vec<kafka::utils::TopicMessage>`, but no field with that name was found
src/main.rs:193         println!("msg.partition: {:?}", msg.partition);
                                                        ^~~~~~~~~~~~~
note: in expansion of format_args!
<std macros>:2:25: 2:56 note: expansion site
<std macros>:1:1: 2:62 note: in expansion of print!
<std macros>:3:1: 3:54 note: expansion site
<std macros>:1:1: 3:58 note: in expansion of println!
src/main.rs:193:9: 193:56 note: expansion site
note: in expansion of for loop expansion
src/main.rs:190:5: 195:6 note: expansion site
src/main.rs:194:39: 194:50 error: attempted access of field `message` on type `collections::vec::Vec<kafka::utils::TopicMessage>`, but no field with that name was found
src/main.rs:194         println!("msg.message: {:?}", msg.message);
                                                      ^~~~~~~~~~~
note: in expansion of format_args!
<std macros>:2:25: 2:56 note: expansion site
<std macros>:1:1: 2:62 note: in expansion of print!
<std macros>:3:1: 3:54 note: expansion site
<std macros>:1:1: 3:58 note: in expansion of println!
src/main.rs:194:9: 194:52 note: expansion site
note: in expansion of for loop expansion
src/main.rs:190:5: 195:6 note: expansion site
error: aborting due to 3 previous errors
```

@Lingling7 it's a bit hard for me to guess what's going on ... but i think this might be it: can you please try the following:

```
let msgs = client.fetch_messages_multi(...).unwrap(); // <-- unwrap() here
for msg in msgs {
   ....
}
```

previously i have (errornously) suggested the same without "unwrapping" and since `Result` (which is returned by `fetch_messages_multi` implements `IntoIterator` it works smoothly with the `for` loop; however, it leaves you with `msg` being of type `Vec<TopicMessage>` which, of course, doesn't have the attributes you are reaching for. i'm sorry for the confusion.

@xitep  Problem solved! Thanks for your help again!!!

:)

Hi, I'm trying to fetch topics by batch. For the fetch multiple topics messages, can I fetch one messages from one topic per time? such as:

```
topic 1, offset:1
topic 2, offset:1
topic 1, offset:2
topic 2, offset:2
......
```

Currently, the code fetching order is fetch all messages in one topic then go to the next topic.
Or I can only fetch all the messages and save to two groups, then read the groups one by one? 

Hello @Lingling7,

fetching messages one-by-one from topics/partitions is something kafka is not designed for. It'll hand out messages in "chunks" from topics/partitions (for efficiency reasons). Since these chunks might be a few hundred messages (for example) it might leave you in the impression that the client first consumes one topic and only then another. But what you really get is a mix of these chunks, which might well originate from different topics.


Fix patch:
diff --git a/examples/console-consumer.rs b/examples/console-consumer.rs
index bcd8336d6..939e1a61c 100644
--- a/examples/console-consumer.rs
+++ b/examples/console-consumer.rs
@@ -6,7 +6,7 @@ use std::{env, fmt, process};
 use std::io::{self, Write};
 use std::ascii::AsciiExt;
 
-use kafka::consumer::{Consumer, GroupOffsetStorage};
+use kafka::consumer::{Consumer, FetchOffset, GroupOffsetStorage};
 
 /// This is a very simple command line application reading from a
 /// specific kafka topic and dumping the messages to standard output.
@@ -27,15 +27,20 @@ fn main() {
 }
 
 fn process(cfg: Config) -> Result<(), Error> {
-    let mut c =
-        try!(Consumer::from_hosts(cfg.brokers, cfg.topic)
-             .with_group(cfg.group)
-             .with_fetch_max_wait_time(1000)
-             .with_fetch_min_bytes(1_000)
-             .with_fetch_max_bytes_per_partition(100_000)
-             .with_retry_max_bytes_limit(1_000_000)
-             .with_offset_storage(cfg.offset_storage)
-             .create());
+    let mut c = {
+        let mut cb = Consumer::from_hosts(cfg.brokers)
+            .with_group(cfg.group)
+            .with_fallback_offset(cfg.fallback_offset)
+            .with_fetch_max_wait_time(1000)
+            .with_fetch_min_bytes(1_000)
+            .with_fetch_max_bytes_per_partition(100_000)
+            .with_retry_max_bytes_limit(1_000_000)
+            .with_offset_storage(cfg.offset_storage);
+        for topic in cfg.topics {
+            cb = cb.with_topic(topic);
+        }
+        try!(cb.create())
+    };
 
     let stdout = io::stdout();
     let mut stdout = stdout.lock();
@@ -54,7 +59,7 @@ fn process(cfg: Config) -> Result<(), Error> {
                 // ~ write to output channel
                 try!(stdout.write_all(&buf));
             }
-            c.consume_messageset(ms);
+            let _ = c.consume_messageset(ms);
         }
         if do_commit {
             try!(c.commit_consumed());
@@ -93,9 +98,10 @@ impl From<io::Error> for Error {
 struct Config {
     brokers: Vec<String>,
     group: String,
-    topic: String,
+    topics: Vec<String>,
     no_commit: bool,
     offset_storage: GroupOffsetStorage,
+    fallback_offset: FetchOffset,
 }
 
 impl Config {
@@ -104,10 +110,12 @@ impl Config {
         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("", "topics", "Specify topics (comma separated)", "NAMES");
         opts.optopt("", "group", "Specify the consumer group", "NAME");
-        opts.optflag("", "no-commit", "Do not commit consumed messages");
-        opts.optopt("", "offsets", "Specify the offset store [zookeeper, kafka]", "STORE");
+        opts.optflag("", "no-commit", "Do not commit group offsets");
+        opts.optopt("", "storage", "Specify the offset store [zookeeper, kafka]", "STORE");
+        opts.optflag("", "earliest", "Fall back to the earliest offset \
+                                      (when no group offset available)");
 
         let m = match opts.parse(&args[1..]) {
             Ok(m) => m,
@@ -118,8 +126,25 @@ impl Config {
             return Err(Error::Literal(opts.usage(&brief)));
         }
 
+        macro_rules! required_list {
+            ($name:expr) => {{
+                let opt = $name;
+                let xs: Vec<_> = match m.opt_str(opt) {
+                    None => return Err(Error::Literal(format!("Required option --{} missing", opt))),
+                    Some(s) => s.split(',').map(|s| s.trim().to_owned()).filter(|s| !s.is_empty()).collect(),
+                };
+                if xs.is_empty() {
+                    return Err(Error::Literal(format!("Invalid --{} specified!", opt)));
+                }
+                xs
+            }}
+        };
+
+        let brokers = required_list!("brokers");
+        let topics = required_list!("topics");
+
         let mut offset_storage = GroupOffsetStorage::Zookeeper;
-        if let Some(s) = m.opt_str("offsets") {
+        if let Some(s) = m.opt_str("storage") {
             if s.eq_ignore_ascii_case("zookeeper") {
                 offset_storage = GroupOffsetStorage::Zookeeper;
             } else if s.eq_ignore_ascii_case("kafka") {
@@ -129,17 +154,16 @@ impl Config {
             }
         }
         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(|| String::new()),
-            topic: m.opt_str("topic")
-                .unwrap_or_else(|| "my-topic".to_owned()),
+            brokers: brokers,
+            group: m.opt_str("group").unwrap_or_else(|| String::new()),
+            topics: topics,
             no_commit: m.opt_present("no-commit"),
             offset_storage: offset_storage,
+            fallback_offset: if m.opt_present("earliest") {
+                FetchOffset::Earliest
+            } else {
+                FetchOffset::Latest
+            },
         })
    }
 }
diff --git a/examples/consume.rs b/examples/consume.rs
index aa807784e..15f162dcb 100644
--- a/examples/consume.rs
+++ b/examples/consume.rs
@@ -23,7 +23,8 @@ fn main() {
 fn consume_messages(group: String, topic: String, brokers: Vec<String>)
                     -> Result<(), KafkaError>
 {
-    let mut con = try!(Consumer::from_hosts(brokers, topic)
+    let mut con = try!(Consumer::from_hosts(brokers)
+                       .with_topic(topic)
                        .with_group(group)
                        .with_fallback_offset(FetchOffset::Earliest)
                        .with_offset_storage(GroupOffsetStorage::Kafka)
@@ -40,7 +41,7 @@ fn consume_messages(group: String, topic: String, brokers: Vec<String>)
             for m in ms.messages() {
                 println!("{}:{}@{}: {:?}", ms.topic(), ms.partition(), m.offset, m.value);
             }
-            con.consume_messageset(ms);
+            let _ = con.consume_messageset(ms);
         }
         try!(con.commit_consumed());
     }
diff --git a/src/client/mod.rs b/src/client/mod.rs
index ad2c20ec5..a68d832a1 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -1071,7 +1071,12 @@ impl KafkaClient {
                 return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
             }
         }
-        __commit_offsets(req, &mut self.state, &mut self.conn_pool, &self.config)
+        if req.topic_partitions.is_empty() {
+            debug!("commit_offsets: no offsets provided");
+            Ok(())
+        } else {
+            __commit_offsets(req, &mut self.state, &mut self.conn_pool, &self.config)
+        }
     }
 
     /// Commit offset of a particular topic partition on behalf of a
diff --git a/src/consumer.rs b/src/consumer.rs
deleted file mode 100644
index d31bbaa53..000000000
--- a/src/consumer.rs
+++ /dev/null
@@ -1,838 +0,0 @@
-//! Kafka Consumer - A higher-level API for consuming a kafka topic.
-//!
-//! 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
-//! use kafka::consumer::{Consumer, FetchOffset, GroupOffsetStorage};
-//!
-//! let mut consumer =
-//!    Consumer::from_hosts(vec!("localhost:9092".to_owned()),
-//!                              "my-topic".to_owned())
-//!       .with_partitions(&[0, 1])
-//!       .with_fallback_offset(FetchOffset::Earliest)
-//!       .with_group("my-group".to_owned())
-//!       .with_offset_storage(GroupOffsetStorage::Kafka)
-//!       .create()
-//!       .unwrap();
-//! loop {
-//!   for ms in consumer.poll().unwrap().iter() {
-//!     for m in ms.messages() {
-//!       println!("{:?}", m);
-//!     }
-//!     consumer.consume_messageset(ms);
-//!   }
-//!   consumer.commit_consumed().unwrap();
-//! }
-//! ```
-//!
-//! 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.
-//!
-//! If the consumer is configured for a non-empty group, it 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 even 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.
-//!
-//! The configuration of a group is optional.  If the consumer has no
-//! group configured, it will behave as if it had one, only that
-//! commiting consumed message offsets resolves into a void operation.
-
-use std::collections::hash_map::{HashMap, Entry};
-use std::collections::VecDeque;
-use std::slice;
-use std::hash::BuildHasherDefault;
-
-use fnv::FnvHasher;
-
-use client::{self, KafkaClient, FetchPartition, CommitOffset, FetchGroupOffset};
-use client::metadata::Topics;
-use error::{Error, KafkaCode, Result};
-use client::fetch;
-
-// public re-exports
-pub use client::fetch::Message;
-pub use client::FetchOffset;
-pub use client::GroupOffsetStorage;
-
-#[cfg(feature = "security")]
-use client::SecurityConfig;
-
-#[cfg(not(feature = "security"))]
-type SecurityConfig = ();
-
-/// The default value for `Builder::with_retry_max_bytes_limit`.
-pub const DEFAULT_RETRY_MAX_BYTES_LIMIT: i32 = 0;
-
-/// The default value for `Builder::with_fallback_offset`.
-pub const DEFAULT_FALLBACK_OFFSET: FetchOffset = FetchOffset::Latest;
-
-type PartitionHasher = BuildHasherDefault<FnvHasher>;
-
-/// The Kafka Consumer
-///
-/// See module level documentation.
-pub struct Consumer {
-    client: KafkaClient,
-    state: State,
-    config: Config,
-}
-
-// The "fetch state" for a particular topci partition.
-#[derive(Debug)]
-struct FetchState {
-    /// ~ specifies the offset which to fetch from
-    offset: i64,
-    /// ~ specifies the max_bytes to be fetched
-    max_bytes: i32,
-}
-
-#[derive(Debug)]
-struct State {
-    /// Contains the information relevant for the next fetch operation
-    /// on the corresponding partitions
-    fetch_offsets: HashMap<i32, FetchState, PartitionHasher>,
-
-    /// Specifies partitions to be fetched on their own in the next
-    /// poll request.
-    retry_partitions: VecDeque<i32>,
-
-    /// Contains the offsets of messages marked as "consumed" (to be
-    /// committed)
-    consumed_offsets: HashMap<i32, i64, PartitionHasher>,
-
-    /// `true` iff the consumed_offsets contain data which needs to be
-    /// committed. Set to `false` after commit.
-    consumed_offsets_dirty: bool,
-}
-
-struct Config {
-    group: String,
-    topic: String,
-    partitions: Vec<i32>,
-    fallback_offset: FetchOffset,
-    retry_max_bytes_limit: i32,
-}
-
-// XXX 1) Support multiple topics
-// XXX 2) Issue IO in a separate (background) thread and pre-fetch messagesets
-// XXX 3) Allow returning to a previous offset (aka seeking)
-
-impl Consumer {
-
-    /// Starts building a consumer for the given topic using the given
-    /// kafka client.
-    pub fn from_client(client: KafkaClient, topic: String) -> Builder {
-        Builder::new(Some(client), Vec::new()).with_topic(topic)
-    }
-
-    /// Starts building a consumer for the given topic bootstraping
-    /// internally a new kafka client from the given kafka hosts.
-    pub fn from_hosts(hosts: Vec<String>, topic: String) -> Builder {
-        Builder::new(None, hosts).with_topic(topic)
-    }
-
-    /// Destroys this consumer returning back the underlying kafka client.
-    pub fn client(self) -> KafkaClient {
-        self.client
-    }
-
-    /// Polls for the next available message data.
-    pub fn poll(&mut self) -> Result<MessageSets> {
-        let (n, resps) = self.fetch_messages();
-        self.process_fetch_responses(n, try!(resps))
-    }
-
-    /// Determines whether this consumer is set up to consume only a
-    /// single partition.
-    pub fn single_partition_consumer(&self) -> bool {
-        self.state.fetch_offsets.len() == 1
-    }
-
-    /// Retrieves the group on which behalf this consumer is acting.
-    /// The empty group name specifies a group-less consumer.
-    pub fn group(&self) -> &str {
-        &self.config.group
-    }
-
-    // ~ returns (number partitions queried, fecth responses)
-    fn fetch_messages(&mut self) -> (u32, Result<Vec<fetch::Response>>) {
-        let topic = &self.config.topic;
-        let fetch_offsets = &self.state.fetch_offsets;
-        // ~ if there's a retry partition ... fetch messages just for
-        // that one. Otherwise try to fetch messages for all assigned
-        // partitions.
-        match self.state.retry_partitions.pop_front() {
-            Some(partition) => {
-                let s = match fetch_offsets.get(&partition) {
-                    Some(fstate) => fstate,
-                    None => return (1, Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition))),
-                };
-                debug!("fetching messages: (topic: {} / fetch-offset: {{{}: {:?}}})",
-                       topic, partition, s);
-                (1, self.client.fetch_messages_for_partition(
-                    &FetchPartition::new(topic, partition, s.offset)
-                        .with_max_bytes(s.max_bytes)))
-            }
-            None => {
-                debug!("fetching messages: (topic: {} / fetch-offsets: {:?})",
-                       topic, fetch_offsets);
-                let reqs = fetch_offsets.iter()
-                    .map(|(&p, s)| {
-                        FetchPartition::new(topic, p, s.offset)
-                            .with_max_bytes(s.max_bytes)
-                    });
-                (fetch_offsets.len() as u32, self.client.fetch_messages(reqs))
-            }
-        }
-    }
-
-    // ~ post process a data retrieved through fetch_messages before
-    // handing them out to client code
-    //   - update the fetch state for the next fetch cycle
-    // ~ num_partitions_queried: the original number of partitions requested/queried for
-    //   the responses
-    // ~ resps: the responses to post process
-    fn process_fetch_responses(&mut self, num_partitions_queried: u32, resps: Vec<fetch::Response>)
-                               -> Result<MessageSets>
-    {
-        let single_partition_consumer = self.single_partition_consumer();
-        let mut empty = true;
-        let mut retry_partitions = &mut self.state.retry_partitions;
-
-        for resp in &resps {
-            for t in resp.topics() {
-                for p in t.partitions() {
-                    let partition = p.partition();
-
-                    // ~ for now, as soon as a partition has an error
-                    // we fail to prevent client programs from not
-                    // noticing.  however, in future we don't need to
-                    // fail immediately, we can try to recover from
-                    // certain errors and retry the fetch operation
-                    // transparently for the caller.
-                    let data = match p.data() {
-                        // XXX need to prevent updating fetch_offsets in case we're gonna fail here
-                        &Err(ref e) => return Err(e.clone()),
-                        &Ok(ref data) => data,
-                    };
-
-                    let mut fetch_state = self.state.fetch_offsets
-                            .get_mut(&partition)
-                            .expect("non-requested partition");
-                    // ~ book keeping
-                    if let Some(last_msg) = data.messages().last() {
-                        fetch_state.offset = last_msg.offset + 1;
-                        empty = false;
-
-                        // ~ reset the max_bytes again to its usual
-                        // value if we had a retry request and finally
-                        // got some data
-                        if fetch_state.max_bytes != self.client.fetch_max_bytes_per_partition() {
-                            let prev_max_bytes = fetch_state.max_bytes;
-                            fetch_state.max_bytes = self.client.fetch_max_bytes_per_partition();
-                            debug!("reset max_bytes for {}:{} from {} to {}",
-                                   t.topic(), partition, prev_max_bytes, fetch_state.max_bytes);
-                        }
-                    } else {
-                        debug!("no data received for {}:{} (max_bytes: {} / fetch_offset: {} / highwatermark_offset: {})",
-                               t.topic(), partition, fetch_state.max_bytes, fetch_state.offset, data.highwatermark_offset());
-
-                        // ~ when a partition is empty but has a
-                        // highwatermark-offset equal to or greater
-                        // than the one we tried to fetch ... we'll
-                        // try to increase the max-fetch-size in the
-                        // next fetch request
-                        if fetch_state.offset < data.highwatermark_offset() {
-                            if fetch_state.max_bytes < self.config.retry_max_bytes_limit {
-                                // ~ try to double the max_bytes
-                                let prev_max_bytes = fetch_state.max_bytes;
-                                let incr_max_bytes = prev_max_bytes + prev_max_bytes;
-                                if incr_max_bytes > self.config.retry_max_bytes_limit {
-                                    fetch_state.max_bytes = self.config.retry_max_bytes_limit;
-                                } else {
-                                    fetch_state.max_bytes = incr_max_bytes;
-                                }
-                                debug!("increased max_bytes for {}:{} from {} to {}",
-                                       t.topic(), partition, prev_max_bytes, fetch_state.max_bytes);
-                            } else if num_partitions_queried == 1 {
-                                // ~ this was a single partition
-                                // request and we didn't get anything
-                                // and we won't be increasing the max
-                                // fetch size ... this is will fail
-                                // forever ... signal the problem to
-                                // the user
-                                return Err(Error::Kafka(KafkaCode::MessageSizeTooLarge));
-                            }
-                            // ~ if this consumer is subscribed to one
-                            // partition only, there's no need to push
-                            // the partition to the 'retry_partitions'
-                            // (this is just a small optimization)
-                            if !single_partition_consumer {
-                                // ~ mark this partition for a retry on its own
-                                debug!("rescheduled for retry: {}:{}", t.topic(), partition);
-                                retry_partitions.push_back(partition)
-                            }
-                        }
-                    }
-                }
-            }
-        }
-
-        // XXX in future, issue one more fetch_messages request in the
-        // background such that the next time the client polls that
-        // request's response will likely be already ready for
-        // consumption
-
-        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 yet.
-    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 - if any.)
-    ///
-    /// See also `Consumer::consume_message` and
-    /// `Consumer::consume_messetset`.
-    pub fn commit_consumed(&mut self) -> Result<()> {
-        if self.config.group.is_empty() {
-            debug!("commit_consumed: ignoring commit request since no group defined");
-            return Ok(());
-        }
-        if !self.state.consumed_offsets_dirty {
-            debug!("commit_consumed: 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!("commit_consumed: commiting consumed offsets \
-                (topic: {} / group: {} / offsets: {:?}",
-               topic, group, offsets);
-        try!(client.commit_offsets(
-            group,
-            offsets.iter().map(|(&p, &o)| CommitOffset::new(topic, p, o))));
-        self.state.consumed_offsets_dirty = false;
-        Ok(())
-    }
-}
-
-// --------------------------------------------------------------------
-
-impl State {
-    fn new(client: &mut KafkaClient, config: &Config) -> Result<State> {
-        let partitions = try!(determine_partitions(config, client.topics()));
-        let consumed_offsets = try!(load_consumed_offsets(config, client, &partitions));
-        let fetch_offsets = try!(load_fetch_states(config, client, &partitions, &consumed_offsets));
-
-        Ok(State {
-            fetch_offsets: fetch_offsets,
-            retry_partitions: VecDeque::new(),
-            consumed_offsets: consumed_offsets,
-            consumed_offsets_dirty: false,
-        })
-    }
-}
-
-/// 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);
-                    return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
-                }
-                Some(_) => ps.push(p),
-            };
-        }
-        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, PartitionHasher>>
-{
-    assert!(!partitions.is_empty());
-
-    let mut offs = HashMap::with_capacity_and_hasher(
-        partitions.len(), PartitionHasher::default());
-    // ~ no group, no persisted consumed offsets
-    if config.group.is_empty() {
-        return Ok(offs);
-    }
-    // ~ otherwise try load them for the group
-    let topic = &config.topic;
-    let tpos = try!(client.fetch_group_offsets(
-        &config.group,
-        partitions.iter().map(|&p_id| FetchGroupOffset::new(topic, p_id))));
-    for tpo in tpos {
-        match tpo.offset {
-            Ok(o) if o != -1 => {
-                offs.insert(tpo.partition, o);
-            }
-            _ => {}
-        }
-    }
-    Ok(offs)
-}
-
-/// Fetches the "next fetch" offsets/states based on the specified
-/// configuration and the given consumed offsets.
-fn load_fetch_states(config: &Config,
-                      client: &mut KafkaClient,
-                      partitions: &[i32],
-                      consumed_offsets: &HashMap<i32, i64, PartitionHasher>)
-                      -> Result<HashMap<i32, FetchState, PartitionHasher>>
-{
-    fn load_partition_offsets(client: &mut KafkaClient, topic: &str, offset: FetchOffset)
-                              -> Result<HashMap<i32, i64, PartitionHasher>>
-    {
-        let offs = try!(client.fetch_topic_offsets(topic, offset));
-        let mut m = HashMap::with_capacity_and_hasher(offs.len(), PartitionHasher::default());
-        for off in offs {
-            m.insert(off.partition, off.offset.unwrap_or(-1));
-        }
-        Ok(m)
-    }
-
-    let max_bytes = client.fetch_max_bytes_per_partition();
-
-    let mut fetch_offsets = HashMap::with_capacity_and_hasher(
-        partitions.len(), PartitionHasher::default());
-
-
-    if consumed_offsets.is_empty() {
-        // ~ if there are no offsets on behalf of the consumer
-        // group - if any - we can directly use fallback offsets.
-        let offs = try!(load_partition_offsets(
-            client, &config.topic, config.fallback_offset));
-        for p in partitions {
-            fetch_offsets.insert(*p, FetchState {
-                offset: *offs.get(p).unwrap_or(&-1),
-                max_bytes: max_bytes
-            });
-        }
-    } else {
-        // 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.
-        for p in partitions {
-            let (&l_off, &e_off) = (latest.get(p).unwrap_or(&-1), earliest.get(p).unwrap_or(&-1));
-            let offset = match consumed_offsets.get(p) {
-                Some(&co) if co >= e_off && co < l_off => co + 1,
-                _ => {
-                    match config.fallback_offset {
-                        FetchOffset::Latest => l_off,
-                        FetchOffset::Earliest => e_off,
-                        _ => {
-                            debug!("cannot determine fetch offset \
-                                    (group: {} / topic: {} / partition: {})",
-                                   &config.group, &config.topic, p);
-                            return Err(Error::Kafka(KafkaCode::Unknown));
-                        }
-                    }
-                }
-            };
-            fetch_offsets.insert(*p, FetchState{ offset: offset, max_bytes: max_bytes });
-        }
-    }
-    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<fetch::Response>,
-
-    /// 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,
-        }
-    }
-}
-
-/// 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
-    }
-
-    #[inline]
-    pub fn partition(&self) -> i32 {
-        self.partition
-    }
-
-    #[inline]
-    pub fn messages(&self) -> &'a [Message<'a>] {
-        self.messages
-    }
-}
-
-/// An iterator over the consumed topic partition message sets.
-pub struct MessageSetsIter<'a> {
-    responses: slice::Iter<'a, fetch::Response>,
-    topics: Option<slice::Iter<'a, fetch::Topic<'a>>>,
-    curr_topic: &'a str,
-    partitions: Option<slice::Iter<'a, fetch::Partition<'a>>>,
-}
-
-impl<'a> Iterator for MessageSetsIter<'a> {
-    type Item = MessageSet<'a>;
-
-    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;
-        }
-    }
-}
-
-// --------------------------------------------------------------------
-
-/// A Kafka Consumer builder easing the process of setting up various
-/// configuration settings.
-#[derive(Debug)]
-pub struct Builder {
-    client: Option<KafkaClient>,
-    hosts: Vec<String>,
-    group: String,
-    topic: String,
-    partitions: Vec<i32>,
-    fallback_offset: FetchOffset,
-    fetch_max_wait_time: i32,
-    fetch_min_bytes: i32,
-    fetch_max_bytes_per_partition: i32,
-    retry_max_bytes_limit: i32,
-    fetch_crc_validation: bool,
-    security_config: Option<SecurityConfig>,
-    group_offset_storage: GroupOffsetStorage,
-}
-
-impl Builder {
-    fn new(client: Option<KafkaClient>, hosts: Vec<String>) -> Builder {
-        let mut b = Builder {
-            client: client,
-            hosts: hosts,
-            fetch_max_wait_time: client::DEFAULT_FETCH_MAX_WAIT_TIME,
-            fetch_min_bytes: client::DEFAULT_FETCH_MIN_BYTES,
-            fetch_max_bytes_per_partition: client::DEFAULT_FETCH_MAX_BYTES_PER_PARTITION,
-            fetch_crc_validation: client::DEFAULT_FETCH_CRC_VALIDATION,
-            retry_max_bytes_limit: DEFAULT_RETRY_MAX_BYTES_LIMIT,
-            group: "".to_owned(),
-            topic: "".to_owned(),
-            partitions: Vec::new(),
-            fallback_offset: DEFAULT_FALLBACK_OFFSET,
-            security_config: None,
-            group_offset_storage: client::DEFAULT_GROUP_OFFSET_STORAGE,
-        };
-        if let Some(ref c) = b.client {
-            b.fetch_max_wait_time = c.fetch_max_wait_time();
-            b.fetch_min_bytes = c.fetch_min_bytes();
-            b.fetch_max_bytes_per_partition = c.fetch_max_bytes_per_partition();
-            b.fetch_crc_validation = c.fetch_crc_validation();
-            b.group_offset_storage = c.group_offset_storage();
-        }
-        b
-    }
-
-    /// Specifies the topic to consume.
-    fn with_topic(mut self, topic: String) -> Builder {
-        self.topic = topic;
-        self
-    }
-
-    /// Specifies the group on whose behalf to maintain consumed
-    /// message offsets.
-    ///
-    /// The group is allowed to be the empty string, in which case the
-    /// resulting consumer will be group-less.
-    pub fn with_group(mut self, group: String) -> Builder {
-        self.group = group;
-        self
-    }
-
-    /// Explicitely specifies the partitions to consume.
-    ///
-    /// If this function is never called, all available partitions of
-    /// the underlying topic will be consumed assumed.
-    pub fn with_partitions(mut self, partitions: &[i32]) -> Builder {
-        self.partitions.extend_from_slice(partitions);
-        self
-    }
-
-    /// Specifies the security config to use.
-    /// See `KafkaClient::new_secure` for more info.
-    #[cfg(feature = "security")]
-    pub fn with_security(mut self, sec: SecurityConfig) -> Builder {
-        self.security_config = Some(sec);
-        self
-    }
-
-    /// Specifies the offset to use when none was committed for the
-    /// underlying group yet or the consumer has no group configured.
-    ///
-    /// Running the underlying group for the first time against a
-    /// topic or running the consumer without a group results in the
-    /// question where to start reading from the topic, since 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 "fallback offset" here corresponds to `time` in
-    /// `KafkaClient::fetch_offsets`.
-    pub fn with_fallback_offset(mut self, fallback_offset: FetchOffset) -> Builder {
-        self.fallback_offset = fallback_offset;
-        self
-    }
-
-    /// See `KafkaClient::set_fetch_max_wait_time`
-    pub fn with_fetch_max_wait_time(mut self, max_wait_time: i32) -> Builder {
-        self.fetch_max_wait_time = max_wait_time;
-        self
-    }
-
-    /// See `KafkaClient::set_fetch_min_bytes`
-    pub fn with_fetch_min_bytes(mut self, min_bytes: i32) -> Builder {
-        self.fetch_min_bytes = min_bytes;
-        self
-    }
-
-    /// See `KafkaClient::set_fetch_max_bytes_per_partition`
-    pub fn with_fetch_max_bytes_per_partition(mut self, max_bytes_per_partition: i32) -> Builder {
-        self.fetch_max_bytes_per_partition = max_bytes_per_partition;
-        self
-    }
-
-    /// See `KafkaClient::set_fetch_crc_validation`
-    pub fn with_fetch_crc_validation(mut self, validate_crc: bool) -> Builder {
-        self.fetch_crc_validation = validate_crc;
-        self
-    }
-
-    /// See `KafkaClient::set_group_offset_storage`
-    pub fn with_offset_storage(mut self, storage: GroupOffsetStorage) -> Builder {
-        self.group_offset_storage = storage;
-        self
-    }
-
-    /// Specifies the upper bound of data bytes to allow fetching from
-    /// a kafka partition when retrying a fetch request due to a too
-    /// big message in the partition.
-    ///
-    /// By default, this consumer will fetch up to
-    /// `KafkaClient::fetch_max_bytes_per_partition` data from each
-    /// partition.  However, when it discovers that there are messages
-    /// in an underlying partition which could not be delivered, the
-    /// request to that partition might be retried a few times with an
-    /// increased `fetch_max_bytes_per_partition`.  The value
-    /// specified here defines a limit to this increment.
-    ///
-    /// A value smaller than the
-    /// `KafkaClient::fetch_max_bytes_per_partition`, e.g. zero, will
-    /// disable the retry feature of this consumer.  The default value
-    /// for this setting is `DEFAULT_RETRY_MAX_BYTES_LIMIT`.
-    ///
-    /// Note: if the consumed topic partitions are known to host large
-    /// messages it is much more efficient to set
-    /// `KafkaClient::fetch_max_bytes_per_partition` appropriately
-    /// instead of relying on the limit specified here.  This limit is
-    /// just an upper bound for already additional retry requests.
-    pub fn with_retry_max_bytes_limit(mut self, limit: i32) -> Builder {
-        self.retry_max_bytes_limit = limit;
-        self
-    }
-
-    #[cfg(not(feature = "security"))]
-    fn new_kafka_client(hosts: Vec<String>, _: Option<SecurityConfig>) -> KafkaClient {
-        KafkaClient::new(hosts)
-    }
-
-    #[cfg(feature = "security")]
-    fn new_kafka_client(hosts: Vec<String>, security: Option<SecurityConfig>) -> KafkaClient {
-        if let Some(security) = security {
-            KafkaClient::new_secure(hosts, security)
-        } else {
-            KafkaClient::new(hosts)
-        }
-    }
-
-    /// Finally creates/builds a new consumer based on the so far
-    /// supplied settings.
-    pub fn create(self) -> Result<Consumer> {
-        let mut client = match self.client {
-            Some(client) => client,
-            None => {
-                let mut client = Self::new_kafka_client(self.hosts, self.security_config);
-                try!(client.load_metadata_all());
-                client
-            }
-        };
-        client.set_fetch_max_wait_time(self.fetch_max_wait_time);
-        client.set_fetch_min_bytes(self.fetch_min_bytes);
-        client.set_fetch_max_bytes_per_partition(self.fetch_max_bytes_per_partition);
-        client.set_group_offset_storage(self.group_offset_storage);
-
-        let config = Config {
-            group: self.group,
-            topic: self.topic,
-            partitions: self.partitions,
-            fallback_offset: self.fallback_offset,
-            retry_max_bytes_limit: self.retry_max_bytes_limit,
-        };
-
-        let state = try!(State::new(&mut client, &config));
-
-        debug!("initialized: (topic: {} / group: {} / state: {:?})",
-               config.topic, config.group, state);
-        Ok(Consumer{ client: client, state: state, config: config })
-    }
-}
diff --git a/src/consumer/assignment.rs b/src/consumer/assignment.rs
new file mode 100644
index 000000000..ecff054b5
--- /dev/null
+++ b/src/consumer/assignment.rs
@@ -0,0 +1,71 @@
+//! Encapsulates topic partition assignments to a consumer. Once
+//! constructed these assignments are not modified.
+
+use std::collections::HashMap;
+use std::ops::Index;
+use std::u32;
+
+/// A read-only configuration for `Consumer`.
+#[derive(Debug)]
+pub struct Assignment {
+    /// ~ name of the topic to consume
+    topic: String,
+    /// ~ list of partitions to consumer, empty if consuming all
+    /// available partitions
+    /// ~ kept in ascending order
+    /// ~ no duplicates
+    partitions: Vec<i32>, 
+}
+
+impl Assignment {
+    pub fn topic(&self) -> &str {
+        &self.topic
+    }
+
+    pub fn partitions(&self) -> &[i32] {
+        &self.partitions
+    }
+}
+
+/// A "pointer" to an assignment stored in `Config`.
+#[derive(Debug, PartialEq, Eq, Hash, Copy, Clone)]
+pub struct AssignmentRef(u32);
+
+/// A set of assignments.
+#[derive(Debug)]
+pub struct Assignments(Vec<Assignment>);
+
+impl Assignments {
+    pub fn as_slice(&self) -> &[Assignment] {
+        &self.0
+    }
+
+    pub fn topic_ref(&self, topic: &str) -> Option<AssignmentRef> {
+        self.0.binary_search_by(|x| x.topic.as_str().cmp(topic))
+            .ok()
+            .map(|i| AssignmentRef(i as u32))
+    }
+}
+
+impl Index<AssignmentRef> for Assignments {
+    type Output = Assignment;
+    fn index(&self, index: AssignmentRef) -> &Self::Output {
+        &self.0[index.0 as usize]
+    }
+}
+
+pub fn from_map(src: HashMap<String, Vec<i32>>) -> Assignments {
+    let mut xs = Vec::with_capacity(src.len());
+    for (topic, mut partitions) in src {
+        partitions.sort();
+        partitions.dedup();
+        xs.push(Assignment {
+            topic: topic,
+            partitions: partitions,
+        });
+    }
+    // ~ sort by topic such has we can apply binary search by that
+    // attribute later
+    xs.sort_by(|a, b| a.topic.cmp(&b.topic));
+    Assignments(xs)
+}
diff --git a/src/consumer/builder.rs b/src/consumer/builder.rs
new file mode 100644
index 000000000..7bdb88e36
--- /dev/null
+++ b/src/consumer/builder.rs
@@ -0,0 +1,231 @@
+use std::collections::HashMap;
+
+use client::{self, KafkaClient, FetchOffset, GroupOffsetStorage};
+use error::{Error, Result};
+
+use super::{Consumer, DEFAULT_FALLBACK_OFFSET, DEFAULT_RETRY_MAX_BYTES_LIMIT};
+use super::config::Config;
+use super::state::State;
+use super::assignment;
+
+#[cfg(feature = "security")]
+use client::SecurityConfig;
+
+#[cfg(not(feature = "security"))]
+type SecurityConfig = ();
+
+/// A Kafka Consumer builder easing the process of setting up various
+/// configuration settings.
+#[derive(Debug)]
+pub struct Builder {
+    client: Option<KafkaClient>,
+    hosts: Vec<String>,
+    group: String,
+    assignments: HashMap<String, Vec<i32>>,
+    fallback_offset: FetchOffset,
+    fetch_max_wait_time: i32,
+    fetch_min_bytes: i32,
+    fetch_max_bytes_per_partition: i32,
+    retry_max_bytes_limit: i32,
+    fetch_crc_validation: bool,
+    security_config: Option<SecurityConfig>,
+    group_offset_storage: GroupOffsetStorage,
+}
+
+// ~ public only to be shared inside the kafka crate; not supposed to
+// be published outside the crate itself
+pub fn new(client: Option<KafkaClient>, hosts: Vec<String>) -> Builder {
+    let mut b = Builder {
+        client: client,
+        hosts: hosts,
+        fetch_max_wait_time: client::DEFAULT_FETCH_MAX_WAIT_TIME,
+        fetch_min_bytes: client::DEFAULT_FETCH_MIN_BYTES,
+        fetch_max_bytes_per_partition: client::DEFAULT_FETCH_MAX_BYTES_PER_PARTITION,
+        fetch_crc_validation: client::DEFAULT_FETCH_CRC_VALIDATION,
+        retry_max_bytes_limit: DEFAULT_RETRY_MAX_BYTES_LIMIT,
+        group: String::new(),
+        assignments: HashMap::new(),
+        fallback_offset: DEFAULT_FALLBACK_OFFSET,
+        security_config: None,
+        group_offset_storage: client::DEFAULT_GROUP_OFFSET_STORAGE,
+    };
+    if let Some(ref c) = b.client {
+        b.fetch_max_wait_time = c.fetch_max_wait_time();
+        b.fetch_min_bytes = c.fetch_min_bytes();
+        b.fetch_max_bytes_per_partition = c.fetch_max_bytes_per_partition();
+        b.fetch_crc_validation = c.fetch_crc_validation();
+        b.group_offset_storage = c.group_offset_storage();
+    }
+    b
+}
+
+impl Builder {
+    /// Specifies the group on whose behalf to maintain consumed
+    /// message offsets.
+    ///
+    /// The group is allowed to be the empty string, in which case the
+    /// resulting consumer will be group-less.
+    pub fn with_group(mut self, group: String) -> Builder {
+        self.group = group;
+        self
+    }
+
+    /// Specifies a topic to consume. All of the available partitions
+    /// of the identified topic will be consumed unless overriden
+    /// later using `with_topic_partitions`.
+    ///
+    /// This method may be called multiple times to assign the
+    /// consumer multiple topics.
+    ///
+    /// This method or `with_topic_partitions` must be called at least
+    /// once, to assign a topic to the consumer.
+    pub fn with_topic(mut self, topic: String) -> Builder {
+        self.assignments.insert(topic, Vec::new());
+        self
+    }
+
+    /// Explicetely specifies topic partitions to consume. Only the
+    /// specified partitions for the identified topic will be consumed
+    /// unless overriden later using `with_topic`.
+    ///
+    /// This method may be called multiple times to subscribe to
+    /// multiple topics.
+    ///
+    /// This method or `with_topic` must be called at least once, to
+    /// assign a topic to the consumer.
+    pub fn with_topic_partitions(mut self, topic: String, partitions: &[i32]) -> Builder {
+        self.assignments.insert(topic, partitions.to_vec());
+        self
+    }
+
+    /// Specifies the security config to use.
+    /// See `KafkaClient::new_secure` for more info.
+    #[cfg(feature = "security")]
+    pub fn with_security(mut self, sec: SecurityConfig) -> Builder {
+        self.security_config = Some(sec);
+        self
+    }
+
+    /// Specifies the offset to use when none was committed for the
+    /// underlying group yet or the consumer has no group configured.
+    ///
+    /// Running the underlying group for the first time against a
+    /// topic or running the consumer without a group results in the
+    /// question where to start reading from the topic, since 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 "fallback offset" here corresponds to `time` in
+    /// `KafkaClient::fetch_offsets`.
+    pub fn with_fallback_offset(mut self, fallback_offset: FetchOffset) -> Builder {
+        self.fallback_offset = fallback_offset;
+        self
+    }
+
+    /// See `KafkaClient::set_fetch_max_wait_time`
+    pub fn with_fetch_max_wait_time(mut self, max_wait_time: i32) -> Builder {
+        self.fetch_max_wait_time = max_wait_time;
+        self
+    }
+
+    /// See `KafkaClient::set_fetch_min_bytes`
+    pub fn with_fetch_min_bytes(mut self, min_bytes: i32) -> Builder {
+        self.fetch_min_bytes = min_bytes;
+        self
+    }
+
+    /// See `KafkaClient::set_fetch_max_bytes_per_partition`
+    pub fn with_fetch_max_bytes_per_partition(mut self, max_bytes_per_partition: i32) -> Builder {
+        self.fetch_max_bytes_per_partition = max_bytes_per_partition;
+        self
+    }
+
+    /// See `KafkaClient::set_fetch_crc_validation`
+    pub fn with_fetch_crc_validation(mut self, validate_crc: bool) -> Builder {
+        self.fetch_crc_validation = validate_crc;
+        self
+    }
+
+    /// See `KafkaClient::set_group_offset_storage`
+    pub fn with_offset_storage(mut self, storage: GroupOffsetStorage) -> Builder {
+        self.group_offset_storage = storage;
+        self
+    }
+
+    /// Specifies the upper bound of data bytes to allow fetching from
+    /// a kafka partition when retrying a fetch request due to a too
+    /// big message in the partition.
+    ///
+    /// By default, this consumer will fetch up to
+    /// `KafkaClient::fetch_max_bytes_per_partition` data from each
+    /// partition.  However, when it discovers that there are messages
+    /// in an underlying partition which could not be delivered, the
+    /// request to that partition might be retried a few times with an
+    /// increased `fetch_max_bytes_per_partition`.  The value
+    /// specified here defines a limit to this increment.
+    ///
+    /// A value smaller than the
+    /// `KafkaClient::fetch_max_bytes_per_partition`, e.g. zero, will
+    /// disable the retry feature of this consumer.  The default value
+    /// for this setting is `DEFAULT_RETRY_MAX_BYTES_LIMIT`.
+    ///
+    /// Note: if the consumed topic partitions are known to host large
+    /// messages it is much more efficient to set
+    /// `KafkaClient::fetch_max_bytes_per_partition` appropriately
+    /// instead of relying on the limit specified here.  This limit is
+    /// just an upper bound for already additional retry requests.
+    pub fn with_retry_max_bytes_limit(mut self, limit: i32) -> Builder {
+        self.retry_max_bytes_limit = limit;
+        self
+    }
+
+    #[cfg(not(feature = "security"))]
+    fn new_kafka_client(hosts: Vec<String>, _: Option<SecurityConfig>) -> KafkaClient {
+        KafkaClient::new(hosts)
+    }
+
+    #[cfg(feature = "security")]
+    fn new_kafka_client(hosts: Vec<String>, security: Option<SecurityConfig>) -> KafkaClient {
+        if let Some(security) = security {
+            KafkaClient::new_secure(hosts, security)
+        } else {
+            KafkaClient::new(hosts)
+        }
+    }
+
+    /// Finally creates/builds a new consumer based on the so far
+    /// supplied settings.
+    ///
+    /// Fails with the `NoTopicsAssigned` error, if neither
+    /// `with_topic` nor `with_topic_partitions` have been called to
+    /// assign at least one topic for consumption.
+    pub fn create(self) -> Result<Consumer> {
+        // ~ fail immediately if there's no topic to be consumed
+        if self.assignments.is_empty() {
+            return Err(Error::NoTopicsAssigned);
+        }
+
+        let mut client = match self.client {
+            Some(client) => client,
+            None => {
+                let mut client = Self::new_kafka_client(self.hosts, self.security_config);
+                try!(client.load_metadata_all());
+                client
+            }
+        };
+        client.set_fetch_max_wait_time(self.fetch_max_wait_time);
+        client.set_fetch_min_bytes(self.fetch_min_bytes);
+        client.set_fetch_max_bytes_per_partition(self.fetch_max_bytes_per_partition);
+        client.set_group_offset_storage(self.group_offset_storage);
+
+        let config = Config {
+            group: self.group,
+            fallback_offset: self.fallback_offset,
+            retry_max_bytes_limit: self.retry_max_bytes_limit
+        };
+        let state = try!(State::new(&mut client, &config, assignment::from_map(self.assignments)));
+        debug!("initialized: Consumer {{ config: {:?}, state: {:?} }}", config, state);
+        Ok(Consumer{ client: client, state: state, config: config })
+    }
+}
diff --git a/src/consumer/config.rs b/src/consumer/config.rs
new file mode 100644
index 000000000..bc9618173
--- /dev/null
+++ b/src/consumer/config.rs
@@ -0,0 +1,8 @@
+use client::FetchOffset;
+
+#[derive(Debug)]
+pub struct Config {
+    pub group: String,
+    pub fallback_offset: FetchOffset,
+    pub retry_max_bytes_limit: i32,
+}
diff --git a/src/consumer/mod.rs b/src/consumer/mod.rs
new file mode 100644
index 000000000..177d1f127
--- /dev/null
+++ b/src/consumer/mod.rs
@@ -0,0 +1,473 @@
+//! Kafka Consumer - A higher-level API for consuming kafka topics.
+//!
+//! A consumer for Kafka topics on behalf of a specified group
+//! providing help in offset management.  The consumer requires at
+//! least one topic for consumption and allows consuming multiple
+//! topics at the same time. Further, clients can restrict the
+//! consumer to only specific topic partitions as demonstrated in the
+//! following example.
+//!
+//! # Example
+//! ```no_run
+//! use kafka::consumer::{Consumer, FetchOffset, GroupOffsetStorage};
+//!
+//! let mut consumer =
+//!    Consumer::from_hosts(vec!("localhost:9092".to_owned()))
+//!       .with_topic_partitions("my-topic".to_owned(), &[0, 1])
+//!       .with_fallback_offset(FetchOffset::Earliest)
+//!       .with_group("my-group".to_owned())
+//!       .with_offset_storage(GroupOffsetStorage::Kafka)
+//!       .create()
+//!       .unwrap();
+//! loop {
+//!   for ms in consumer.poll().unwrap().iter() {
+//!     for m in ms.messages() {
+//!       println!("{:?}", m);
+//!     }
+//!     consumer.consume_messageset(ms);
+//!   }
+//!   consumer.commit_consumed().unwrap();
+//! }
+//! ```
+//!
+//! Please refer to the documentation of the individual "with" methods
+//! used to set up the consumer. These contain further information or
+//! links to such.
+//!
+//! A call to `.poll()` on the consumer 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 topics. Individual messages are embedded in the
+//! retrieved messagesets and can be processed using the `messages()`
+//! iterator.  Due to this embedding, individual messsages's lifetime
+//! is bound to the `MessageSet` they are part of. Typically, client
+//! code access the raw data/bytes, parses it into custom data types
+//! and passes that for further processing within the application.
+//! Altough unconvenient, this helps in reducing the number of
+//! allocations within the pipeline of processing incoming messages.
+//!
+//! If the consumer is configured for a non-empty group, it 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 even 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.
+//!
+//! The configuration of a group is optional.  If the consumer has no
+//! group configured, it will behave as if it had one, only that
+//! commiting consumed message offsets resolves into a void operation.
+
+use std::collections::hash_map::Entry;
+use std::slice;
+
+use client::{KafkaClient, FetchPartition, CommitOffset};
+use error::{Error, KafkaCode, Result};
+use client::fetch;
+
+// public re-exports
+pub use client::fetch::Message;
+pub use client::FetchOffset;
+pub use client::GroupOffsetStorage;
+pub use self::builder::Builder;
+
+mod assignment;
+mod config;
+mod state;
+mod builder;
+
+/// The default value for `Builder::with_retry_max_bytes_limit`.
+pub const DEFAULT_RETRY_MAX_BYTES_LIMIT: i32 = 0;
+
+/// The default value for `Builder::with_fallback_offset`.
+pub const DEFAULT_FALLBACK_OFFSET: FetchOffset = FetchOffset::Latest;
+
+/// The Kafka Consumer
+///
+/// See module level documentation.
+pub struct Consumer {
+    client: KafkaClient,
+    state: state::State,
+    config: config::Config,
+}
+
+// XXX 1) Allow returning to a previous offset (aka seeking)
+// XXX 2) Issue IO in a separate (background) thread and pre-fetch messagesets
+
+impl Consumer {
+
+    /// Starts building a consumer using the given kafka client.
+    pub fn from_client(client: KafkaClient) -> Builder {
+        builder::new(Some(client), Vec::new())
+    }
+
+    /// Starts building a consumer bootstraping internally a new kafka
+    /// client from the given kafka hosts.
+    pub fn from_hosts(hosts: Vec<String>) -> Builder {
+        builder::new(None, hosts)
+    }
+
+    /// Destroys this consumer returning back the underlying kafka client.
+    pub fn client(self) -> KafkaClient {
+        self.client
+    }
+
+    /// Polls for the next available message data.
+    pub fn poll(&mut self) -> Result<MessageSets> {
+        let (n, resps) = self.fetch_messages();
+        self.process_fetch_responses(n, try!(resps))
+    }
+
+    /// Determines whether this consumer is set up to consume only a
+    /// single topic partition.
+    fn single_partition_consumer(&self) -> bool {
+        self.state.fetch_offsets.len() == 1
+    }
+
+    /// Retrieves the group on which behalf this consumer is acting.
+    /// The empty group name specifies a group-less consumer.
+    pub fn group(&self) -> &str {
+        &self.config.group
+    }
+
+    // ~ returns (number partitions queried, fecth responses)
+    fn fetch_messages(&mut self) -> (u32, Result<Vec<fetch::Response>>) {
+        // ~ if there's a retry partition ... fetch messages just for
+        // that one. Otherwise try to fetch messages for all assigned
+        // partitions.
+        match self.state.retry_partitions.pop_front() {
+            Some(tp) => {
+                let s = match self.state.fetch_offsets.get(&tp) {
+                    Some(fstate) => fstate,
+                    None => return (1, Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition))),
+                };
+                let topic = self.state.topic_name(tp.topic_ref);
+                debug!("fetching messages: (fetch-offset: {{\"{}:{}\": {:?}}})",
+                       topic, tp.partition, s);
+                (1, self.client.fetch_messages_for_partition(
+                    &FetchPartition::new(topic, tp.partition, s.offset)
+                        .with_max_bytes(s.max_bytes)))
+            }
+            None => {
+                let client = &mut self.client;
+                let state = &self.state;
+                debug!("fetching messages: (fetch-offsets: {:?})", state.fetch_offsets_debug());
+                let reqs = state.fetch_offsets.iter()
+                    .map(|(tp, s)| {
+                        let topic = state.topic_name(tp.topic_ref);
+                        FetchPartition::new(topic, tp.partition, s.offset)
+                            .with_max_bytes(s.max_bytes)
+                    });
+                (state.fetch_offsets.len() as u32, client.fetch_messages(reqs))
+            }
+        }
+    }
+
+    // ~ post process a data retrieved through fetch_messages before
+    // handing them out to client code
+    //   - update the fetch state for the next fetch cycle
+    // ~ num_partitions_queried: the original number of partitions requested/queried for
+    //   the responses
+    // ~ resps: the responses to post process
+    fn process_fetch_responses(&mut self, num_partitions_queried: u32, resps: Vec<fetch::Response>)
+                               -> Result<MessageSets>
+    {
+        let single_partition_consumer = self.single_partition_consumer();
+        let mut empty = true;
+        let mut retry_partitions = &mut self.state.retry_partitions;
+
+        for resp in &resps {
+            for t in resp.topics() {
+                let topic_ref = self.state.assignments.topic_ref(t.topic())
+                    .expect("unknown topic in response");
+
+                for p in t.partitions() {
+                    let tp = state::TopicPartition {
+                        topic_ref: topic_ref,
+                        partition: p.partition(),
+                    };
+
+                    // ~ for now, as soon as a partition has an error
+                    // we fail to prevent client programs from not
+                    // noticing.  however, in future we don't need to
+                    // fail immediately, we can try to recover from
+                    // certain errors and retry the fetch operation
+                    // transparently for the caller.
+                    let data = match p.data() {
+                        // XXX need to prevent updating fetch_offsets in case we're gonna fail here
+                        &Err(ref e) => return Err(e.clone()),
+                        &Ok(ref data) => data,
+                    };
+
+                    let mut fetch_state = self.state.fetch_offsets
+                            .get_mut(&tp)
+                            .expect("non-requested partition");
+                    // ~ book keeping
+                    if let Some(last_msg) = data.messages().last() {
+                        fetch_state.offset = last_msg.offset + 1;
+                        empty = false;
+
+                        // ~ reset the max_bytes again to its usual
+                        // value if we had a retry request and finally
+                        // got some data
+                        if fetch_state.max_bytes != self.client.fetch_max_bytes_per_partition() {
+                            let prev_max_bytes = fetch_state.max_bytes;
+                            fetch_state.max_bytes = self.client.fetch_max_bytes_per_partition();
+                            debug!("reset max_bytes for {}:{} from {} to {}",
+                                   t.topic(), tp.partition, prev_max_bytes, fetch_state.max_bytes);
+                        }
+                    } else {
+                        debug!("no data received for {}:{} (max_bytes: {} / fetch_offset: {} / highwatermark_offset: {})",
+                               t.topic(), tp.partition, fetch_state.max_bytes, fetch_state.offset, data.highwatermark_offset());
+
+                        // ~ when a partition is empty but has a
+                        // highwatermark-offset equal to or greater
+                        // than the one we tried to fetch ... we'll
+                        // try to increase the max-fetch-size in the
+                        // next fetch request
+                        if fetch_state.offset < data.highwatermark_offset() {
+                            if fetch_state.max_bytes < self.config.retry_max_bytes_limit {
+                                // ~ try to double the max_bytes
+                                let prev_max_bytes = fetch_state.max_bytes;
+                                let incr_max_bytes = prev_max_bytes + prev_max_bytes;
+                                if incr_max_bytes > self.config.retry_max_bytes_limit {
+                                    fetch_state.max_bytes = self.config.retry_max_bytes_limit;
+                                } else {
+                                    fetch_state.max_bytes = incr_max_bytes;
+                                }
+                                debug!("increased max_bytes for {}:{} from {} to {}",
+                                       t.topic(), tp.partition, prev_max_bytes, fetch_state.max_bytes);
+                            } else if num_partitions_queried == 1 {
+                                // ~ this was a single partition
+                                // request and we didn't get anything
+                                // and we won't be increasing the max
+                                // fetch size ... this is will fail
+                                // forever ... signal the problem to
+                                // the user
+                                return Err(Error::Kafka(KafkaCode::MessageSizeTooLarge));
+                            }
+                            // ~ if this consumer is subscribed to one
+                            // partition only, there's no need to push
+                            // the partition to the 'retry_partitions'
+                            // (this is just a small optimization)
+                            if !single_partition_consumer {
+                                // ~ mark this partition for a retry on its own
+                                debug!("rescheduled for retry: {}:{}", t.topic(), tp.partition);
+                                retry_partitions.push_back(tp)
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        // XXX in future, issue one more fetch_messages request in the
+        // background such that the next time the client polls that
+        // request's response will likely be already ready for
+        // consumption
+
+        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.
+    pub fn last_consumed_message(&self, topic: &str, partition: i32) -> Option<i64> {
+        self.state.topic_ref(topic)
+            .and_then(|tref| self.state.consumed_offsets.get(
+                &state::TopicPartition { topic_ref: tref, partition: partition }))
+            .map(|co| co.offset)
+    }
+
+    /// Marks the message at the specified offset in the specified
+    /// topic 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.
+    ///
+    /// Results in an error if the specified topic partition is not
+    /// being consumed by this consumer.
+    pub fn consume_message(&mut self, topic: &str, partition: i32, offset: i64) -> Result<()> {
+        let topic_ref = match self.state.topic_ref(topic) {
+            None => return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)),
+            Some(topic_ref) => topic_ref,
+        };
+        let tp = state::TopicPartition {topic_ref: topic_ref, partition: partition };
+        match self.state.consumed_offsets.entry(tp) {
+            Entry::Vacant(v) => {
+                v.insert(state::ConsumedOffset { offset: offset, dirty: true });
+            }
+            Entry::Occupied(mut v) => {
+                let o = v.get_mut();
+                if offset > o.offset {
+                    o.offset = offset;
+                    o.dirty = true;
+                }
+            }
+        }
+        Ok(())
+    }
+
+    /// 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>) -> Result<()> {
+        if !msgs.messages.is_empty() {
+            self.consume_message(msgs.topic, msgs.partition, msgs.messages.last().unwrap().offset)
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Persists the so-far "marked as consumed" messages (on behalf
+    /// of this consumer's group for the underlying topic - if any.)
+    ///
+    /// See also `Consumer::consume_message` and
+    /// `Consumer::consume_messetset`.
+    pub fn commit_consumed(&mut self) -> Result<()> {
+        if self.config.group.is_empty() {
+            debug!("commit_consumed: ignoring commit request since no group defined");
+            return Ok(());
+        }
+        debug!("commit_consumed: commiting dirty-only consumer offsets (group: {} / offsets: {:?}",
+               self.config.group, self.state.consumed_offsets_debug());
+        let (client, state) = (&mut self.client, &mut self.state);
+        try!(client.commit_offsets(
+            &self.config.group,
+            state.consumed_offsets.iter()
+                .filter(|&(_, o)| o.dirty)
+                .map(|(tp, o)| {
+                    let topic = state.topic_name(tp.topic_ref);
+                    CommitOffset::new(topic, tp.partition, o.offset)
+                })));
+        for (_, co) in &mut state.consumed_offsets {
+            if co.dirty {
+                co.dirty = false;
+            }
+        }
+        Ok(())
+    }
+}
+
+// --------------------------------------------------------------------
+
+/// 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<fetch::Response>,
+
+    /// 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,
+        }
+    }
+}
+
+/// 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
+    }
+
+    #[inline]
+    pub fn partition(&self) -> i32 {
+        self.partition
+    }
+
+    #[inline]
+    pub fn messages(&self) -> &'a [Message<'a>] {
+        self.messages
+    }
+}
+
+/// An iterator over the consumed topic partition message sets.
+pub struct MessageSetsIter<'a> {
+    responses: slice::Iter<'a, fetch::Response>,
+    topics: Option<slice::Iter<'a, fetch::Topic<'a>>>,
+    curr_topic: &'a str,
+    partitions: Option<slice::Iter<'a, fetch::Partition<'a>>>,
+}
+
+impl<'a> Iterator for MessageSetsIter<'a> {
+    type Item = MessageSet<'a>;
+
+    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/consumer/state.rs b/src/consumer/state.rs
new file mode 100644
index 000000000..569c9c1fd
--- /dev/null
+++ b/src/consumer/state.rs
@@ -0,0 +1,338 @@
+use std::collections::HashMap;
+use std::collections::VecDeque;
+use std::fmt;
+use std::hash::BuildHasherDefault;
+
+use fnv::FnvHasher;
+
+use client::{KafkaClient, FetchGroupOffset, FetchOffset};
+use client::metadata::Topics;
+use error::{Error, Result, KafkaCode};
+
+use super::assignment::{Assignment, AssignmentRef, Assignments};
+use super::config::Config;
+
+pub type PartitionHasher = BuildHasherDefault<FnvHasher>;
+
+// The "fetch state" for a particular topci partition.
+#[derive(Debug)]
+pub struct FetchState {
+    /// ~ specifies the offset which to fetch from
+    pub offset: i64,
+    /// ~ specifies the max_bytes to be fetched
+    pub max_bytes: i32,
+}
+
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct TopicPartition {
+    /// ~ indirect reference to the topic through config.topic(..)
+    pub topic_ref: AssignmentRef,
+    /// ~ the partition to retry
+    pub partition: i32,
+}
+
+#[derive(Debug)]
+pub struct ConsumedOffset {
+    /// ~ the consumed offset
+    pub offset: i64,
+    /// ~ true if the consumed offset is chnaged but not committed to
+    /// kafka yet
+    pub dirty: bool,
+}
+
+pub struct State {
+    /// Contains the topic partitions the consumer is assigned to
+    /// consume; this is a _read-only_ data structure
+    pub assignments: Assignments,
+
+    /// Contains the information relevant for the next fetch operation
+    /// on the corresponding partitions
+    pub fetch_offsets: HashMap<TopicPartition, FetchState, PartitionHasher>,
+
+    /// Specifies partitions to be fetched on their own in the next
+    /// poll request.
+    pub retry_partitions: VecDeque<TopicPartition>,
+
+    /// Contains the offsets of messages marked as "consumed" (to be
+    /// committed)
+    pub consumed_offsets: HashMap<TopicPartition, ConsumedOffset, PartitionHasher>,
+}
+
+impl<'a> fmt::Debug for State {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "State {{ assignments: {:?}, \
+                   fetch_offsets: {:?}, \
+                   retry_partitions: {:?}, \
+                   consumed_offsets: {:?} }}",
+               self.assignments,
+               self.fetch_offsets_debug(),
+               TopicPartitionsDebug { state: self, tps: &self.retry_partitions },
+               self.consumed_offsets_debug())
+    }
+}
+
+impl State {
+    pub fn new(client: &mut KafkaClient, config: &Config, assignments: Assignments)
+               -> Result<State>
+    {
+        let (consumed_offsets, fetch_offsets) = {
+            let subscriptions = {
+                let xs = assignments.as_slice();
+                let mut subs = Vec::with_capacity(xs.len());
+                for x in xs {
+                    subs.push(try!(determine_partitions(x, client.topics())));
+                }
+                subs
+            };
+            let n = subscriptions.iter().map(|s| s.partitions.len()).fold(0, |acc, n| acc + n);
+            let consumed = try!(load_consumed_offsets(client, &config.group, &assignments, &subscriptions, n));
+            let fetch_next = try!(load_fetch_states(client, config, &assignments, &subscriptions, &consumed, n));
+            (consumed, fetch_next)
+        };
+        Ok(State {
+            assignments: assignments,
+            fetch_offsets: fetch_offsets,
+            retry_partitions: VecDeque::new(),
+            consumed_offsets: consumed_offsets,
+        })
+    }
+
+    pub fn topic_name(&self, assignment: AssignmentRef) -> &str {
+        self.assignments[assignment].topic()
+    }
+
+    pub fn topic_ref(&self, name: &str) -> Option<AssignmentRef> {
+        self.assignments.topic_ref(name)
+    }
+
+    /// Returns a wrapper around `self.fetch_offsets` for nice dumping
+    /// in debug messages
+    pub fn fetch_offsets_debug<'a>(&'a self) -> OffsetsMapDebug<'a, FetchState> {
+        OffsetsMapDebug { state: self, offsets: &self.fetch_offsets }
+    }
+
+    pub fn consumed_offsets_debug<'a>(&'a self) -> OffsetsMapDebug<'a, ConsumedOffset> {
+        OffsetsMapDebug { state: self, offsets: &self.consumed_offsets }
+    }
+}
+
+// Specifies the actual partitions of a topic to be consumed
+struct Subscription<'a> {
+    assignment: &'a Assignment, // the assignment - user configuration
+    partitions: Vec<i32>, // the actual partitions to be consumed
+}
+
+/// Determines the partitions to be consumed according to the
+/// specified topic and requested partitions configuration. Returns an
+/// ordered list of the partition ids to consume.
+fn determine_partitions<'a>(assignment: &'a Assignment, metadata: Topics)
+                            -> Result<Subscription<'a>>
+{
+    let topic = assignment.topic();
+    let req_partitions = assignment.partitions();
+    let avail_partitions = match metadata.partitions(topic) {
+        // ~ fail if the underlying topic is unkonwn to the given client
+        None => {
+            debug!("determine_partitions: no such topic: {} (all metadata: {:?})",
+                   topic, metadata);
+            return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
+        }
+        Some(tp) => tp,
+    };
+    let ps = if req_partitions.is_empty() {
+        // ~ no partitions configured ... use all available
+        let mut ps: Vec<i32> = Vec::with_capacity(avail_partitions.len());
+        for p in avail_partitions {
+            ps.push(p.id());
+        }
+        ps
+    } else {
+        // ~ validate that all partitions we're going to consume are
+        // available
+        let mut ps: Vec<i32> = Vec::with_capacity(req_partitions.len());
+        for &p in req_partitions {
+            match avail_partitions.partition(p) {
+                None => {
+                    debug!("determine_partitions: no such partition: \"{}:{}\" \
+                            (all metadata: {:?})", topic, p, metadata);
+                    return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
+                }
+                Some(_) => ps.push(p),
+            };
+        }
+        ps
+    };
+    Ok(Subscription { assignment: assignment, partitions: ps })
+}
+
+// Fetches the so-far commited/consumed offsets for the configured
+// group/topic/partitions.
+fn load_consumed_offsets(
+    client: &mut KafkaClient, group: &str, assignments: &Assignments,
+    subscriptions: &[Subscription], result_capacity: usize)
+    -> Result<HashMap<TopicPartition, ConsumedOffset, PartitionHasher>>
+{
+    assert!(!subscriptions.is_empty());
+    // ~ pre-allocate the right size
+    let mut offs = HashMap::with_capacity_and_hasher(result_capacity, PartitionHasher::default());
+    // ~ no group, no persisted consumed offsets
+    if group.is_empty() {
+        return Ok(offs);
+    }
+    // ~ otherwise try load them for the group
+    let tpos = try!(client.fetch_group_offsets(
+        group,
+        subscriptions.iter()
+            .flat_map(|s| {
+                let topic = s.assignment.topic();
+                s.partitions.iter().map(move |&p| FetchGroupOffset::new(topic, p))
+            })));
+    for tpo in tpos {
+        match tpo.offset {
+            Ok(o) if o != -1 => {
+                offs.insert(
+                    TopicPartition {
+                        topic_ref: assignments.topic_ref(&tpo.topic).expect("non-assigned topic"),
+                        partition: tpo.partition,
+                    },
+                    ConsumedOffset {
+                        offset: o,
+                        dirty: false,
+                    });
+            }
+            _ => {}
+        }
+    }
+    Ok(offs)
+}
+
+/// Fetches the "next fetch" offsets/states based on the specified
+/// subscriptions and the given consumed offsets.
+fn load_fetch_states(
+    client: &mut KafkaClient, config: &Config,
+    assignments: &Assignments, subscriptions: &[Subscription],
+    consumed_offsets: &HashMap<TopicPartition, ConsumedOffset, PartitionHasher>,
+    result_capacity: usize)
+    -> Result<HashMap<TopicPartition, FetchState, PartitionHasher>>
+{
+    fn load_partition_offsets(client: &mut KafkaClient, topics: &[&str], offset: FetchOffset)
+                              -> Result<HashMap<String, HashMap<i32, i64, PartitionHasher>>>
+    {
+        let toffs = try!(client.fetch_offsets(topics, offset));
+        let mut m = HashMap::with_capacity(toffs.len());
+        for (topic, poffs) in toffs {
+            let mut pidx = HashMap::with_capacity_and_hasher(poffs.len(), PartitionHasher::default());
+            for poff in poffs {
+                pidx.insert(poff.partition, poff.offset.unwrap_or(-1));
+            }
+            m.insert(topic, pidx);
+        }
+        Ok(m)
+    }
+
+    let mut fetch_offsets = HashMap::with_capacity_and_hasher(
+        result_capacity, PartitionHasher::default());
+    let max_bytes = client.fetch_max_bytes_per_partition();
+    let subscription_topics: Vec<_> = subscriptions.iter().map(|s| s.assignment.topic()).collect();
+    if consumed_offsets.is_empty() {
+        // ~ if there are no offsets on behalf of the consumer
+        // group - if any - we can directly use the fallback offsets.
+        let offsets = try!(load_partition_offsets(
+            client, &subscription_topics, config.fallback_offset));
+        for s in subscriptions {
+            let topic_ref = assignments.topic_ref(s.assignment.topic())
+                .expect("unassigned subscription");
+            match offsets.get(s.assignment.topic()) {
+                None => {
+                    debug!("load_fetch_states: failed to load fallback offsets for: {}",
+                           s.assignment.topic());
+                    return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
+                }
+                Some(offsets) => {
+                    for p in &s.partitions {
+                        fetch_offsets.insert(
+                            TopicPartition {
+                                topic_ref: topic_ref,
+                                partition: *p,
+                            },
+                            FetchState {
+                                offset: *offsets.get(p).unwrap_or(&-1),
+                                max_bytes: max_bytes,
+                            });
+                    }
+                }
+            }
+        }
+    } else {
+        // fetch the earliest and latest available offsets
+        let latest = try!(load_partition_offsets(client, &subscription_topics, FetchOffset::Latest));
+        let earliest = try!(load_partition_offsets(client, &subscription_topics, FetchOffset::Earliest));
+        // ~ for each subscribed 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.
+        for s in subscriptions {
+            let topic_ref = assignments.topic_ref(s.assignment.topic())
+                .expect("unassigned subscription");
+            for p in &s.partitions {
+                let l_off = *latest.get(s.assignment.topic()).and_then(|ps| ps.get(p)).unwrap_or(&-1);
+                let e_off = *earliest.get(s.assignment.topic()).and_then(|ps| ps.get(p)).unwrap_or(&-1);
+
+                let tp = TopicPartition { topic_ref: topic_ref, partition: *p };
+                let offset = match consumed_offsets.get(&tp) {
+                    Some(co) if co.offset >= e_off && co.offset < l_off => co.offset + 1,
+                    _ => {
+                        match config.fallback_offset {
+                            FetchOffset::Latest => l_off,
+                            FetchOffset::Earliest => e_off,
+                            _ => {
+                                debug!("cannot determine fetch offset \
+                                        (group: {} / topic: {} / partition: {})",
+                                       &config.group, s.assignment.topic(), p);
+                                return Err(Error::Kafka(KafkaCode::Unknown));
+                            }
+                        }
+                    }
+                };
+                fetch_offsets.insert(tp, FetchState { offset: offset, max_bytes: max_bytes });
+            }
+        }
+    }
+    Ok(fetch_offsets)
+}
+
+pub struct OffsetsMapDebug<'a, T: 'a> {
+    state: &'a State,
+    offsets: &'a HashMap<TopicPartition, T, PartitionHasher>,
+}
+
+impl<'a, T: fmt::Debug + 'a> fmt::Debug for OffsetsMapDebug<'a, T> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        try!(write!(f, "{{"));
+        for (i, (tp, v)) in self.offsets.iter().enumerate() {
+            if i != 0 {
+                try!(write!(f, ", "));
+            }
+            let topic = self.state.topic_name(tp.topic_ref);
+            try!(write!(f, "\"{}:{}\": {:?}", topic, tp.partition, v));
+        }
+        write!(f, "}}")
+    }
+}
+
+struct TopicPartitionsDebug<'a> {
+    state: &'a State,
+    tps: &'a VecDeque<TopicPartition>,
+}
+
+impl<'a> fmt::Debug for TopicPartitionsDebug<'a> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        try!(write!(f, "["));
+        for (i, tp) in self.tps.iter().enumerate() {
+            if i != 0 {
+                try!(write!(f, " ,"));
+            }
+            try!(write!(f, "\"{}:{}\"", self.state.topic_name(tp.topic_ref), tp.partition));
+        }
+        write!(f, "]")
+    }
+}
diff --git a/src/error.rs b/src/error.rs
index 0e6a8b2c9..7e52d06b1 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -45,6 +45,8 @@ pub enum Error {
     StringDecodeError,
     /// Unable to reach any host
     NoHostReachable,
+    /// Unable to set up `Consumer` due to missing topic assignments
+    NoTopicsAssigned,
 }
 
 /// Various errors reported by a remote Kafka server.
@@ -204,6 +206,7 @@ impl Clone for Error {
             &Error::CodecError => Error::CodecError,
             &Error::StringDecodeError => Error::StringDecodeError,
             &Error::NoHostReachable => Error::NoHostReachable,
+            &Error::NoTopicsAssigned => Error::NoTopicsAssigned,
         }
     }
 }
@@ -223,6 +226,7 @@ impl error::Error for Error {
             Error::CodecError => "Encoding/Decoding error",
             Error::StringDecodeError => "String decoding error",
             Error::NoHostReachable => "No host reachable",
+            Error::NoTopicsAssigned => "No topic assigned",
         }
     }
 
@@ -248,7 +252,8 @@ impl fmt::Display for Error {
             Error::UnexpectedEOF => write!(f, "Unexpected EOF"),
             Error::CodecError => write!(f, "Encoding/Decoding Error"),
             Error::StringDecodeError => write!(f, "String decoding error"),
-            Error::NoHostReachable => write!(f, "No Host Reachable"),
+            Error::NoHostReachable => write!(f, "No host reachable"),
+            Error::NoTopicsAssigned => write!(f, "No topic assigned"),
         }
     }
 }


Test patch: