Return number of messages in TopicPartition.
        
        - ``topic_partition`` (list of TopicPartition)
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
1155432
@Then("^The number of partitions in topic '(.+?)' should be '(.+?)''?$")
    public void checkNumberOfPartitions(String topic_name, int numOfPartitions) throws Exception {
        Assertions.assertThat(commonspec.getKafkaUtils().getPartitions(topic_name)).isEqualTo(numOfPartitions);

    }
1271539
def get_monitor(self, topics):
        
        for monitor in self.get_monitors(MON_TOPIC_ATTR == ",".join(topics)):
            return monitor  # return the first one, even if there are multiple
        return None
107938
def message(topic, message = nil)
      raise ArgumentError, "block or message, not both" if block_given? && message

      message = yield if block_given?
      message = message.to_s.gsub(%r!\s+!, " ")
      topic = formatted_topic(topic, block_given?)
      out = topic + message
      messages << out
      out
    end
2603
def _get_message(self, block=True, timeout=0.1, get_partition_info=None,
                     update_offset=True):
        
        start_at = time.time()
        while self.queue.empty():
            # We're out of messages, go grab some more.
            log.debug('internal queue empty, fetching more messages')
            with FetchContext(self, block, timeout):
                self._fetch()

            if not block or time.time() > (start_at + timeout):
                break

        try:
            partition, message = self.queue.get_nowait()

            if update_offset:
                # Update partition offset
                self.offsets[partition] = message.offset + 1

                # Count, check and commit messages if necessary
                self.count_since_commit += 1
                self._auto_commit()

            if get_partition_info is None:
                get_partition_info = self.partition_info
            if get_partition_info:
                return partition, message
            else:
                return message
        except queue.Empty:
            log.debug('internal queue empty after fetch - returning None')
            return None
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)
856453
def get_messages(self):
        
        if not self._messages:  # if not already cached
            query = self._dbman.query('thread:' + self._id)
            thread = next(query.search_threads())

            def accumulate(acc, msg):
                M = Message(self._dbman, msg, thread=self)
                acc[M] = []
                r = msg.get_replies()
                if r is not None:
                    for m in r:
                        acc[M].append(accumulate(acc, m))
                return M

            self._messages = {}
            for m in thread.get_toplevel_messages():
                self._toplevel_messages.append(accumulate(self._messages, m))
        return self._messages
1242256
def get_number_of_partitions_for(self, ar):
        
        # fetch the number of partitions from the request
        uid = api.get_uid(ar)
        num = self.request.get("primary", {}).get(uid)

        if num is None:
            # get the number of partitions from the template
            template = ar.getTemplate()
            if template:
                num = len(template.getPartitions())
            else:
                num = DEFAULT_NUMBER_OF_PARTITIONS
        try:
            num = int(num)
        except (TypeError, ValueError):
            num = DEFAULT_NUMBER_OF_PARTITIONS
        return num
122788
def get_position(self, topic_partition=None):
        

        if isinstance(topic_partition, TopicPartition):
            return self.consumer.position(topic_partition)
        else:
            raise TypeError("topic_partition must be of type TopicPartition, create it with Create TopicPartition keyword.")
534108
def getStringPartition(self):
        
        res = ''
        for s in self.partitions[self.index].getSlice():
            start = ''
            stop = ''
            if s.start is not None:
                start = int(s.start)
            if s.stop is not None:
                stop = int(s.stop)
            res += '{0}:{1},'.format(start, stop)
        return res
1058024
def clear_messages(topic:, partition:)
      return unless @buffer.key?(topic) && @buffer[topic].key?(partition)

      @size -= @buffer[topic][partition].count
      @bytesize -= @buffer[topic][partition].map(&:bytesize).reduce(0, :+)

      @buffer[topic].delete(partition)
      @buffer.delete(topic) if @buffer[topic].empty?
    end
658359
def get_topic_tree(rs, topic, depth=0):
    

    # Break if we're in too deep.
    if depth > rs._depth:
        rs._warn("Deep recursion while scanning topic trees!")
        return []

    # Collect an array of all topics.
    topics = [topic]

    # Does this topic include others?
    if topic in rs._includes:
        # Try each of these.
        for includes in sorted(rs._includes[topic]):
            topics.extend(get_topic_tree(rs, includes, depth + 1))

    # Does this topic inherit others?
    if topic in rs._lineage:
        # Try each of these.
        for inherits in sorted(rs._lineage[topic]):
            topics.extend(get_topic_tree(rs, inherits, depth + 1))

    return topics
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
20694
def partitions_for_topic(self, topic):
        
        if topic not in self._partitions:
            return None
        return set(self._partitions[topic].keys())
803942
private static List<String> getTopics(List<KafkaTopicPartitionState<TopicAndPartition>> partitions) {
		HashSet<String> uniqueTopics = new HashSet<>();
		for (KafkaTopicPartitionState<TopicAndPartition> fp: partitions) {
			uniqueTopics.add(fp.getTopic());
		}
		return new ArrayList<>(uniqueTopics);
	}
700804
def total_msgs(xml):
    ''''''
    count = 0
    for x in xml:
        count += len(x.message)
    return count
538558
def split_message(message, max_length):
    """"""
    ms = []
    while len(message) > max_length:
        ms.append(message[:max_length])
        message = message[max_length:]
    ms.append(message)
    return ms