### Question:
BackupManager { public CompletableFuture<?> submit(UUID uuid, File source) { requireNonNull(uuid, "uuid is null"); requireNonNull(source, "source is null"); if (backupStore.isEmpty()) { return completedFuture(null); } pendingBackups.incrementAndGet(); CompletableFuture<?> future = runAsync(new BackgroundBackup(uuid, source), executorService); future.whenComplete((none, throwable) -> pendingBackups.decrementAndGet()); return future; } @Inject  BackupManager(Optional<BackupStore> backupStore, StorageService storageService, BackupConfig config);  BackupManager(Optional<BackupStore> backupStore, StorageService storageService, int backupThreads); @PreDestroy void shutdown(); CompletableFuture<?> submit(UUID uuid, File source); @Managed int getPendingBackupCount(); @Managed @Flatten BackupStats getStats();  }### Answer:
@Test public void testSimple() throws Exception { assertEmptyStagingDirectory(); assertBackupStats(0, 0, 0); List<CompletableFuture<?>> futures = new ArrayList<>(); List<UUID> uuids = new ArrayList<>(5); for (int i = 0; i < 5; i++) { File file = new File(temporary, "file" + i); Files.write("hello world", file, UTF_8); uuids.add(randomUUID()); futures.add(backupManager.submit(uuids.get(i), file)); } futures.forEach(CompletableFuture::join); for (UUID uuid : uuids) { assertTrue(backupStore.shardExists(uuid)); } assertBackupStats(5, 0, 0); assertEmptyStagingDirectory(); }
@Test public void testFailure() throws Exception { assertEmptyStagingDirectory(); assertBackupStats(0, 0, 0); File file = new File(temporary, "failure"); Files.write("hello world", file, UTF_8); try { backupManager.submit(FAILURE_UUID, file).get(10, SECONDS); fail("expected exception"); } catch (ExecutionException wrapper) { PrestoException e = (PrestoException) wrapper.getCause(); assertEquals(e.getErrorCode(), RAPTOR_BACKUP_ERROR.toErrorCode()); assertEquals(e.getMessage(), "Backup failed for testing"); } assertBackupStats(0, 1, 0); assertEmptyStagingDirectory(); }
@Test public void testCorruption() throws Exception { assertEmptyStagingDirectory(); assertBackupStats(0, 0, 0); File file = new File(temporary, "corrupt"); Files.write("hello world", file, UTF_8); try { backupManager.submit(CORRUPTION_UUID, file).get(10, SECONDS); fail("expected exception"); } catch (ExecutionException wrapper) { PrestoException e = (PrestoException) wrapper.getCause(); assertEquals(e.getErrorCode(), RAPTOR_BACKUP_CORRUPTION.toErrorCode()); assertEquals(e.getMessage(), "Backup is corrupt after write: " + CORRUPTION_UUID); } File quarantineBase = storageService.getQuarantineFile(CORRUPTION_UUID); assertFile(new File(quarantineBase.getPath() + ".original")); assertFile(new File(quarantineBase.getPath() + ".restored")); assertBackupStats(0, 1, 1); assertEmptyStagingDirectory(); }