docs: private static Callable<Void> getDownloadChunkCallable(
        final SnowflakeChunkDownloader downloader,
        final SnowflakeResultChunk resultChunk,
        final String qrmk, final int chunkIndex,
        final Map<String, String> chunkHeadersMap,
        final int networkTimeoutInMilli, boolean useJsonParserV2)
    {
      return new Callable<Void>()
      {
        public Void call() throws Exception
        {
          try
          {
            // set the chunk state to be in progress
            try
            {
              resultChunk.getLock().lock();
              resultChunk.setDownloadState(DownloadState.IN_PROGRESS);
            }
            finally
            {
              resultChunk.getLock().unlock();
            }
  
            logger.debug("Downloading chunk {}, url={}",
                         chunkIndex, resultChunk.getUrl());
  
            long startTime = System.currentTimeMillis();
  
            HttpResponse response = getResultChunk(resultChunk.getUrl());
  
            /*
             * return error if we don't get a response or the response code
             * means failure.
             */
            if (response == null
                || response.getStatusLine().getStatusCode() != 200)
            {
              logger.error("Error fetching chunk from: {}",
                           resultChunk.getUrl());
  
              SnowflakeUtil.logResponseDetails(response, logger);
  
              throw new SnowflakeSQLException(SqlState.IO_ERROR,
                                              ErrorCode.NETWORK_ERROR
                                                  .getMessageCode(),
                                              "Error encountered when downloading a result chunk: HTTP "
                                              + "status="
                                              + ((response != null)
                                                 ? response.getStatusLine().getStatusCode()
                                                 : "null response"));
            }
  
            InputStream jsonInputStream;
            final HttpEntity entity = response.getEntity();
            try
            {
              // read the chunk data
              InputStream is =
                  new HttpUtil.HttpInputStream(entity.getContent());
  
              // Determine the format of the response, if it is not
              // either plain text or gzip, raise an error.
              Header encoding = response.getFirstHeader("Content-Encoding");
              if (encoding != null)
              {
                if (encoding.getValue().equalsIgnoreCase("gzip"))
                {
                  /* specify buffer size for GZIPInputStream */
                  is = new GZIPInputStream(is, STREAM_BUFFER_SIZE);
                }
                else
                {
                  throw
                      new SnowflakeSQLException(
                          SqlState.INTERNAL_ERROR,
                          ErrorCode.INTERNAL_ERROR.getMessageCode(),
                          "Exception: unexpected compression got " +
                          encoding.getValue());
                }
              }
  
              if (useJsonParserV2)
              {
                jsonInputStream = is;
              }
              else
              {
                // Build a sequence of streams to wrap the input stream
                // with '[' ... ']' to be able to plug this in the
                // Jackson JSON parser.
                // gzip stream uses 64KB
                // no buffering as json parser does it internally
                jsonInputStream =
                    new SequenceInputStream(
                        Collections.enumeration(Arrays.asList(
                            new ByteArrayInputStream("[".getBytes(
                                StandardCharsets.UTF_8)),
                            is,
                            new ByteArrayInputStream("]".getBytes(
                                StandardCharsets.UTF_8)))));
              }
            }
            catch (Exception ex)
            {
              logger.error(
                  "Failed to uncompress data: {}",
                  response);
  
              throw ex;
            }
  
            // remember the download time
            resultChunk.setDownloadTime(System.currentTimeMillis() - startTime);
            downloader.addDownloadTime(resultChunk.getDownloadTime());
  
            startTime = System.currentTimeMillis();
  
            // trace the response if requested
            logger.debug("Json response: {}", response);
  
            JsonNode resultData = null;
  
            // parse the result json
            try
            {
              if (downloader.useJsonParserV2)
              {
                parseJsonToChunkV2(jsonInputStream, resultChunk);
              }
              else
              {
                parseJsonToChunk(jsonInputStream, resultChunk);
              }
            }
            catch (Exception ex)
            {
              logger.error("Exception when parsing result", ex);
  
              throw new SnowflakeSQLException(ex, SqlState.INTERNAL_ERROR,
                                              ErrorCode.INTERNAL_ERROR
                                                  .getMessageCode(),
                                              "Exception: " +
                                              ex.getLocalizedMessage() +
                                              "\nBad result json: " + response.toString());
            }
            finally
            {
              // close the buffer reader will close underlying stream
              jsonInputStream.close();
            }
  
            // add parsing time
            resultChunk.setParseTime(System.currentTimeMillis() - startTime);
            downloader.addParsingTime(resultChunk.getParseTime());
  
            // remember the result data (it can be null if using rowsets)
            resultChunk.setResultData(resultData);
  
            logger.debug(
                "Finished preparing chunk data for {}, " +
                "total download time={}ms, total parse time={}ms",
                resultChunk.getUrl(),
                resultChunk.getDownloadTime(),
                resultChunk.getParseTime());
  
            try
            {
              resultChunk.getLock().lock();
              logger.debug(
                  "get lock to change the chunk to be ready to consume");
  
              logger.debug(
                  "wake up consumer if it is waiting for a chunk to be "
                  + "ready");
  
              resultChunk.setDownloadState(DownloadState.SUCCESS);
              resultChunk.getDownloadCondition().signal();
            }
            finally
            {
              logger.debug(
                  "Downloaded chunk {}, free lock", chunkIndex);
  
              resultChunk.getLock().unlock();
            }
          }
          catch (Throwable ex)
          {
            try
            {
              logger.debug("get lock to set chunk download error");
              resultChunk.getLock().lock();
  
              resultChunk.setDownloadState(DownloadState.FAILURE);
              StringWriter errors = new StringWriter();
              ex.printStackTrace(new PrintWriter(errors));
              resultChunk.setDownloadError(errors.toString());
  
              logger.debug(
                  "wake up consumer if it is waiting for a chunk to be ready");
  
              resultChunk.getDownloadCondition().signal();
            }
            finally
            {
              logger.debug("Failed to download chunk {}, free lock",
                           chunkIndex);
              resultChunk.getLock().unlock();
            }
  
            logger.error(
                "Exception encountered ({}:{}) fetching chunk from: {}",
                ex.getClass().getName(),
                ex.getLocalizedMessage(),
                resultChunk.getUrl());
  
            logger.error("Exception: ", ex);
          }
  
          return null;
        }
  
        private void parseJsonToChunkV2(InputStream jsonInputStream,
                                        SnowflakeResultChunk resultChunk)
        throws IOException, SnowflakeSQLException
        {
          /*
           * This is a hand-written binary parser that
           * handle.
           *   [ "c1", "c2", null, ... ],
           *   [ null, "c2", "c3", ... ],
           *   ...
           *   [ "c1", "c2", "c3", ... ],
           * in UTF-8
           * The number of rows is known and the number of expected columns
           * is also known.
           */
          ResultJsonParserV2 jp = new ResultJsonParserV2();
          jp.startParsing(resultChunk);
  
          byte[] buf = new byte[STREAM_BUFFER_SIZE];
          int len;
          while ((len = jsonInputStream.read(buf)) != -1)
          {
            jp.continueParsing(ByteBuffer.wrap(buf, 0, len));
          }
  
          jp.endParsing();
        }
  
        private void parseJsonToChunk(InputStream jsonInputStream,
                                      SnowflakeResultChunk resultChunk)
        throws IOException, SnowflakeSQLException
        {
          /*
           * This is a hand-written customized parser that
           * handle.
           * [
           *   [ "c1", "c2", null, ... ],
           *   [ null, "c2", "c3", ... ],
           *   ...
           *   [ "c1", "c2", "c3", ... ],
           * ]
           * The number of rows is known and the number of expected columns
           * is also known.
           */
          try (JsonParser jp = jsonFactory.createParser(new InputStreamReader(jsonInputStream, "UTF-8")))
          {
            JsonToken currentToken;
  
            // Get the first token and make sure it is the start of an array
            currentToken = jp.nextToken();
            if (currentToken != JsonToken.START_ARRAY)
            {
              throw
                  new SnowflakeSQLException(
                      SqlState.INTERNAL_ERROR,
                      ErrorCode.INTERNAL_ERROR.getMessageCode(),
                      "Exception1: expected '[' " +
                      "got " +
                      currentToken.asString());
            }
  
            // For all the rows...
            while (jp.nextToken() != JsonToken.END_ARRAY)
            {
              // Position to the current row in the result
              resultChunk.addRow(mapper.readValue(jp, Object[].class));
            }
            resultChunk.ensureRowsComplete();
          }
        }
  
        private HttpResponse getResultChunk(String chunkUrl) throws URISyntaxException, IOException, SnowflakeSQLException
        {
          URIBuilder uriBuilder = new URIBuilder(chunkUrl);
  
          HttpGet httpRequest = new HttpGet(uriBuilder.build());
  
          if (chunkHeadersMap != null && chunkHeadersMap.size() != 0)
          {
            for (Map.Entry<String, String> entry : chunkHeadersMap.entrySet())
            {
              logger.debug("Adding header key={}, value={}",
                           entry.getKey(), entry.getValue());
              httpRequest.addHeader(entry.getKey(), entry.getValue());
            }
          }
          // Add SSE-C headers
          else if (qrmk != null)
          {
            httpRequest.addHeader(SSE_C_ALGORITHM, SSE_C_AES);
            httpRequest.addHeader(SSE_C_KEY, qrmk);
            logger.debug("Adding SSE-C headers");
          }
  
          logger.debug("Fetching result: {}", resultChunk.getUrl());
  
          //TODO move this s3 request to HttpUtil class. In theory, upper layer
          //TODO does not need to know about http client
          CloseableHttpClient httpClient = HttpUtil.getHttpClient();
  
          // fetch the result chunk
          HttpResponse response =
              RestRequest.execute(httpClient,
                                  httpRequest,
                                  networkTimeoutInMilli / 1000, // retry timeout
                                  0, // no socketime injection
                                  null, // no canceling
                                  false, // no cookie
                                  false, // no retry
                                  false // no request_guid
              );
  
          logger.debug("Call returned for URL: {}",
                       chunkUrl);
          return response;
        }
      };
    }