Merge the partition offsets of a single topic from multiple responses.

    :param partition_offsets: list of dict partition: offset
    :returns: dict partition: offset
542977
def merge_partition_offsets(*partition_offsets):
    
    output = dict()
    for partition_offset in partition_offsets:
        for partition, offset in six.iteritems(partition_offset):
            prev_offset = output.get(partition, 0)
            output[partition] = max(prev_offset, offset)
    return output
336959
def get_number_of_messages_in_topicpartition(self, topic_partition=None):
        

        if isinstance(topic_partition, TopicPartition):
            topic_partition = [topic_partition]

        number_of_messages = 0
        assignment = self.consumer.assignment()

        self.consumer.unsubscribe()
        for Partition in topic_partition:
            if not isinstance(Partition, TopicPartition):
                raise TypeError("topic_partition must be of type TopicPartition, create it with Create TopicPartition keyword.")

            self.assign_to_topic_partition(Partition)

            self.consumer.seek_to_end(Partition)
            end = self.consumer.position(Partition)
            self.consumer.seek_to_beginning(Partition)
            start = self.consumer.position(Partition)
            number_of_messages += end-start

        self.consumer.unsubscribe()
        self.consumer.assign(assignment)
        return number_of_messages
297423
def assign_to_topic_partition(self, topic_partition=None):
        

        if isinstance(topic_partition, TopicPartition):
            topic_partition = [topic_partition]
        if not self._is_assigned(topic_partition):
            self.consumer.assign(topic_partition)
765543
def next_offset_for(topic, partition)
      offset = @processed_offsets.fetch(topic, {}).fetch(partition) {
        committed_offset_for(topic, partition)
      }

      # A negative offset means that no offset has been committed, so we need to
      # resolve the default offset for the topic.
      if offset < 0
        resolve_offset(topic, partition)
      else
        # The next offset is the last offset.
        offset
      end
    end
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
1331623
def decode_produce_response(cls, response):
        
        return [
            kafka.structs.ProduceResponsePayload(topic, partition, error, offset)
            for topic, partitions in response.topics
            for partition, error, offset in partitions
        ]
917631
def replicated_partitions
      result, mutex = [], Mutex.new
      threads = cluster.partitions.map do |partition|
        Thread.new do
          Thread.abort_on_exception = true
          select = partition.replicas.include?(self)
          mutex.synchronize { result << partition } if select
        end
      end
      threads.each(&:join)
      result
    end
536917
def index_partition(self, partition, force=False):
        
        self.backend.partition_index.index_one(partition, force=force)
932540
def decode_offset_commit_response(cls, data):
        
        ((correlation_id,), cur) = relative_unpack('>i', data, 0)
        ((num_topics,), cur) = relative_unpack('>i', data, cur)

        for _ in xrange(num_topics):
            (topic, cur) = read_short_string(data, cur)
            ((num_partitions,), cur) = relative_unpack('>i', data, cur)

            for _ in xrange(num_partitions):
                ((partition, error), cur) = relative_unpack('>ih', data, cur)
                yield OffsetCommitResponse(topic, partition, error)
1335050
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._chunk_size == -1 and self._src_block_list is not None:
                num_bytes = self._src_block_list[self._chunk_num].size
            else:
                if self._offset + self._chunk_size > self._src_ase.size:
                    num_bytes = self._src_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
            return Offsets(
                chunk_num=chunk_num,
                num_bytes=num_bytes,
                range_start=range_start,
                range_end=range_end,
            ), resume_bytes
54125
public List<Long> getOffsets(OffsetRequest offsetRequest) {
        ILog log = getLog(offsetRequest.topic, offsetRequest.partition);
        if (log != null) {
            return log.getOffsetsBefore(offsetRequest);
        }
        return ILog.EMPTY_OFFSETS;
    }
702366
def exchange_partitions(self, partitionSpecs, source_db, source_table_name, dest_db, dest_table_name):
    
    self.send_exchange_partitions(partitionSpecs, source_db, source_table_name, dest_db, dest_table_name)
    return self.recv_exchange_partitions()
462592
def decode_list_offset_response(cls, response):
        
        return [
            kafka.structs.ListOffsetResponsePayload(topic, partition, error, timestamp, offset)
            for topic, partitions in response.topics
            for partition, error, timestamp, offset in partitions
        ]
929808
def build_new_offsets(cls, client, topics_offset_data, topic_partitions, current_offsets):
        
        new_offsets = defaultdict(dict)
        try:
            for topic, partitions in six.iteritems(topic_partitions):
                # Validate current offsets in range of low and highmarks
                # Currently we only validate for positive offsets and warn
                # if out of range of low and highmarks
                valid_partitions = set()
                for topic_partition_offsets in current_offsets[topic]:
                    partition = topic_partition_offsets.partition
                    valid_partitions.add(partition)
                    # Skip the partition not present in list
                    if partition not in topic_partitions[topic]:
                        continue
                    lowmark = topic_partition_offsets.lowmark
                    highmark = topic_partition_offsets.highmark
                    new_offset = topics_offset_data[topic][partition]
                    if new_offset < 0:
                        print(
                            "Error: Given offset: {offset} is negative"
                            .format(offset=new_offset),
                            file=sys.stderr,
                        )
                        sys.exit(1)
                    if new_offset < lowmark or new_offset > highmark:
                        print(
                            "Warning: Given offset {offset} for topic-partition "
                            "{topic}:{partition} is outside the range of lowmark "
                            "{lowmark} and highmark {highmark}".format(
                                offset=new_offset,
                                topic=topic,
                                partition=partition,
                                lowmark=lowmark,
                                highmark=highmark,
                            )
                        )
                    new_offsets[topic][partition] = new_offset
                if not set(partitions).issubset(valid_partitions):
                    print(
                        "Error: Some invalid partitions {partitions} for topic "
                        "{topic} found. Valid partition-list {valid_partitions}. "
                        "Exiting...".format(
                            partitions=', '.join([str(p) for p in partitions]),
                            valid_partitions=', '.join([str(p) for p in valid_partitions]),
                            topic=topic,
                        ),
                        file=sys.stderr,
                    )
                    sys.exit(1)
        except KeyError as ex:
            print(
                "Error: Possible invalid topic or partition. Error msg: {ex}. "
                "Exiting...".format(ex=ex),
            )
            sys.exit(1)
        return new_offsets
1351268
def _get_disorder_mappings(composition, partitions):
        

        def _get_replacement_dict_from_partition(partition):
            d = {}  # to be passed to Structure.replace_species()
            for sp_list in partition:
                if len(sp_list) > 1:
                    total_occ = sum([composition[sp] for sp in sp_list])
                    merged_comp = {sp: composition[sp] / total_occ for sp in sp_list}
                    for sp in sp_list:
                        d[sp] = merged_comp
            return d

        disorder_mapping = [_get_replacement_dict_from_partition(p)
                            for p in partitions]

        return disorder_mapping
771449
function(topic, cb) {
    if (topic.params.offset === undefined || topic.params.offset === -1) {
      var fetchFrom = topic.broker;

      if (topic.params.fetchFromCoordinator) {
        fetchFrom = _groupCoordinator;
      }

      fetchFrom.offsetFetch({
        group: _options.group,
        topic: topic.params.topic,
        partition: topic.params.partition,
        fetchFromCoordinator: topic.params.fetchFromCoordinator
      }, function(err, response) {
        if (err) { return cb(err); }
        var partition = _.findWhere(_.findWhere(response.topics, {name: topic.params.topic}).partitions, {partitionId: topic.params.partition});
        if (partition.error === 0) {
          if (partition.offsetLo === 0xffffffff &&
              partition.offsetHi === 0xffffffff) {

            // BUG in Kafka - if the topic offset has not been set then an error condition is returned and
            // -1 is set for the offset, once an offset is set this condition is cleared
            topic.params.offset = 0;
          } else {
            topic.params.offset = partition.offsetLo;
          }
          cb(null, topic);
        }
        else {
          cb(partition.error, null);
        }
      });
    }
    else {
      cb(null, topic);
    }
  }
91347
def _recursive_merged_items(self, index):
        
        subdirs = [os.path.join(d, "parts", str(index)) for d in self.localdirs]
        m = ExternalMerger(self.agg, self.memory_limit, self.serializer, subdirs,
                           self.scale * self.partitions, self.partitions, self.batch)
        m.pdata = [{} for _ in range(self.partitions)]
        limit = self._next_limit()

        for j in range(self.spills):
            path = self._get_spill_dir(j)
            p = os.path.join(path, str(index))
            with open(p, 'rb') as f:
                m.mergeCombiners(self.serializer.load_stream(f), 0)

            if get_used_memory() > limit:
                m._spill()
                limit = self._next_limit()

        return m._external_items()