'Fetch the specified number of messages
Keyword Arguments:
count: Indicates the maximum number of messages to be fetched
block: If True, the API will block till all messages are fetched.
If block is a positive integer the API will block until that
many messages are fetched.
timeout: When blocking is requested the function will block for
the specified time (in seconds) until count messages is
fetched. If None, it will block forever.'

def get_messages(self, count=1, block=True, timeout=10):

    messages = []
    self.size.value = count
    self.events.pause.clear()
    if (timeout is not None):
        max_time = (time.time() + timeout)
    new_offsets = {}
    while ((count > 0) and ((timeout is None) or (timeout > 0))):
        if self.queue.empty():
            self.events.start.set()
        block_next_call = ((block is True) or (block > len(messages)))
        try:
            (partition, message) = self.queue.get(block_next_call, timeout)
        except queue.Empty:
            break
        _msg = ((partition, message) if self.partition_info else message)
        messages.append(_msg)
        new_offsets[partition] = (message.offset + 1)
        count -= 1
        if (timeout is not None):
            timeout = (max_time - time.time())
    self.size.value = 0
    self.events.start.clear()
    self.events.pause.set()
    self.offsets.update(new_offsets)
    self.count_since_commit += len(messages)
    self._auto_commit()
    return messages
