docs: public SnowflakeResultChunk getNextChunkToConsume() throws InterruptedException,
                                                               SnowflakeSQLException
    {
      // free previous chunk data and submit a new chunk for downloading
      if (this.nextChunkToConsume > 0)
      {
        int prevChunk = this.nextChunkToConsume - 1;
  
        // free the chunk data for previous chunk
        logger.debug("free chunk data for chunk #{}",
                     prevChunk);
  
        long chunkMemUsage = chunks.get(prevChunk).computeNeededChunkMemory();
        if (this.nextChunkToDownload < this.chunks.size())
        {
          // Reuse the set of object to avoid reallocation
          // It is important to do this BEFORE starting the next download
          chunkDataCache.add(this.chunks.get(prevChunk));
        }
        else
        {
          // clear the cache if we don't need it anymore
          chunkDataCache.clear();
        }
  
        // Free any memory the previous chunk might hang on
        this.chunks.get(prevChunk).freeData();
  
        releaseCurrentMemoryUsage(prevChunk, chunkMemUsage);
  
      }
  
      // if no more chunks, return null
      if (this.nextChunkToConsume >= this.chunks.size())
      {
        logger.debug("no more chunk");
        return null;
      }
  
      // prefetch next chunks
      try
      {
        startNextDownloaders();
      }
      catch (OutOfMemoryError outOfMemoryError)
      {
        logOutOfMemoryError();
        StringWriter errors = new StringWriter();
        outOfMemoryError.printStackTrace(new PrintWriter(errors));
        throw new SnowflakeSQLException(SqlState.INTERNAL_ERROR,
                                        ErrorCode.INTERNAL_ERROR.getMessageCode(),
                                        errors);
      }
  
      SnowflakeResultChunk currentChunk = this.chunks.get(nextChunkToConsume);
  
      if (currentChunk.getDownloadState() == DownloadState.SUCCESS)
      {
        logger.debug("chunk #{} is ready to consume", nextChunkToConsume);
        nextChunkToConsume++;
        if (nextChunkToConsume == this.chunks.size())
        {
          // make sure to release the last chunk
          releaseCurrentMemoryUsage(
              nextChunkToConsume - 1, chunks.get(nextChunkToConsume - 1).computeNeededChunkMemory());
        }
        return currentChunk;
      }
      else
      {
        // the chunk we want to consume is not ready yet, wait for it
        try
        {
          logger.debug("chunk #{} is not ready to consume",
                       nextChunkToConsume);
  
          currentChunk.getLock().lock();
          logger.debug("consumer get lock to check chunk state");
  
          while (currentChunk.getDownloadState() != DownloadState.SUCCESS &&
                 currentChunk.getDownloadState() != DownloadState.FAILURE)
          {
            logger.debug("wait for chunk #{} to be ready, current"
                         + "chunk state is: {}",
                         nextChunkToConsume, currentChunk.getDownloadState());
  
            long startTime = System.currentTimeMillis();
            if (!currentChunk.getDownloadCondition().await(downloadedConditionTimeoutInSeconds, TimeUnit.SECONDS))
            {
              currentChunk.setDownloadState(DownloadState.FAILURE);
              currentChunk.setDownloadError(String.format("Timeout waiting for the download of chunk #%d" +
                                                          "(Total chunks: %d)", nextChunkToConsume, this.chunks.size()));
            }
            this.numberMillisWaitingForChunks +=
                (System.currentTimeMillis() - startTime);
  
            logger.debug(
                "woken up from waiting for chunk #{} to be ready",
                nextChunkToConsume);
          }
  
          // downloader thread encountered an error
          if (currentChunk.getDownloadState() == DownloadState.FAILURE)
          {
            logger.error("downloader encountered error: {}",
                         currentChunk.getDownloadError());
  
            if (currentChunk.getDownloadError().contains("java.lang.OutOfMemoryError: Java heap space"))
            {
              logOutOfMemoryError();
            }
  
            throw new SnowflakeSQLException(SqlState.INTERNAL_ERROR,
                                            ErrorCode.INTERNAL_ERROR.getMessageCode(),
                                            currentChunk.getDownloadError());
          }
  
          logger.debug("chunk #{} is ready to consume",
                       nextChunkToConsume);
  
          nextChunkToConsume++;
  
          // next chunk to consume is ready for consumption
          return currentChunk;
        }
        finally
        {
          logger.debug("consumer free lock");
  
          boolean terminateDownloader = (currentChunk.getDownloadState() == DownloadState.FAILURE);
          // release the unlock always
          currentChunk.getLock().unlock();
          if (nextChunkToConsume == this.chunks.size())
          {
            // make sure to release the last chunk
            releaseCurrentMemoryUsage(
                nextChunkToConsume - 1, chunks.get(nextChunkToConsume - 1).computeNeededChunkMemory());
          }
          if (terminateDownloader)
          {
            logger.debug("Download result fail. Shut down the chunk downloader");
            terminate();
          }
        }
      }
    }