docs: def consume_messages(self, batchsize):
          """ Get messages batch from the reservoir """
          if not self._reservoir:
              self.finished = True
              return
          for msg in self._reservoir[:batchsize]:
              yield msg
          self._reservoir = self._reservoir[batchsize:]