docs: @Override
    public void upload(
        SFSession connection,
        String command,
        int parallelism,
        boolean uploadFromStream,
        String remoteStorageLocation,
        File srcFile,
        String destFileName,
        InputStream inputStream,
        FileBackedOutputStream fileBackedOutputStream,
        StorageObjectMetadata meta,
        String stageRegion) throws SnowflakeSQLException
    {
      final long originalContentLength = meta.getContentLength();
      final List<FileInputStream> toClose = new ArrayList<>();
      SFPair<InputStream, Boolean> uploadStreamInfo =
          createUploadStream(srcFile, uploadFromStream,
                             inputStream, fileBackedOutputStream,
                             ((S3ObjectMetadata) meta).getS3ObjectMetadata(),
                             originalContentLength, toClose);
  
      ObjectMetadata s3Meta;
      if (meta instanceof S3ObjectMetadata)
      {
        s3Meta = ((S3ObjectMetadata) meta).getS3ObjectMetadata();
      }
      else
      {
        throw new IllegalArgumentException("Unexpected metadata object type");
      }
  
      TransferManager tx = null;
      int retryCount = 0;
      do
      {
        try
        {
  
          logger.debug("Creating executor service for transfer" +
                       "manager with {} threads", parallelism);
  
          // upload files to s3
          tx = new TransferManager(this.amazonClient,
                                   SnowflakeUtil.createDefaultExecutorService(
                                       "s3-transfer-manager-uploader-",
                                       parallelism));
  
          final Upload myUpload;
  
          if (uploadStreamInfo.right)
          {
            myUpload = tx.upload(remoteStorageLocation, destFileName,
                                 uploadStreamInfo.left, s3Meta);
          }
          else
          {
            PutObjectRequest putRequest =
                new PutObjectRequest(remoteStorageLocation, destFileName, srcFile);
            putRequest.setMetadata(s3Meta);
  
            myUpload = tx.upload(putRequest);
          }
  
          myUpload.waitForCompletion();
  
          // get out
          for (FileInputStream is : toClose)
            IOUtils.closeQuietly(is);
          return;
        }
        catch (Exception ex)
        {
  
          handleS3Exception(ex, ++retryCount, "upload",
                            connection, command, this);
          if (uploadFromStream && fileBackedOutputStream == null)
          {
            throw new SnowflakeSQLException(ex, SqlState.SYSTEM_ERROR,
                                            ErrorCode.IO_ERROR.getMessageCode(),
                                            "Encountered exception during upload: " +
                                            ex.getMessage() + "\nCannot retry upload from stream.");
          }
          uploadStreamInfo = createUploadStream(srcFile, uploadFromStream,
                                                inputStream, fileBackedOutputStream,
                                                s3Meta, originalContentLength, toClose);
        }
        finally
        {
          if (tx != null)
          {
            tx.shutdownNow(false);
          }
        }
      }
      while (retryCount <= getMaxRetries());
  
      for (FileInputStream is : toClose)
        IOUtils.closeQuietly(is);
  
      throw new SnowflakeSQLException(SqlState.INTERNAL_ERROR,
                                      ErrorCode.INTERNAL_ERROR.getMessageCode(),
                                      "Unexpected: upload unsuccessful without exception!");
    }