class ShardCleaner {

    @VisibleForTesting
    synchronized void cleanLocalShardsImmediately(Set<UUID> local)
    {
        // get shards assigned to the local node
        Set<UUID> assigned = dao.getNodeShards(currentNode, null).stream()
                .map(ShardMetadata::getShardUuid)
                .collect(toSet());

        // only delete local files that are not assigned
        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 public  ShardCleaner(
            DaoSupplier<ShardDao> shardDaoSupplier,
            Ticker ticker,
            NodeManager nodeManager,
            StorageService storageService,
            Optional<BackupStore> backupStore,
            ShardCleanerConfig config);
    public  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);

}