code: def get_messages(self, count=1, block=True, timeout=10):
          
          messages = []
  
          # Give a size hint to the consumers. Each consumer process will fetch
          # a maximum of "count" messages. This will fetch more messages than
          # necessary, but these will not be committed to kafka. Also, the extra
          # messages can be provided in subsequent runs
          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):
              # Trigger consumption only if the queue is empty
              # By doing this, we will ensure that consumers do not
              # go into overdrive and keep consuming thousands of
              # messages when the user might need only a few
              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()
  
          # Update and commit offsets if necessary
          self.offsets.update(new_offsets)
          self.count_since_commit += len(messages)
          self._auto_commit()
  
          return messages
docs: 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.