class ShardCleaner {

    @VisibleForTesting
    void cleanBackupShards()
    {
        Set<UUID> processing = newConcurrentHashSet();
        BlockingQueue<UUID> completed = new LinkedBlockingQueue<>();
        boolean fill = true;

        while (!Thread.currentThread().isInterrupted()) {
            // get a new batch if any completed and we are under the batch size
            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;
            }

            // skip any that are already processing and mark remaining as processing
            uuids = ImmutableSet.copyOf(difference(uuids, processing));
            processing.addAll(uuids);

            // execute deletes
            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);
                            }
                        });
            }

            // get the next batch of completed deletes
            int desired = min(100, processing.size());
            Collection<UUID> done = drain(completed, desired, 100, MILLISECONDS);
            if (done.isEmpty()) {
                continue;
            }

            // remove completed deletes from database
            processing.removeAll(done);
            dao.deleteCleanedShards(done);
            backupShardsCleaned.update(done.size());
            fill = true;
        }
    }

    @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);

}