docs: def commit(self):
          """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
          """
          if not self._config['group_id']:
              logger.warning('Cannot commit without a group_id!')
              raise KafkaConfigurationError(
                  'Attempted to commit offsets '
                  'without a configured consumer group (group_id)'
              )
  
          # API supports storing metadata with each commit
          # but for now it is unused
          metadata = b''
  
          offsets = self._offsets.task_done
          commits = []
          for topic_partition, task_done_offset in six.iteritems(offsets):
  
              # Skip if None
              if task_done_offset is None:
                  continue
  
              # Commit offsets as the next offset to fetch
              # which is consistent with the Java Client
              # task_done is marked by messages consumed,
              # so add one to mark the next message for fetching
              commit_offset = (task_done_offset + 1)
  
              # Skip if no change from previous committed
              if commit_offset == self._offsets.commit[topic_partition]:
                  continue
  
              commits.append(
                  OffsetCommitRequest(topic_partition[0], topic_partition[1],
                                      commit_offset, metadata)
              )
  
          if commits:
              logger.info('committing consumer offsets to group %s', self._config['group_id'])
              resps = self._client.send_offset_commit_request(
                  kafka_bytestring(self._config['group_id']), commits,
                  fail_on_error=False
              )
  
              for r in resps:
                  check_error(r)
                  topic_partition = (r.topic, r.partition)
                  task_done = self._offsets.task_done[topic_partition]
                  self._offsets.commit[topic_partition] = (task_done + 1)
  
              if self._config['auto_commit_enable']:
                  self._reset_auto_commit()
  
              return True
  
          else:
              logger.info('No new offsets found to commit in group %s', self._config['group_id'])
              return False