pub fn commit_offsets<'a, J, I>(&mut self, group: &str, offsets: I) -> Result<()>
        where J: AsRef<CommitOffset<'a>>,
              I: IntoIterator<Item = J>
    {
        let mut req = protocol::OffsetCommitRequest::new(group,
                                                         self.config.offset_commit_version,
                                                         self.state.next_correlation_id(),
                                                         &self.config.client_id);
        for o in offsets {
            let o = o.as_ref();
            if self.state.contains_topic_partition(o.topic, o.partition) {
                req.add(o.topic, o.partition, o.offset, "");
            } else {
                return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
            }
        }
        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 for a topic partitions on behalf of a consumer group.
///
/// # Examples
///
/// ```no_run
/// use kafka::client::{KafkaClient, CommitOffset};
///
/// let mut client = KafkaClient::new(vec!["localhost:9092".to_owned()]);
/// client.load_metadata_all().unwrap();
/// client.commit_offsets("my-group",
///     &[CommitOffset::new("my-topic", 0, 100),
///       CommitOffset::new("my-topic", 1, 99)])
///    .unwrap();
/// ```
///
/// In this example, we commit the offset 100 for the topic
/// partition "my-topic:0" and 99 for the topic partition
/// "my-topic:1".  Once successfully committed, these can then be
/// retrieved using `fetch_group_offsets` even from another
/// process or at much later point in time to resume comusing the
/// topic partitions as of these offsets.
Commit offset for a topic partitions on behalf of a consumer group.
Commit offset for a topic partitions on behalf of a consumer group .
pub fn commit_offsets<'a, J, I>(&mut self, group: &str, offsets: I) -> Result<()>
        where J: AsRef<CommitOffset<'a>>,
              I: IntoIterator<Item = J>
    {
        let mut req = protocol::OffsetCommitRequest::new(group,
                                                         self.config.offset_commit_version,
                                                         self.state.next_correlation_id(),
                                                         &self.config.client_id);
        for o in offsets {
            let o = o.as_ref();
            if self.state.contains_topic_partition(o.topic, o.partition) {
                req.add(o.topic, o.partition, o.offset, "");
            } else {
                return Err(Error::Kafka(KafkaCode::UnknownTopicOrPartition));
            }
        }
        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)
        }
    }
pub fn commit_offsets < ' a , J , I > ( & mut self , group : & str , offsets : I ) -> Result < ( ) > where J : AsRef < CommitOffset < ' a > > , I : IntoIterator < Item = J > { let mut req = protocol :: OffsetCommitRequest :: new ( group , self . config . offset_commit_version , self . state . next_correlation_id ( ) , & self . config . client_id ) ; for o in offsets { let o = o . as_ref ( ) ; if self . state . contains_topic_partition ( o . topic , o . partition ) { req . add ( o . topic , o . partition , o . offset , " " ) ; } else { return Err ( Error :: Kafka ( KafkaCode :: UnknownTopicOrPartition ) ) ; } } if req . topic_partitions . is_empty ( ) { debug ! ( " " ) ; Ok ( ( ) ) } else { __commit_offsets ( req , & mut self . state , & mut self . conn_pool , & self . config ) } }
Commit offset for a topic partitions on behalf of a consumer group.
Commit offset for a topic partitions on behalf of a consumer group .
&mut self
group
&str
offsets
I
examples
```no_run
use kafka::client::{KafkaClient, CommitOffset}.

let mut client = KafkaClient::new(vec!["localhost:9092".to_owned()]);
client.load_metadata_all().unwrap();
client.commit_offsets("my-group",
&[CommitOffset::new("my-topic", 0, 100),
CommitOffset::new("my-topic", 1, 99)])
.unwrap();
```

In this example, we commit the offset 100 for the topic
partition "my-topic:0" and 99 for the topic partition
"my-topic:1".  Once successfully committed, these can then be
retrieved using `fetch_group_offsets` even from another
process or at much later point in time to resume comusing the
topic partitions as of these offsets.
` ` ` no_run use kafka :: client :: { KafkaClient CommitOffset } . let mut client = KafkaClient :: new ( vec! [ " localhost : 9092 " . to_owned () ] ) ; client . load_metadata_all () . unwrap () ; client . commit_offsets ( " my - group " & [ CommitOffset :: new ( " my - topic " 0 100 ) CommitOffset :: new ( " my - topic " 1 99 ) ] ) . unwrap () ; ` ` ` In this example we commit the offset 100 for the topic partition " my - topic : 0 " and 99 for the topic partition " my - topic : 1 " . Once successfully committed these can then be retrieved using ` fetch_group_offsets ` even from another process or at much later point in time to resume comusing the topic partitions as of these offsets .