code: 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);
          }
      }
docs: Fills the future with whatever results were received by the full batch currently at the head
  of the completed batch queue. Those results may be retrieved messages, or an exception. This
  method assumes that you are holding the finished tasks lock locks when invoking it. violate
  this assumption at your own peril