docs: public static Callable<Void> getUploadFileCallable(
        final StageInfo stage,
        final String srcFilePath,
        final FileMetadata metadata,
        final SnowflakeStorageClient client,
        final SFSession connection,
        final String command,
        final InputStream inputStream,
        final boolean sourceFromStream,
        final int parallel,
        final File srcFile,
        final RemoteStoreFileEncryptionMaterial encMat)
    {
      return new Callable<Void>()
      {
        public Void call() throws Exception
        {
  
          logger.debug("Entering getUploadFileCallable...");
  
          InputStream uploadStream = inputStream;
  
          File fileToUpload = null;
  
          if (uploadStream == null)
          {
            try
            {
              uploadStream = new FileInputStream(srcFilePath);
            }
            catch (FileNotFoundException ex)
            {
              metadata.resultStatus = ResultStatus.ERROR;
              metadata.errorDetails = ex.getMessage();
              throw ex;
            }
          }
  
          // this shouldn't happen
          if (metadata == null)
          {
            throw new SnowflakeSQLException(SqlState.INTERNAL_ERROR,
                                            ErrorCode.INTERNAL_ERROR.getMessageCode(),
                                            "missing file metadata for: " + srcFilePath);
          }
  
          String destFileName = metadata.destFileName;
  
          long uploadSize;
  
          String digest = null;
  
          logger.debug("Dest file name={}");
  
          // Temp file that needs to be cleaned up when upload was successful
          FileBackedOutputStream fileBackedOutputStream = null;
  
          // SNOW-16082: we should catpure exception if we fail to compress or
          // calcuate digest.
          try
          {
            if (metadata.requireCompress)
            {
              InputStreamWithMetadata compressedSizeAndStream = (encMat == null ?
                                                                 compressStreamWithGZIPNoDigest(uploadStream) :
                                                                 compressStreamWithGZIP(uploadStream));
  
              fileBackedOutputStream =
                  compressedSizeAndStream.fileBackedOutputStream;
  
              // update the size
              uploadSize = compressedSizeAndStream.size;
              digest = compressedSizeAndStream.digest;
  
              if (compressedSizeAndStream.fileBackedOutputStream.getFile() != null)
              {
                fileToUpload =
                    compressedSizeAndStream.fileBackedOutputStream.getFile();
              }
  
              logger.debug("New size after compression: {}", uploadSize);
            }
            else if (stage.getStageType() != StageInfo.StageType.LOCAL_FS)
            {
              // If it's not local_fs, we store our digest in the metadata
              // In local_fs, we don't need digest, and if we turn it on, we will consume whole uploadStream, which local_fs uses.
              InputStreamWithMetadata result = computeDigest(uploadStream,
                                                             sourceFromStream);
              digest = result.digest;
              fileBackedOutputStream = result.fileBackedOutputStream;
              uploadSize = result.size;
  
              if (!sourceFromStream)
              {
                fileToUpload = srcFile;
              }
              else if (result.fileBackedOutputStream.getFile() != null)
              {
                fileToUpload = result.fileBackedOutputStream.getFile();
              }
            }
            else
            {
              if (!sourceFromStream && (srcFile != null))
              {
                fileToUpload = srcFile;
              }
  
              // if stage is local_fs and upload source is stream, upload size
              // does not matter since 1) transfer did not require size 2) no
              // output from uploadStream api is required
              uploadSize = sourceFromStream ? 0 : srcFile.length();
            }
  
            logger.debug("Started copying file from: {} to {}:{} destName: {} " +
                         "auto compressed? {} size={}", srcFilePath, stage.getStageType().name(), stage.getLocation(),
                         destFileName, (metadata.requireCompress ? "yes" : "no"),
                         uploadSize);
  
            // Simulated failure code.
            if (connection.getInjectFileUploadFailure()
                != null && srcFilePath.endsWith(
                (connection).getInjectFileUploadFailure()))
            {
              throw new SnowflakeSimulatedUploadFailure(
                  srcFile != null ? srcFile.getName() : "Unknown");
            }
  
            // upload it
            switch (stage.getStageType())
            {
              case LOCAL_FS:
                pushFileToLocal(stage.getLocation(),
                                srcFilePath, destFileName, uploadStream,
                                fileBackedOutputStream);
                break;
  
              case S3:
              case AZURE:
                pushFileToRemoteStore(stage,
                                      destFileName,
                                      uploadStream, fileBackedOutputStream, uploadSize,
                                      digest, metadata.destCompressionType,
                                      client, connection, command, parallel, fileToUpload,
                                      (fileToUpload == null), encMat);
                metadata.isEncrypted = encMat != null;
                break;
            }
          }
          catch (SnowflakeSimulatedUploadFailure ex)
          {
            // This code path is used for Simulated failure code in tests.
            // Never happen in production
            metadata.resultStatus = ResultStatus.ERROR;
            metadata.errorDetails = ex.getMessage();
            throw ex;
          }
          catch (Throwable ex)
          {
            logger.error(
                "Exception encountered during file upload", ex);
            metadata.resultStatus = ResultStatus.ERROR;
            metadata.errorDetails = ex.getMessage();
            throw ex;
          }
          finally
          {
            if (fileBackedOutputStream != null)
            {
              try
              {
                fileBackedOutputStream.reset();
              }
              catch (IOException ex)
              {
                logger.debug("failed to clean up temp file: {}", ex);
              }
            }
            if (inputStream == null)
            {
              IOUtils.closeQuietly(uploadStream);
            }
          }
  
  
          logger.debug("filePath: {}", srcFilePath);
  
          // set dest size
          metadata.destFileSize = uploadSize;
  
          // mark the file as being uploaded
          metadata.resultStatus = ResultStatus.UPLOADED;
  
          return null;
        }
      };
    }