### Question:
ShardCleaner { @VisibleForTesting synchronized void cleanLocalShardsImmediately(Set<UUID> local) { Set<UUID> assigned = dao.getNodeShards(currentNode, null).stream() .map(ShardMetadata::getShardUuid) .collect(toSet()); Set<UUID> deletions = Sets.difference(local, assigned); for (UUID uuid : deletions) { deleteFile(storageService.getStorageFile(uuid)); } localShardsCleaned.update(deletions.size()); log.info("Cleaned %s local shards immediately", deletions.size()); } @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 testCleanLocalShardsImmediately() throws Exception { assertEquals(cleaner.getLocalShardsCleaned().getTotalCount(), 0); TestingShardDao shardDao = dbi.onDemand(TestingShardDao.class); MetadataDao metadataDao = dbi.onDemand(MetadataDao.class); long tableId = metadataDao.insertTable("test", "test", false, false, null, 0); UUID shard1 = randomUUID(); UUID shard2 = randomUUID(); UUID shard3 = randomUUID(); Set<UUID> shards = ImmutableSet.of(shard1, shard2, shard3); for (UUID shard : shards) { shardDao.insertShard(shard, tableId, null, 0, 0, 0, 0); createShardFile(shard); assertTrue(shardFileExists(shard)); } int node1 = shardDao.insertNode("node1"); int node2 = shardDao.insertNode("node2"); shardDao.insertShardNode(shard1, node1); shardDao.insertShardNode(shard3, node2); Set<UUID> local = cleaner.getLocalShards(); cleaner.cleanLocalShardsImmediately(local); assertEquals(cleaner.getLocalShardsCleaned().getTotalCount(), 2); assertTrue(shardFileExists(shard1)); assertFalse(shardFileExists(shard2)); assertFalse(shardFileExists(shard3)); }