code: @Override
      public void run() {
          while (true) {
              int prefetchBatchSize;
              boolean nackQueueMessages = false;
              List<Message> messages = null;
              try {
                  if (isClosed()) {
                      break;
                  }
                  
                  synchronized (stateLock) {
                      waitForStart();
                      waitForPrefetch();
                      prefetchBatchSize = Math.min(numberOfMessagesToFetch(), SQSMessagingClientConstants.MAX_BATCH);
                  }
  
                  if (!isClosed()) {
                      messages = getMessages(prefetchBatchSize);
                  }
  
                  if (messages != null && !messages.isEmpty()) {
                      processReceivedMessages(messages);
                  }
              } catch (InterruptedException e) {
                  nackQueueMessages = true;
                  break;
              } catch (Throwable e) {
                  LOG.error("Unexpected exception when prefetch messages:", e);
                  nackQueueMessages = true;
                  throw new RuntimeException(e);
              } finally {
                  if (isClosed() || nackQueueMessages) {
                      nackQueueMessages();
                  }
              }
          }
      }
docs: Runs until the message consumer is closed and in-progress SQS
  <code>receiveMessage</code> call returns.
  <P>
  This blocks if configured number of prefetched messages are already
  received or connection has not started yet.
  <P>
  After consumer is closed, all the messages inside internal buffer will
  be negatively acknowledged.