def commit(self):
if not self._config['<STR_LIT>']:<EOL><INDENT>logger.warning('<STR_LIT>')<EOL>raise KafkaConfigurationError(<EOL>'<STR_LIT>'<EOL>'<STR_LIT>'<EOL>)<EOL><DEDENT>metadata = b'<STR_LIT>'<EOL>offsets = self._offsets.task_done<EOL>commits = []<EOL>for topic_partition, task_done_offset in six.iteritems(offsets):<EOL><INDENT>if task_done_offset is None:<EOL><INDENT>continue<EOL><DEDENT>commit_offset = (task_done_offset + <NUM_LIT:1>)<EOL>if commit_offset == self._offsets.commit[topic_partition]:<EOL><INDENT>continue<EOL><DEDENT>commits.append(<EOL>OffsetCommitRequest(topic_partition[<NUM_LIT:0>], topic_partition[<NUM_LIT:1>],<EOL>commit_offset, metadata)<EOL>)<EOL><DEDENT>if commits:<EOL><INDENT>logger.info('<STR_LIT>', self._config['<STR_LIT>'])<EOL>resps = self._client.send_offset_commit_request(<EOL>kafka_bytestring(self._config['<STR_LIT>']), commits,<EOL>fail_on_error=False<EOL>)<EOL>for r in resps:<EOL><INDENT>check_error(r)<EOL>topic_partition = (r.topic, r.partition)<EOL>task_done = self._offsets.task_done[topic_partition]<EOL>self._offsets.commit[topic_partition] = (task_done + <NUM_LIT:1>)<EOL><DEDENT>if self._config['<STR_LIT>']:<EOL><INDENT>self._reset_auto_commit()<EOL><DEDENT>return True<EOL><DEDENT>else:<EOL><INDENT>logger.info('<STR_LIT>', self._config['<STR_LIT>'])<EOL>return False<EOL><DEDENT>
Store consumed message offsets (marked via task_done())
        to kafka cluster for this consumer_group.

        Returns:
            True on success, or False if no offsets were found for commit

        Note:
            this functionality requires server version >=0.8.1.1
            https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetCommit/FetchAPI