Modify offsets metadata so that the partition offsets
    have null payloads.

    :param offsets: dict {<topic>: {<partition>: <offset>}}
    :returns: a dict topic: partition: offset
973555
def nullify_offsets(offsets):
    
    result = {}
    for topic, partition_offsets in six.iteritems(offsets):
        result[topic] = _nullify_partition_offsets(partition_offsets)
    return result
1340379
def reset_partition_offset(self, partition):
        
        LATEST = -1
        EARLIEST = -2
        if self.auto_offset_reset == 'largest':
            reqs = [OffsetRequestPayload(self.topic, partition, LATEST, 1)]
        elif self.auto_offset_reset == 'smallest':
            reqs = [OffsetRequestPayload(self.topic, partition, EARLIEST, 1)]
        else:
            # Let's raise an reasonable exception type if user calls
            # outside of an exception context
            if sys.exc_info() == (None, None, None):
                raise OffsetOutOfRangeError('Cannot reset partition offsets without a '
                                            'valid auto_offset_reset setting '
                                            '(largest|smallest)')
            # Otherwise we should re-raise the upstream exception
            # b/c it typically includes additional data about
            # the request that triggered it, and we do not want to drop that
            raise # pylint: disable=E0704

        # send_offset_request
        log.info('Resetting topic-partition offset to %s for %s:%d',
                 self.auto_offset_reset, self.topic, partition)
        try:
            (resp, ) = self.client.send_offset_request(reqs)
        except KafkaError as e:
            log.error('%s sending offset request for %s:%d',
                      e.__class__.__name__, self.topic, partition)
        else:
            self.offsets[partition] = resp.offsets[0]
            self.fetch_offsets[partition] = resp.offsets[0]
            return resp.offsets[0]
1007081
def offset(self, n):
        

        idx = self.index
        if idx is not None:
            sibling = idx + n
            container = self.container
            if 0 <= sibling < len(container):
                return container[sibling]
757737
def from_offset(self, value):
        '''
        
        '''
        if not self.params:
            self.params = dict({'from':value})
            return self
        self.params['from'] = value
        return self
280128
def offset(self, offset):
        
        if self._offset != offset:
            self.dirty = True
            self._offset = offset
        return self
947758
def parse_headers(lines, offset=0):
    
    headers = {}
    for header_line in lines[offset:]:
        header_match = HEADER_LINE_RE.match(header_line)
        if header_match:
            key = header_match.group('key')
            key = re.sub(r'\\.', _unescape_header, key)
            if key not in headers:
                value = header_match.group('value')
                value = re.sub(r'\\.', _unescape_header, value)
                headers[key] = value
    return headers
133754
def _init_offsets(self, batchsize):
        
        upper_offsets = previous_lower_offsets = self._lower_offsets
        if not upper_offsets:
            upper_offsets = self.latest_offsets
        self._upper_offsets = {p: o for p, o in upper_offsets.items() if o > self._min_lower_offsets[p]}

        # remove db dupes not used anymore
        if self._dupes:
            for p in list(six.iterkeys(self._dupes)):
                if p not in self._upper_offsets:
                    db = self._dupes.pop(p)
                    db.close()
                    os.remove(db.filename)

        partition_batchsize = 0
        if self._upper_offsets:
            partition_batchsize = max(int(batchsize * self.__scan_excess), batchsize)
            self._lower_offsets = self._upper_offsets.copy()
            total_offsets_run = 0
            for p in sorted(self._upper_offsets.keys()):
                # readjust partition_batchsize when a partition scan starts from latest offset
                if total_offsets_run > 0 and partition_batchsize > batchsize:
                    partition_batchsize = batchsize
                if partition_batchsize > 0:
                    self._lower_offsets[p] = max(self._upper_offsets[p] - partition_batchsize, self._min_lower_offsets[p])
                    offsets_run = self._upper_offsets[p] - self._lower_offsets[p]
                    total_offsets_run += offsets_run
                    partition_batchsize = partition_batchsize - offsets_run
                else:
                    break
            log.info('Offset run: %d', total_offsets_run)
            # create new consumer if partition list changes
            if previous_lower_offsets is not None and set(previous_lower_offsets.keys()) != set(self._lower_offsets):
                self._create_scan_consumer(self._lower_offsets.keys())

            # consumer must restart from newly computed lower offsets
            self._update_offsets(self._lower_offsets)
        log.info('Initial offsets for topic %s: %s', self._topic, repr(self._lower_offsets))
        log.info('Target offsets for topic %s: %s', self._topic, repr(self._upper_offsets))

        return batchsize
559839
def parse_consumer_offsets(cls, json_file):
        """"""
        with open(json_file, 'r') as consumer_offsets_json:
            try:
                parsed_offsets = {}
                parsed_offsets_data = json.load(consumer_offsets_json)
                # Create new dict with partition-keys as integers
                parsed_offsets['groupid'] = parsed_offsets_data['groupid']
                parsed_offsets['offsets'] = {}
                for topic, topic_data in six.iteritems(parsed_offsets_data['offsets']):
                    parsed_offsets['offsets'][topic] = {}
                    for partition, offset in six.iteritems(topic_data):
                        parsed_offsets['offsets'][topic][int(partition)] = offset
                return parsed_offsets
            except ValueError:
                print(
                    "Error: Given consumer-data json data-file {file} could not be "
                    "parsed".format(file=json_file),
                    file=sys.stderr,
                )
                raise
217115
def next_offsets(self):
        # type: (Descriptor) -> Offsets
        
        resume_bytes = self._resume()
        with self._meta_lock:
            if self._chunk_num >= self._total_chunks:
                return None, resume_bytes
            if self._offset + self._chunk_size > self._ase.size:
                num_bytes = self._ase.size - self._offset
            else:
                num_bytes = self._chunk_size
            chunk_num = self._chunk_num
            range_start = self._offset
            range_end = self._offset + num_bytes - 1
            self._offset += num_bytes
            self._chunk_num += 1
            if self._ase.is_encrypted and self._offset >= self._ase.size:
                pad = True
            else:
                pad = False
            return Offsets(
                chunk_num=chunk_num,
                num_bytes=num_bytes,
                range_start=range_start,
                range_end=range_end,
                pad=pad,
            ), resume_bytes
309138
def add_offsets(self, offset_ns=None):
        
        if offset_ns is None:
            offset_ns = self.ns

        onset = 0
        offset = 0

        for token_id, token_str in self.get_tokens():
            offset = onset + len(token_str)
            self.node[token_id]['{0}:{1}'.format(offset_ns, 'onset')] = onset
            self.node[token_id]['{0}:{1}'.format(offset_ns, 'offset')] = offset
            onset = offset + 1
617509
def pending(self, partitions=None):
        
        if partitions is None:
            partitions = self.offsets.keys()

        total = 0
        reqs = []

        for partition in partitions:
            reqs.append(OffsetRequestPayload(self.topic, partition, -1, 1))

        resps = self.client.send_offset_request(reqs)
        for resp in resps:
            partition = resp.partition
            pending = resp.offsets[0]
            offset = self.offsets[partition]
            total += pending - offset

        return total
264335
def get_topic_partition_with_error(cluster_config, error, fetch_unavailable_brokers=False):
    """"""

    metadata = get_topic_partition_metadata(cluster_config.broker_list)
    affected_partitions = set()
    if fetch_unavailable_brokers:
        unavailable_brokers = set()
    with ZK(cluster_config) as zk:
        for partitions in metadata.values():
            for partition_metadata in partitions.values():
                if int(partition_metadata.error) == error:
                    if fetch_unavailable_brokers:
                        unavailable_brokers |= get_unavailable_brokers(zk, partition_metadata)
                    affected_partitions.add((partition_metadata.topic, partition_metadata.partition))

    if fetch_unavailable_brokers:
        return affected_partitions, unavailable_brokers
    else:
        return affected_partitions
1260757
def offset(self, offset):
        
        query = self._copy()
        query._offset = offset
        return query
292419
def position(self, partition):
        
        if not isinstance(partition, TopicPartition):
            raise TypeError('partition must be a TopicPartition namedtuple')
        assert self._subscription.is_assigned(partition), 'Partition is not assigned'
        offset = self._subscription.assignment[partition].position
        if offset is None:
            self._update_fetch_positions([partition])
            offset = self._subscription.assignment[partition].position
        return offset
471171
def decode_offset_response(cls, response):
        
        return [
            kafka.structs.OffsetResponsePayload(topic, partition, error, tuple(offsets))
            for topic, partitions in response.topics
            for partition, error, offsets in partitions
        ]
427343
private static List<TopicPartition> convertKafkaPartitions(List<KafkaTopicPartitionState<TopicPartition>> partitions) {
		ArrayList<TopicPartition> result = new ArrayList<>(partitions.size());
		for (KafkaTopicPartitionState<TopicPartition> p : partitions) {
			result.add(p.getKafkaPartitionHandle());
		}
		return result;
	}
995298
def clear_offsets_excluding(excluded)
      # Clear all offsets that aren't in `excluded`.
      @processed_offsets.each do |topic, partitions|
        partitions.keep_if do |partition, _|
          excluded.fetch(topic, []).include?(partition)
        end
      end

      # Clear the cached commits from the brokers.
      @committed_offsets = nil
      @resolved_offsets.clear
    end