class ShardEjector {

    @VisibleForTesting
    void process()
    {
        checkState(backupStore.isPresent(), "backup store must be present");

        // get the size of assigned shards for each node
        Map<String, Long> nodes = shardManager.getNodeBytes();

        Set<String> activeNodes = nodeSupplier.getWorkerNodes().stream()
                .map(Node::getNodeIdentifier)
                .collect(toSet());

        // only include active nodes
        nodes = new HashMap<>(filterKeys(nodes, activeNodes::contains));

        if (nodes.isEmpty()) {
            return;
        }

        // get current node size
        if (!nodes.containsKey(currentNode)) {
            return;
        }
        long nodeSize = nodes.get(currentNode);

        // get average node size
        long averageSize = round(nodes.values().stream()
                .mapToLong(Long::longValue)
                .average()
                .getAsDouble());
        long maxSize = round(averageSize * 1.01);

        // skip if not above max
        if (nodeSize <= maxSize) {
            return;
        }

        // only include nodes that are below threshold
        nodes = new HashMap<>(filterValues(nodes, size -> size <= averageSize));

        // get non-bucketed node shards by size, largest to smallest
        List<ShardMetadata> shards = shardManager.getNodeShards(currentNode).stream()
                .filter(shard -> shard.getBucketNumber().isEmpty())
                .sorted(comparingLong(ShardMetadata::getCompressedSize).reversed())
                .collect(toList());

        // eject shards while current node is above max
        Queue<ShardMetadata> queue = new ArrayDeque<>(shards);
        while ((nodeSize > maxSize) && !queue.isEmpty()) {
            ShardMetadata shard = queue.remove();
            long shardSize = shard.getCompressedSize();
            UUID shardUuid = shard.getShardUuid();

            // verify backup exists
            if (!backupStore.get().shardExists(shardUuid)) {
                log.warn("No backup for shard: %s", shardUuid);
            }

            // pick target node
            String target = pickTargetNode(nodes, shardSize, averageSize);
            if (target == null) {
                return;
            }
            long targetSize = nodes.get(target);

            // stats
            log.info("Moving shard %s to node %s (shard: %s, node: %s, average: %s, target: %s)",
                    shardUuid, target, shardSize, nodeSize, averageSize, targetSize);
            shardsEjected.update(1);

            // update size
            nodes.put(target, targetSize + shardSize);
            nodeSize -= shardSize;

            // move assignment
            shardManager.replaceShardAssignment(shard.getTableId(), shardUuid, target, false);

            // delete local file
            File file = storageService.getStorageFile(shardUuid);
            if (file.exists() && !file.delete()) {
                log.warn("Failed to delete shard file: %s", file);
            }
        }
    }

    @Inject public  ShardEjector(
            NodeManager nodeManager,
            NodeSupplier nodeSupplier,
            ShardManager shardManager,
            StorageService storageService,
            StorageManagerConfig config,
            Optional<BackupStore> backupStore,
            RaptorConnectorId connectorId);
    public  ShardEjector(
            String currentNode,
            NodeSupplier nodeSupplier,
            ShardManager shardManager,
            StorageService storageService,
            Duration interval,
            Optional<BackupStore> backupStore,
            String connectorId);

}