### Question:
ShardCleaner { @VisibleForTesting void cleanBackupShards() { Set<UUID> processing = newConcurrentHashSet(); BlockingQueue<UUID> completed = new LinkedBlockingQueue<>(); boolean fill = true; while (!Thread.currentThread().isInterrupted()) { Set<UUID> uuids = ImmutableSet.of(); if (fill && (processing.size() < CLEANABLE_SHARDS_BATCH_SIZE)) { uuids = dao.getCleanableShardsBatch(maxTimestamp(backupCleanTime)); fill = false; } if (uuids.isEmpty() && processing.isEmpty()) { break; } uuids = ImmutableSet.copyOf(difference(uuids, processing)); processing.addAll(uuids); for (UUID uuid : uuids) { runAsync(() -> backupStore.get().deleteShard(uuid), backupExecutor) .thenAccept(v -> completed.add(uuid)) .whenComplete((v, e) -> { if (e != null) { log.error(e, "Error cleaning backup shard: %s", uuid); backupJobErrors.update(1); processing.remove(uuid); } }); } int desired = min(100, processing.size()); Collection<UUID> done = drain(completed, desired, 100, MILLISECONDS); if (done.isEmpty()) { continue; } processing.removeAll(done); dao.deleteCleanedShards(done); backupShardsCleaned.update(done.size()); fill = true; } } @Inject  ShardCleaner(
            DaoSupplier<ShardDao> shardDaoSupplier,
            Ticker ticker,
            NodeManager nodeManager,
            StorageService storageService,
            Optional<BackupStore> backupStore,
            ShardCleanerConfig config);  ShardCleaner(
            DaoSupplier<ShardDao> shardDaoSupplier,
            String currentNode,
            boolean coordinator,
            Ticker ticker,
            StorageService storageService,
            Optional<BackupStore> backupStore,
            Duration maxTransactionAge,
            Duration transactionCleanerInterval,
            Duration localCleanerInterval,
            Duration localCleanTime,
            Duration backupCleanerInterval,
            Duration backupCleanTime,
            int backupDeletionThreads,
            Duration maxCompletedTransactionAge); @PostConstruct void start(); @PreDestroy void shutdown(); @Managed @Nested CounterStat getTransactionJobErrors(); @Managed @Nested CounterStat getBackupJobErrors(); @Managed @Nested CounterStat getLocalJobErrors(); @Managed @Nested CounterStat getBackupShardsQueued(); @Managed @Nested CounterStat getBackupShardsCleaned(); @Managed @Nested CounterStat getLocalShardsCleaned(); @Managed void startBackupCleanup(); @Managed void startLocalCleanup(); @Managed void startLocalCleanupImmediately();  }### Answer:
@Test public void testCleanBackupShards() throws Exception { assertEquals(cleaner.getBackupShardsCleaned().getTotalCount(), 0); TestingDao dao = dbi.onDemand(TestingDao.class); UUID shard1 = randomUUID(); UUID shard2 = randomUUID(); UUID shard3 = randomUUID(); long now = System.currentTimeMillis(); Timestamp time1 = new Timestamp(now - HOURS.toMillis(25)); Timestamp time2 = new Timestamp(now - HOURS.toMillis(23)); dao.insertDeletedShard(shard1, time1); dao.insertDeletedShard(shard2, time1); dao.insertDeletedShard(shard3, time2); createShardBackups(shard1, shard2, shard3); cleaner.cleanBackupShards(); assertEquals(cleaner.getBackupShardsCleaned().getTotalCount(), 2); assertFalse(shardBackupExists(shard1)); assertFalse(shardBackupExists(shard2)); assertTrue(shardBackupExists(shard3)); assertQuery("SELECT shard_uuid FROM deleted_shards", row(shard3)); }