@VisibleForTesting
    static void internalMoveTempFileToStore(
            File incomingFile,
            @Nullable JobID jobId,
            BlobKey blobKey,
            File storageFile,
            Logger log,
            @Nullable BlobStore blobStore,
            MoveFileOperation moveFileOperation)
            throws IOException {

        boolean success = false;
        try {
            // first check whether the file already exists
            if (!storageFile.exists()) {
                // persist the blob via the blob store
                if (blobStore != null) {
                    blobStore.put(incomingFile, jobId, blobKey);
                }

                try {
                    moveFileOperation.moveFile(incomingFile, storageFile);
                    incomingFile = null;

                } catch (FileAlreadyExistsException ignored) {
                    log.warn(
                            "Detected concurrent file modifications. This should only happen if multiple"
                                    + "BlobServer use the same storage directory.");
                }
            } else {
                log.warn(
                        "File upload for an existing file with key {} for job {}. This may indicate a duplicate upload or a hash collision. Ignoring newest upload.",
                        blobKey,
                        jobId);
            }
            success = true;
        } finally {
            if (!success) {
                if (blobStore != null) {
                    blobStore.delete(jobId, blobKey);
                }

                if (!storageFile.delete() && storageFile.exists()) {
                    log.warn("Could not delete the storage file {}.", storageFile);
                }
            }

            if (incomingFile != null && !incomingFile.delete() && incomingFile.exists()) {
                log.warn(
                        "Could not delete the staging file {} for blob key {} and job {}.",
                        incomingFile,
                        blobKey,
                        jobId);
            }
        }
    }
@Test
    void testCleanupIfMoveTempFileToStoreFails() throws IOException {
        final File storageFile = tempDir.resolve(UUID.randomUUID().toString()).toFile();

        final File incomingFile = TempDirUtils.newFile(tempDir);
        Files.write(incomingFile.toPath(), new byte[] {1, 2, 3, 4});

        final FileSystemBlobStore blobStore =
                new FileSystemBlobStore(
                        new LocalFileSystem(), TempDirUtils.newFolder(tempDir).toString());

        final JobID jobId = new JobID();
        final BlobKey blobKey = BlobKey.createKey(BlobKey.BlobType.PERMANENT_BLOB);
        assertThatThrownBy(
                        () ->
                                BlobUtils.internalMoveTempFileToStore(
                                        incomingFile,
                                        jobId,
                                        blobKey,
                                        storageFile,
                                        LOG,
                                        blobStore,
                                        (source, target) -> {
                                            throw new IOException("Test Failure");
                                        }))
                .isInstanceOf(IOException.class);

        assertThatThrownBy(
                        () ->
                                blobStore.get(
                                        jobId,
                                        blobKey,
                                        tempDir.resolve(UUID.randomUUID().toString()).toFile()))
                .isInstanceOf(FileNotFoundException.class);
        assertThat(incomingFile).doesNotExist();
        assertThat(storageFile).doesNotExist();
    }