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

Issue:
utils::{TopicPartitionOffset, PartitionOffset} have Result fields
Currently, the `utils::{TopicPartitionOffset, PartitionOffset}` structs' offset fields are `Result`s. In my opinion, it's kind of strange to have a struct with a `Result` in the middle of it for anything but maybe custom `Error` types. It can make error handling on the user-side really inconvenient.

Additionally, it's not entirely clear to me why `TopicPartitionOffset` is needed at all—it seems that it exists only to flatten responses, which leads to [lots of inefficient cloning](https://github.com/spicavigo/kafka-rust/blob/master/src/client/mod.rs#L1340) of data that it already had. I think in an instance like the linked one, it would be more appropriate to return a map which maps topics to `PartitionOffset`s; that way, only a move of one string is necessary, rather than a clone for every single offset response.

What do you think? I tried to start by changing the `Result`s to just their underlying type, but I can see it's a non-trivial refactor, so I thought I'd get your input first.



Discussion / hints:
Many thanks for the investigation!
- I agree that the existence of `TopicPartitionOffset` is sub-optimal. In particular for `KafkaClient::fetch_group_topic_offsets` and `KafkaClient::fetch_group_offsets` a return value of `Map<String, Vec<PartitionOffset>>` as you implied would be desirable. This will definitely avoid the needless cloning and help clients in processing the response more easily. Feel free to go ahead.
- For `KafkaClient::produce_message`, which also returns `TopicPartitionOffsets`, the situation is a bit unclear. As you can see from the [TODO in the code](https://github.com/spicavigo/kafka-rust/blob/master/src/client/mod.rs#L1013) the return value should be changed anyway especially in regards to partitions which failed to acknowledge the data. Feel free to change it to a `Map<String, Vec<PartitionOffset>>` as well (my idea about the TODO is to return a `PartitionOffsetOrDataFailedToBeDelivered` in some future so that the client side can retry the operation for the failed partition.)
- Regarding the `Result` within the structs. Well, the idea is to _not_ access the offsets value (i64) in the struct if kafka reported an error for the partition. This means the error can be only of type `kafka::error::KafkaCode`. I'm fine with changing it to `Result<i64, KafkaCode>` (i guess this will even reduce the size of the struct). The introduction of the `Result` in the two mentioned structs came after mistakenly ignoring the error in an earlier version of kafka-rust in the consumer. So I'm rather voting for keeping it.

> I agree that the existence of `TopicPartitionOffset` is sub-optimal. In particular for `KafkaClient::fetch_group_topic_offsets` and `KafkaClient::fetch_group_offsets` a return value of `Map<String, Vec<PartitionOffset>>` as you implied would be desirable. This will definitely avoid the needless cloning and help clients in processing the response more easily. Feel free to go ahead.

Cool, sounds good.

For the last two bullet points, yeah, it's not immediately clear how to handle the return value there. However, I think at least part of the confusion (at least for me) is that the Kafka broker's `ProduceResponse` contains _both_ the offset and the error code, if any, so it doesn't really make sense to wrap the offset in a `Result` there anyway, since either way, the acknowledgement message will contain some offset.

For _fetching_ group offsets, I think it makes sense to have them in a `Result` because either the offset exists, or returns -1, or some error happened. I think there are two ways to handle the errors when fetching offsets:
1. Wrap all partition/offset data into one top-level `Result`, with all offset data available if there were no errors. [This is currently what fetching does](https://github.com/spicavigo/kafka-rust/blob/master/src/client/mod.rs#L1365), so if the top-level `Result` returned by `__fetch_group_offsets` is `Ok`, then so will all the `offset` fields, so having to unwrap them is just unnecessary pain.
2. Handle partition errors individually (probably the more robust choice). In this case, knowing which partition failed to fetch its offsets is useful, so I think it does make sense to either store the partition data with its `Result` for the offset, which is what it's doing now, or just return a top-level `Result` that contains either the partition _and_ offset together, or an `Err` that describes which partition it failed to fetch. (This way would also have the added benefit of being able to use `PartitionOffset` on the producer side as well.) Either way works, but in my opinion, while the latter is more of a pain to implement on the lib side, I think it makes error handling easier for the user-side. However, I defer to your judgment, since this is your lib :)

But right now, the current state of the lib is kind of inconsistent. It has a struct that contains a `Result` field as if to handle individual partition errors, but then doesn't really make use of it. I think if we're just going to quit on the first partition error, then having the `Result` field is just unnecessary. But handling partition errors individually may allow for more robust error handling, and allow the user to use the portion of the offsets that were fetched successfully. What do you think is the best way forward?

Also, just wanted to let you know that I don't often get free time to work on for-fun projects, so I can't promise I'll be able to move forward with this issue in a timely manner, unfortunately :/


Fix patch:
diff --git a/examples/console-producer.rs b/examples/console-producer.rs
index dbaa96649..0db30719f 100644
--- a/examples/console-producer.rs
+++ b/examples/console-producer.rs
@@ -153,11 +153,15 @@ fn produce_impl_inbatches(producer: &mut Producer, src: &mut BufRead, cfg: &Conf
 
 fn send_batch(producer: &mut Producer, batch: &[Record<(), Trimmed>]) -> Result<(), Error> {
     let rs = try!(producer.send_all(batch));
+
     for r in rs {
-        if let Err(e) = r.offset {
-            return Err(From::from(e));
+        for tpc in r.partition_confirms {
+            if let Err(code) = tpc.offset {
+                return Err(Error::Kafka(kafka::error::Error::Kafka(code)));
+            }
         }
     }
+
     Ok(())
 }
 
diff --git a/examples/offset-monitor.rs b/examples/offset-monitor.rs
index cbf4260ab..0605defa1 100644
--- a/examples/offset-monitor.rs
+++ b/examples/offset-monitor.rs
@@ -116,30 +116,26 @@ impl State {
     {
         // ~ get the latest topic offsets
         let latests = try!(client.fetch_topic_offsets(topic, FetchOffset::Latest));
+
         for l in latests {
             let off = self.offsets.get_mut(l.partition as usize)
                 .expect("[topic offset] non-existent partition");
             off.prev_latest = off.curr_latest;
-            off.curr_latest = l.offset.unwrap_or(-1);
+            off.curr_latest = l.offset;
         }
+
         if !group.is_empty() {
             // ~ get the current group offsets
             let groups = try!(client.fetch_group_topic_offsets(group, topic));
             for g in groups {
                 let off = self.offsets.get_mut(g.partition as usize)
                     .expect("[group offset] non-existent partition");
-                match g.offset {
-                    Ok(v) => {
-                        // ~ it's quite likely that we fetched group offsets
-                        // which are a bit ahead of the topic's latest offset
-                        // since we issued the fetch-latest-offset request
-                        // earlier than the request for the group offsets
-                        off.curr_lag = cmp::max(0, off.curr_latest - v - self.lag_decr);
-                    }
-                    Err(_) => {
-                        off.curr_lag = -1;
-                    }
-                }
+
+                // ~ it's quite likely that we fetched group offsets
+                // which are a bit ahead of the topic's latest offset
+                // since we issued the fetch-latest-offset request
+                // earlier than the request for the group offsets
+                off.curr_lag = cmp::max(0, off.curr_latest - g.offset - self.lag_decr);
             }
         }
         Ok(())
diff --git a/examples/topic-metadata.rs b/examples/topic-metadata.rs
index 723dcf5a4..bf56df028 100644
--- a/examples/topic-metadata.rs
+++ b/examples/topic-metadata.rs
@@ -65,30 +65,28 @@ fn dump_metadata(cfg: Config) -> Result<(), String> {
         for (topic, offsets) in offsets {
             topic_width = cmp::max(topic_width, topic.len());
             let mut offs = Vec::with_capacity(offsets.len());
+
             for _ in 0..offsets.len() {
                 offs.push(Offsets::default());
             }
+
             for offset in offsets {
-                match offset.offset {
-                    Ok(o) => offs[offset.partition as usize].latest = o,
-                    Err(e) => debug!("Cannot determine latest offset for {}:{}: {}",
-                                     topic, offset.partition, e),
-                }
+                offs[offset.partition as usize].latest = offset.offset;
             }
+
             m.insert(topic, offs);
         }
+
         offsets = try!(client.fetch_offsets(&topics, FetchOffset::Earliest)
                        .map_err(|e| e.to_string()));
+
         for (topic, offsets) in offsets {
             let mut offs = m.get_mut(&topic).expect("unknown topic");
             for offset in offsets {
-                match offset.offset {
-                    Ok(o) => offs[offset.partition as usize].earliest = o,
-                    Err(e) => debug!("Cannot determine earliest offset for {}:{}: {}",
-                                     topic, offset.partition, e),
-                }
+                offs[offset.partition as usize].earliest = offset.offset;
             }
         }
+
         (topic_width + 2, m)
     };
     // ~ produce the output
diff --git a/rustfmt.toml b/rustfmt.toml
index f1cba768d..dd3ac2566 100644
--- a/rustfmt.toml
+++ b/rustfmt.toml
@@ -1,3 +1,4 @@
 max_width = 100
 ideal_with = 80
 fn_call_width = 80
+write_mode = "overwrite"
diff --git a/src/client/mod.rs b/src/client/mod.rs
index 625149d20..c47d04ffa 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -4,6 +4,8 @@
 //! The entry point into this module is `KafkaClient` obtained by a
 //! call to `KafkaClient::new()`.
 
+use std;
+use std::collections::hash_map;
 use std::collections::hash_map::HashMap;
 use std::io::Cursor;
 use std::iter::Iterator;
@@ -14,7 +16,6 @@ use std::time::{Duration, Instant};
 // pub re-export
 pub use compression::Compression;
 pub use utils::PartitionOffset;
-pub use utils::TopicPartitionOffset;
 
 #[cfg(feature = "security")]
 pub use self::network::SecurityConfig;
@@ -350,6 +351,30 @@ impl<'a> AsRef<FetchPartition<'a>> for FetchPartition<'a> {
     }
 }
 
+/// A confirmation of messages sent back by the Kafka broker
+/// to confirm delivery of producer messages.
+#[derive(Debug)]
+pub struct ProduceConfirm {
+    /// The topic the messages were sent to.
+    pub topic: String,
+
+    /// The list of individual confirmations for each offset and partition.
+    pub partition_confirms: Vec<ProducePartitionConfirm>,
+}
+
+/// A confirmation of messages sent back by the Kafka broker
+/// to confirm delivery of producer messages for a particular topic.
+#[derive(Debug)]
+pub struct ProducePartitionConfirm {
+    /// The offset assigned to the first message in the message set appended
+    /// to this partition, or an error if one occurred.
+    pub offset: std::result::Result<i64, KafkaCode>,
+
+    /// The partition to which the message(s) were appended.
+    pub partition: i32,
+}
+
+
 // --------------------------------------------------------------------
 
 impl KafkaClient {
@@ -825,15 +850,57 @@ impl KafkaClient {
                                                                           now,
                                                                           req));
             for tp in resp.topic_partitions {
-                let e = res.entry(tp.topic).or_insert(vec![]);
-                for p in tp.partitions {
-                    e.push(p.into_offset());
+                let mut entry = res.entry(tp.topic);
+                let mut new_resp_offsets = None;
+                let mut err = None;
+                // Use an explicit scope here to allow insertion into a vacant entry
+                // below
+                {
+                    // Return a &mut to the response we will be collecting into to
+                    // return from this function. If there are some responses we have
+                    // already prepared, keep collecting into that; otherwise, make a
+                    // new collection to return.
+                    let resp_offsets = match entry {
+                        hash_map::Entry::Occupied(ref mut e) => e.get_mut(),
+                        hash_map::Entry::Vacant(_) => {
+                            new_resp_offsets = Some(Vec::new());
+                            new_resp_offsets.as_mut().unwrap()
+                        }
+                    };
+                    for p in tp.partitions {
+                        let partition_offset = match p.into_offset() {
+                            Ok(po) => po,
+                            Err(code) => {
+                                err = Some((p.partition, code));
+                                break;
+                            }
+                        };
+                        resp_offsets.push(partition_offset);
+                    }
+                }
+                if let Some((partition, code)) = err {
+                    let topic = KafkaClient::get_key_from_entry(entry);
+                    return Err(Error::TopicPartitionError(topic, partition, code));
+                }
+                if let hash_map::Entry::Vacant(e) = entry {
+                    // unwrap is ok because if it is Vacant, it would have
+                    // been made into a Some above
+                    e.insert(new_resp_offsets.unwrap());
                 }
             }
         }
+
         Ok(res)
     }
 
+    /// Takes ownership back from the given HashMap Entry.
+    fn get_key_from_entry<'a, K: 'a, V: 'a>(entry: hash_map::Entry<'a, K, V>) -> K {
+        match entry {
+            hash_map::Entry::Occupied(e) => e.remove_entry().0,
+            hash_map::Entry::Vacant(e) => e.into_key(),
+        }
+    }
+
     /// Fetch offset for a single topic.
     ///
     /// # Examples
@@ -1017,7 +1084,7 @@ impl KafkaClient {
                                           acks: RequiredAcks,
                                           ack_timeout: Duration,
                                           messages: I)
-                                          -> Result<Vec<TopicPartitionOffset>>
+                                          -> Result<Vec<ProduceConfirm>>
         where J: AsRef<ProduceMessage<'a, 'b>>,
               I: IntoIterator<Item = J>
     {
@@ -1115,7 +1182,7 @@ impl KafkaClient {
     pub fn fetch_group_offsets<'a, J, I>(&mut self,
                                          group: &str,
                                          partitions: I)
-                                         -> Result<Vec<TopicPartitionOffset>>
+                                         -> Result<HashMap<String, Vec<PartitionOffset>>>
         where J: AsRef<FetchGroupOffset<'a>>,
               I: IntoIterator<Item = J>
     {
@@ -1123,17 +1190,15 @@ impl KafkaClient {
                                                         self.config.offset_fetch_version,
                                                         self.state.next_correlation_id(),
                                                         &self.config.client_id);
-        let mut num = 0;
         for p in partitions {
             let p = p.as_ref();
             if self.state.contains_topic_partition(p.topic, p.partition) {
                 req.add(p.topic, p.partition);
-                num += 1;
             } else {
                 return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
             }
         }
-        __fetch_group_offsets(req, num, &mut self.state, &mut self.conn_pool, &self.config)
+        __fetch_group_offsets(req, &mut self.state, &mut self.conn_pool, &self.config)
     }
 
     /// Fetch offset for all partitions of a particular topic of a consumer group
@@ -1150,22 +1215,24 @@ impl KafkaClient {
     pub fn fetch_group_topic_offsets(&mut self,
                                      group: &str,
                                      topic: &str)
-                                     -> Result<Vec<TopicPartitionOffset>> {
+                                     -> Result<Vec<PartitionOffset>> {
         let mut req = protocol::OffsetFetchRequest::new(group,
                                                         self.config.offset_fetch_version,
                                                         self.state.next_correlation_id(),
                                                         &self.config.client_id);
-        let num;
+
         match self.state.partitions_for(topic) {
             None => return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)),
             Some(tp) => {
-                num = tp.len();
                 for (id, _) in tp {
                     req.add(topic, id);
                 }
             }
         }
-        __fetch_group_offsets(req, num, &mut self.state, &mut self.conn_pool, &self.config)
+
+        Ok(try!(__fetch_group_offsets(req, &mut self.state, &mut self.conn_pool, &self.config))
+            .remove(topic)
+            .unwrap_or_else(Vec::new))
     }
 }
 
@@ -1174,7 +1241,7 @@ impl KafkaClientInternals for KafkaClient {
                                                required_acks: i16,
                                                ack_timeout: i32,
                                                messages: I)
-                                               -> Result<Vec<TopicPartitionOffset>>
+                                               -> Result<Vec<ProduceConfirm>>
         where J: AsRef<ProduceMessage<'a, 'b>>,
               I: IntoIterator<Item = J>
     {
@@ -1312,13 +1379,11 @@ fn __commit_offsets(req: protocol::OffsetCommitRequest,
     }
 }
 
-// ~ 'num_partitions' is merely a hint to pre-allocation inside this method
 fn __fetch_group_offsets(req: protocol::OffsetFetchRequest,
-                         num_partitions: usize,
                          state: &mut state::ClientState,
                          conn_pool: &mut network::Connections,
                          config: &ClientConfig)
-                         -> Result<Vec<TopicPartitionOffset>> {
+                         -> Result<HashMap<String, Vec<PartitionOffset>>> {
     let mut attempt = 1;
     loop {
         let now = Instant::now();
@@ -1333,21 +1398,15 @@ fn __fetch_group_offsets(req: protocol::OffsetFetchRequest,
         };
 
         let mut retry_code = None;
-        let mut v = Vec::with_capacity(num_partitions);
+        let mut topic_map = HashMap::with_capacity(r.topic_partitions.len());
 
         'rproc: for tp in r.topic_partitions {
+            let mut partition_offsets = Vec::with_capacity(tp.partitions.len());
+
             for p in tp.partitions {
-                let mut o = p.get_offsets(tp.topic.clone());
-                match o.offset {
-                    Ok(_) => {
-                        v.push(o);
-                    }
-                    Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition)) => {
-                        // ~ occurs only on protocol v0 when no offset available
-                        // for the group in question; we'll align the behavior
-                        // with protocol v1.
-                        o.offset = Ok(-1);
-                        v.push(o);
+                match p.get_offsets() {
+                    Ok(o) => {
+                        partition_offsets.push(o);
                     }
                     Err(Error::Kafka(e @ KafkaCode::GroupLoadInProgress)) => {
                         retry_code = Some(e);
@@ -1366,7 +1425,10 @@ fn __fetch_group_offsets(req: protocol::OffsetFetchRequest,
                     }
                 }
             }
+
+            topic_map.insert(tp.topic, partition_offsets);
         }
+
         // ~ have we processed the result successfully or shall we
         // retry once more?
         match retry_code {
@@ -1382,7 +1444,7 @@ fn __fetch_group_offsets(req: protocol::OffsetFetchRequest,
                 }
             }
             None => {
-                return Ok(v);
+                return Ok(topic_map);
             }
         }
     }
@@ -1409,7 +1471,7 @@ fn __fetch_messages(conn_pool: &mut network::Connections,
 fn __produce_messages(conn_pool: &mut network::Connections,
                       reqs: HashMap<&str, protocol::ProduceRequest>,
                       no_acks: bool)
-                      -> Result<Vec<TopicPartitionOffset>> {
+                      -> Result<Vec<ProduceConfirm>> {
     let now = Instant::now();
     if no_acks {
         for (host, req) in reqs {
@@ -1417,7 +1479,7 @@ fn __produce_messages(conn_pool: &mut network::Connections,
         }
         Ok(vec![])
     } else {
-        let mut res: Vec<TopicPartitionOffset> = vec![];
+        let mut res: Vec<ProduceConfirm> = vec![];
         for (host, req) in reqs {
             let resp =
                 try!(__send_receive::<_, protocol::ProduceResponse>(conn_pool, &host, now, req));
diff --git a/src/client_internals.rs b/src/client_internals.rs
index 69c90fde9..3e29f9a0b 100644
--- a/src/client_internals.rs
+++ b/src/client_internals.rs
@@ -1,7 +1,8 @@
 //! A crate private module to expose `KafkaClient` internals for use
 //! within this crate but not outside of it.
 
-use client;
+use client::ProduceMessage;
+use producer::ProduceConfirm;
 use error::Result;
 
 pub trait KafkaClientInternals {
@@ -9,7 +10,7 @@ pub trait KafkaClientInternals {
                                                required_acks: i16,
                                                ack_timeout: i32,
                                                messages: I)
-                                               -> Result<Vec<client::TopicPartitionOffset>>
-        where J: AsRef<client::ProduceMessage<'a, 'b>>,
+                                               -> Result<Vec<ProduceConfirm>>
+        where J: AsRef<ProduceMessage<'a, 'b>>,
               I: IntoIterator<Item = J>;
 }
diff --git a/src/consumer/state.rs b/src/consumer/state.rs
index b89ccb0fd..2309f3ef6 100644
--- a/src/consumer/state.rs
+++ b/src/consumer/state.rs
@@ -205,20 +205,19 @@ fn load_consumed_offsets(client: &mut KafkaClient,
                                                            FetchGroupOffset::new(topic, p)
                                                        })
                                                    })));
-    for tpo in tpos {
-        match tpo.offset {
-            Ok(o) if o != -1 => {
+    for (topic, pos) in tpos {
+        for po in pos {
+            if po.offset != -1 {
                 offs.insert(TopicPartition {
-                                topic_ref: assignments.topic_ref(&tpo.topic)
+                                topic_ref: assignments.topic_ref(&topic)
                                     .expect("non-assigned topic"),
-                                partition: tpo.partition,
+                                partition: po.partition,
                             },
                             ConsumedOffset {
-                                offset: o,
+                                offset: po.offset,
                                 dirty: false,
                             });
             }
-            _ => {}
         }
     }
     Ok(offs)
@@ -242,9 +241,11 @@ fn load_fetch_states(client: &mut KafkaClient,
         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));
+                pidx.insert(poff.partition, poff.offset);
             }
+
             m.insert(topic, pidx);
         }
         Ok(m)
diff --git a/src/error.rs b/src/error.rs
index fb10fab6c..f30a1361e 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -8,7 +8,6 @@ use std::fmt;
 #[cfg(feature = "security")]
 use openssl::ssl;
 
-
 /// A type for results generated by this crate's functions where the `Err` type
 /// is hard-wired to `enums::Error`.
 ///
@@ -21,8 +20,15 @@ pub type Result<T> = result::Result<T, Error>;
 pub enum Error {
     /// Input/Output error while communicating with Kafka
     Io(io::Error),
+
     /// An error as reported by a remote Kafka server
     Kafka(KafkaCode),
+
+    /// An error when transmitting a request for a particular topic and partition.
+    /// Contains the topic and partition of the request that failed, and the
+    /// error code as reported by the Kafka server, respectively.
+    TopicPartitionError(String, i32, KafkaCode),
+
     /// An error as reported by OpenSsl
     #[cfg(feature = "security")]
     Ssl(ssl::error::SslError),
@@ -31,22 +37,30 @@ pub enum Error {
     /// server speaking a newer protocol version (than the one this
     /// library supports)
     UnsupportedProtocol,
+
     /// Failure to correctly parse the server response by this library
     /// due to an unsupported compression format of the data
     UnsupportedCompression,
+
     /// Failure to decode a snappy compressed response from Kafka
     #[cfg(feature = "snappy")]
     InvalidInputSnappy,
+
     /// Failure to decode a response due to an insufficient number of bytes available
     UnexpectedEOF,
+
     /// Failure to decode or encode a response or request respectively
     CodecError,
+
     /// Failure to decode a string into a valid utf8 byte sequence
     StringDecodeError,
+
     /// Unable to reach any host
     NoHostReachable,
+
     /// Unable to set up `Consumer` due to missing topic assignments
     NoTopicsAssigned,
+
     /// An invalid user-provided duration
     InvalidDuration,
 }
@@ -191,6 +205,9 @@ impl Clone for Error {
         match self {
             &Error::Io(ref err) => Error::Io(io::Error::new(err.kind(), "Io Error")),
             &Error::Kafka(x) => Error::Kafka(x),
+            &Error::TopicPartitionError(ref topic, partition, error_code) => {
+                Error::TopicPartitionError(topic.clone(), partition, error_code)
+            }
             #[cfg(feature = "security")]
             &Error::Ssl(ref x) => {
                 match x {
@@ -225,6 +242,7 @@ impl error::Error for Error {
         match *self {
             Error::Io(ref err) => error::Error::description(err),
             Error::Kafka(_) => "Kafka Error",
+            Error::TopicPartitionError(_, _, _) => "Error in request for topic and partition",
             #[cfg(feature = "security")]
             Error::Ssl(ref err) => error::Error::description(err),
             Error::UnsupportedProtocol => "Unsupported protocol version",
@@ -253,6 +271,9 @@ impl fmt::Display for Error {
         match *self {
             Error::Io(ref err) => err.fmt(f),
             Error::Kafka(ref c) => write!(f, "Kafka Error ({:?})", c),
+            Error::TopicPartitionError(ref topic, ref partition, ref error_code) => {
+                write!(f, "Topic Partition Error ({:?}, {:?}, {:?})", topic, partition, error_code)
+            }
             #[cfg(feature = "security")]
             Error::Ssl(ref err) => err.fmt(f),
             Error::UnsupportedProtocol => write!(f, "Unsupported protocol version"),
diff --git a/src/lib.rs b/src/lib.rs
index a65da04ad..9c2052b56 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -41,13 +41,3 @@ mod protocol;
 mod compression;
 
 pub use self::error::{Error, Result};
-
-trait KafkaClientInternals {
-    fn internal_produce_messages<'a, 'b, I, J>(&mut self,
-                                               ack: client::RequiredAcks,
-                                               ack_timeout: i32,
-                                               messages: I)
-                                               -> Result<Vec<client::TopicPartitionOffset>>
-        where J: AsRef<client::ProduceMessage<'a, 'b>>,
-              I: IntoIterator<Item = J>;
-}
diff --git a/src/producer.rs b/src/producer.rs
index b7d48025c..2e8881761 100644
--- a/src/producer.rs
+++ b/src/producer.rs
@@ -63,14 +63,8 @@ use std::collections::HashMap;
 use std::fmt;
 use std::hash::{Hasher, SipHasher, BuildHasher, BuildHasherDefault};
 use std::time::Duration;
-
 use client::{self, KafkaClient};
-// public re-exports
-pub use client::Compression;
-pub use client::RequiredAcks;
-use error::Result;
-use utils::TopicPartitionOffset;
-
+use error::{Error, Result};
 use ref_slice::ref_slice;
 
 #[cfg(feature = "security")]
@@ -78,10 +72,12 @@ use client::SecurityConfig;
 
 #[cfg(not(feature = "security"))]
 type SecurityConfig = ();
-
 use client_internals::KafkaClientInternals;
 use protocol;
 
+// public re-exports
+pub use client::{Compression, RequiredAcks, ProduceConfirm, ProducePartitionConfirm};
+
 /// The default value for `Builder::with_ack_timeout`.
 pub const DEFAULT_ACK_TIMEOUT_MILLIS: u64 = 30 * 1000;
 
@@ -254,20 +250,29 @@ impl<P: Partitioner> Producer<P> {
               V: AsBytes
     {
         let mut rs = try!(self.send_all(ref_slice(rec)));
+
         if self.config.required_acks == 0 {
             // ~ with no required_acks we get no response and
             // consider the send-data request blindly as successful
             Ok(())
         } else {
             assert_eq!(1, rs.len());
-            rs.pop().unwrap().offset.map(|_| ())
+            let mut produce_confirm = rs.pop().unwrap();
+
+            assert_eq!(1, produce_confirm.partition_confirms.len());
+            produce_confirm.partition_confirms
+                .pop()
+                .unwrap()
+                .offset
+                .map(|_| ())
+                .map_err(Error::Kafka)
         }
     }
 
-    /// Synchronously send all of the specified messages to Kafka.
-    pub fn send_all<'a, K, V>(&mut self,
-                              recs: &[Record<'a, K, V>])
-                              -> Result<Vec<TopicPartitionOffset>>
+    /// Synchronously send all of the specified messages to Kafka. To validate
+    /// that all of the specified records have been successfully delivered,
+    /// inspection of the offsets on the returned confirms is necessary.
+    pub fn send_all<'a, K, V>(&mut self, recs: &[Record<'a, K, V>]) -> Result<Vec<ProduceConfirm>>
         where K: AsBytes,
               V: AsBytes
     {
diff --git a/src/protocol/consumer.rs b/src/protocol/consumer.rs
index 9a2a85acd..b650706f9 100644
--- a/src/protocol/consumer.rs
+++ b/src/protocol/consumer.rs
@@ -1,8 +1,8 @@
 use std::io::{Read, Write};
 
 use codecs::{self, ToByte, FromByte};
-use error::{self, Error, Result};
-use utils::TopicPartitionOffset;
+use error::{self, Error, Result, KafkaCode};
+use utils::PartitionOffset;
 
 use super::{HeaderRequest, HeaderResponse};
 use super::{API_KEY_OFFSET_FETCH, API_KEY_OFFSET_COMMIT, API_KEY_GROUP_COORDINATOR, API_VERSION};
@@ -186,14 +186,24 @@ pub struct PartitionOffsetFetchResponse {
 }
 
 impl PartitionOffsetFetchResponse {
-    pub fn get_offsets(&self, topic: String) -> TopicPartitionOffset {
-        TopicPartitionOffset {
-            topic: topic,
-            partition: self.partition,
-            offset: match Error::from_protocol(self.error) {
-                None => Ok(self.offset),
-                Some(e) => Err(e),
-            },
+    pub fn get_offsets(&self) -> Result<PartitionOffset> {
+        match Error::from_protocol(self.error) {
+            Some(Error::Kafka(KafkaCode::UnknownTopicOrPartition)) => {
+                // ~ occurs only on protocol v0 when no offset available
+                // for the group in question; we'll align the behavior
+                // with protocol v1.
+                Ok(PartitionOffset {
+                    partition: self.partition,
+                    offset: -1,
+                })
+            }
+            Some(e) => Err(e),
+            None => {
+                Ok(PartitionOffset {
+                    partition: self.partition,
+                    offset: self.offset,
+                })
+            }
         }
     }
 }
diff --git a/src/protocol/offset.rs b/src/protocol/offset.rs
index b34dc1b46..261e8671b 100644
--- a/src/protocol/offset.rs
+++ b/src/protocol/offset.rs
@@ -1,7 +1,8 @@
 use std::io::{Read, Write};
 
+use std;
 use codecs::{ToByte, FromByte};
-use error::{Error, Result};
+use error::{Result, KafkaCode};
 use utils::PartitionOffset;
 use super::{HeaderRequest, HeaderResponse};
 use super::{API_KEY_OFFSET, API_VERSION};
@@ -116,18 +117,20 @@ pub struct PartitionOffsetResponse {
 }
 
 impl PartitionOffsetResponse {
-    pub fn into_offset(self) -> PartitionOffset {
-        PartitionOffset {
-            partition: self.partition,
-            offset: match Error::from_protocol(self.error) {
-                None => {
-                    Ok(match self.offset.first() {
-                        Some(offs) => *offs,
-                        None => -1,
-                    })
-                }
-                Some(e) => Err(e),
-            },
+    pub fn into_offset(&self) -> std::result::Result<PartitionOffset, KafkaCode> {
+        match KafkaCode::from_protocol(self.error) {
+            Some(code) => Err(code),
+            None => {
+                let offset = match self.offset.first() {
+                    Some(offs) => *offs,
+                    None => -1,
+                };
+
+                Ok(PartitionOffset {
+                    partition: self.partition,
+                    offset: offset,
+                })
+            }
         }
     }
 }
diff --git a/src/protocol/produce.rs b/src/protocol/produce.rs
index b774b7cae..bdbe8f1cc 100644
--- a/src/protocol/produce.rs
+++ b/src/protocol/produce.rs
@@ -7,9 +7,9 @@ use compression::gzip;
 #[cfg(feature = "snappy")]
 use compression::snappy;
 
-use error::{Error, Result};
-use utils::TopicPartitionOffset;
+use error::{KafkaCode, Result};
 
+use producer::{ProduceConfirm, ProducePartitionConfirm};
 use super::{HeaderRequest, HeaderResponse};
 use super::{API_KEY_PRODUCE, API_VERSION};
 use super::to_crc;
@@ -257,31 +257,35 @@ pub struct PartitionProduceResponse {
 }
 
 impl ProduceResponse {
-    pub fn get_response(&self) -> Vec<TopicPartitionOffset> {
+    pub fn get_response(self) -> Vec<ProduceConfirm> {
         self.topic_partitions
-            .iter()
-            .flat_map(|ref tp| tp.get_response(tp.topic.clone()))
+            .into_iter()
+            .map(|tp| tp.get_response())
             .collect()
     }
 }
 
 impl TopicPartitionProduceResponse {
-    pub fn get_response(&self, topic: String) -> Vec<TopicPartitionOffset> {
-        self.partitions
+    pub fn get_response(self) -> ProduceConfirm {
+        let confirms = self.partitions
             .iter()
-            .map(|ref p| p.get_response(topic.clone()))
-            .collect()
+            .map(|ref p| p.get_response())
+            .collect();
+
+        ProduceConfirm {
+            topic: self.topic,
+            partition_confirms: confirms,
+        }
     }
 }
 
 impl PartitionProduceResponse {
-    pub fn get_response(&self, topic: String) -> TopicPartitionOffset {
-        TopicPartitionOffset {
-            topic: topic,
+    pub fn get_response(&self) -> ProducePartitionConfirm {
+        ProducePartitionConfirm {
             partition: self.partition,
-            offset: match Error::from_protocol(self.error) {
+            offset: match KafkaCode::from_protocol(self.error) {
                 None => Ok(self.offset),
-                Some(e) => Err(e),
+                Some(code) => Err(code),
             },
         }
     }
diff --git a/src/utils.rs b/src/utils.rs
index dfd4402be..97c9d216d 100644
--- a/src/utils.rs
+++ b/src/utils.rs
@@ -2,20 +2,10 @@
 //!
 //! This module is _not_ exposed to the public directly.
 
-use error::Result;
-
 /// A retrieved offset for a particular partition in the context of an
 /// already known topic.
 #[derive(Debug)]
 pub struct PartitionOffset {
-    pub offset: Result<i64>,
-    pub partition: i32,
-}
-
-/// A retrieved offset of a particular topic partition.
-#[derive(Debug)]
-pub struct TopicPartitionOffset {
-    pub offset: Result<i64>,
-    pub topic: String,
+    pub offset: i64,
     pub partition: i32,
 }


Test patch: