docs: private void fufillFuture(ReceiveMessageFuture future) {
          ReceiveMessageBatchTask task = finishedTasks.getFirst();
          ReceiveMessageResult result = new ReceiveMessageResult();
          LinkedList<Message> messages = new LinkedList<Message>();
          result.setMessages(messages);
          Exception exception = task.getException();
          int numRetrieved = 0;
          boolean batchDone = false;
          while (numRetrieved < future.getRequestedSize()) {
              Message msg = task.removeMessage();
              // a non-empty batch can still give back a null
              // message if the message expired.
              if (msg != null) {
                  messages.add(msg);
                  ++numRetrieved;
              } else {
                  batchDone = true;
                  break;
              }
          }
          // we may have just drained the batch.
          batchDone = batchDone || task.isEmpty() || (exception != null);
          if (batchDone) {
              finishedTasks.removeFirst();
          }
          result.setMessages(messages);
  
          // if after the above runs the exception is not null,
          // the finished batch has encountered an error, and we will
          // report that in the Future. Otherwise, we will fill
          // the future with the receive result
          if (exception != null) {
              future.setFailure(exception);
          } else {
              future.setSuccess(result);
          }
      }