class UploaderController extends DataBridgeController {

    @SuppressFBWarnings(value = "BC_UNCONFIRMED_CAST_OF_RETURN_VALUE",
        justification = "manifestReader.readJsonManifest will always return an UploaderInputManifestDto object.")
    public void performUpload(RegServerAccessParamsDto regServerAccessParamsDto, File manifestPath, S3FileTransferRequestParamsDto params,
        Boolean createNewVersion, Boolean force, Integer maxRetryAttempts, Integer retryDelaySecs)
        throws InterruptedException, JAXBException, IOException, URISyntaxException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException
    {
        boolean cleanUpS3KeyPrefixOnFailure = false;
        BusinessObjectDataKey businessObjectDataKey = null;

        try
        {
            // Process manifest file
            UploaderInputManifestDto manifest = manifestReader.readJsonManifest(manifestPath);

            String storageName = getStorageNameFromManifest(manifest);
            manifest.setStorageName(storageName);

            // Validate local files and prepare a list of source files to copy to S3.
            List<File> sourceFiles = getValidatedLocalFiles(params.getLocalPath(), manifest.getManifestFiles());

            // Validate that we do not have duplicate files listed in the manifest file.
            List<File> duplicateFiles = findDuplicateFiles(sourceFiles);

            if (!duplicateFiles.isEmpty())
            {
                throw new IllegalArgumentException(
                    String.format("Manifest contains duplicate file names. Duplicates: [\"%s\"]", StringUtils.join(duplicateFiles, "\", \"")));
            }

            // Initialize uploader web client.
            uploaderWebClient.setRegServerAccessParamsDto(regServerAccessParamsDto);

            // Handle the latest business object data version if one exists.
            checkLatestBusinessObjectDataVersion(manifest, force);

            // Pre-register a new version of business object data in UPLOADING state with the registration server.
            BusinessObjectData businessObjectData = uploaderWebClient.preRegisterBusinessObjectData(manifest, storageName, createNewVersion);

            // Get business object data key.
            businessObjectDataKey = businessObjectDataHelper.getBusinessObjectDataKey(businessObjectData);

            // Get the business object data version.
            Integer businessObjectDataVersion = businessObjectDataKey.getBusinessObjectDataVersion();

            // Add credential provider.
            params.getAdditionalAwsCredentialsProviders().add(new AutoRefreshCredentialProvider()
            {
                @Override
                public AwsCredential getNewAwsCredential() throws Exception
                {
                    return uploaderWebClient.getBusinessObjectDataUploadCredential(manifest, storageName, businessObjectDataVersion, null).getAwsCredential();
                }
            });

            // Get S3 key prefix from the business object data pre-registration response.
            String s3KeyPrefix = IterableUtils.get(businessObjectData.getStorageUnits(), 0).getStorageDirectory().getDirectoryPath();

            // Get S3 bucket information.
            Storage storage = uploaderWebClient.getStorage(storageName);

            // Get S3 bucket name.  Please note that since this value is required we pass a "true" flag.
            String s3BucketName =
                storageHelper.getStorageAttributeValueByName(configurationHelper.getProperty(ConfigurationValue.S3_ATTRIBUTE_NAME_BUCKET_NAME), storage, true);

            // Set the KMS ID, if available
            String kmsKeyId =
                storageHelper.getStorageAttributeValueByName(configurationHelper.getProperty(ConfigurationValue.S3_ATTRIBUTE_NAME_KMS_KEY_ID), storage, false);
            params.setKmsKeyId(kmsKeyId);

            // Special handling for the maxThreads command line option.
            params.setMaxThreads(adjustIntegerValue(params.getMaxThreads(), MIN_THREADS, MAX_THREADS));

            // Populate several missing fields in the S3 file transfer request parameters DTO.
            params.setS3BucketName(s3BucketName);
            // Since the S3 key prefix represents a directory, we add a trailing '/' character to it.
            params.setS3KeyPrefix(s3KeyPrefix + "/");
            params.setFiles(sourceFiles);

            // Check if the destination S3 key prefix is empty.
            // When listing S3 files, by default, we do not ignore 0 byte objects that represent S3 directories.
            if (s3Service.listDirectory(params).isEmpty())
            {
                cleanUpS3KeyPrefixOnFailure = true;
            }
            else
            {
                throw new IllegalStateException(String
                    .format("The destination S3 folder is not empty. S3 Bucket Name: \"%s\". S3 key prefix: \"%s\".", params.getS3BucketName(),
                        params.getS3KeyPrefix()));
            }

            // Upload files.
            s3Service.uploadFileList(params);

            // Get the list of files uploaded to S3 key prefix.
            if (LOGGER.isInfoEnabled())
            {
                logS3KeyPrefixContents(params);
            }

            // Add storage files to the business object data.
            addStorageFilesWithRetry(businessObjectDataKey, manifest, params, storage.getName(), maxRetryAttempts, retryDelaySecs);

            // Change status of the business object data to VALID.
            uploaderWebClient.updateBusinessObjectDataStatus(businessObjectDataKey, BusinessObjectDataStatusEntity.VALID);
        }
        catch (InterruptedException | JAXBException | IOException | URISyntaxException e)
        {
            // If we got to the point of checking the target S3 key prefix before this failure
            // occurred, let's rollback the data transfer (clean up the S3 key prefix).
            if (cleanUpS3KeyPrefixOnFailure)
            {
                LOGGER.info(String
                    .format("Rolling back the S3 data transfer by deleting keys/objects with prefix \"%s\" from bucket \"%s\".", params.getS3KeyPrefix(),
                        params.getS3BucketName()));
                s3Service.deleteDirectoryIgnoreException(params);
            }

            // If a new business object data version got pre-registered, update it's status to INVALID.
            if (businessObjectDataKey != null)
            {
                uploaderWebClient.updateBusinessObjectDataStatusIgnoreException(businessObjectDataKey, BusinessObjectDataStatusEntity.INVALID);
            }

            throw e;
        }
    }

}